laminar_core/cluster/control/process_lease/
authority.rs

1//! Deadline-bounded fencing and current-participant verification.
2
3use std::future::Future;
4use std::sync::Arc;
5use std::time::Duration;
6
7use object_store::ObjectStore;
8use uuid::Uuid;
9
10use crate::cluster::discovery::NodeId;
11
12use super::{
13    now_millis, ProcessLease, ProcessLeaseError, ProcessLeaseFence, ProcessLeaseOutcome,
14    ProcessLeaseStore,
15};
16
17/// Shared authority for proving that an exact process incarnation has been durably revoked.
18pub struct ProcessLeaseAuthority {
19    store: Arc<dyn ObjectStore>,
20    ttl: Duration,
21    ttl_ms: i64,
22}
23
24impl std::fmt::Debug for ProcessLeaseAuthority {
25    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        formatter
27            .debug_struct("ProcessLeaseAuthority")
28            .field("ttl", &self.ttl)
29            .finish_non_exhaustive()
30    }
31}
32
33impl ProcessLeaseAuthority {
34    /// Bind all stable-node lease namespaces on one shared object store and TTL.
35    ///
36    /// # Errors
37    /// Rejects a zero, sub-millisecond, fractional-millisecond, or oversized TTL.
38    pub fn new(store: Arc<dyn ObjectStore>, ttl: Duration) -> Result<Self, ProcessLeaseError> {
39        let ttl_ms = i64::try_from(ttl.as_millis())
40            .ok()
41            .filter(|value| *value > 0)
42            .ok_or_else(|| ProcessLeaseError::Invalid("process lease TTL is invalid".into()))?;
43        if Duration::from_millis(u64::try_from(ttl_ms).unwrap_or(u64::MAX)) != ttl {
44            return Err(ProcessLeaseError::Invalid(
45                "process lease TTL must use whole milliseconds".into(),
46            ));
47        }
48        Ok(Self { store, ttl, ttl_ms })
49    }
50
51    /// Open one stable-node namespace over the shared authority.
52    #[must_use]
53    pub fn store_for(&self, node: NodeId) -> Arc<ProcessLeaseStore> {
54        Arc::new(ProcessLeaseStore::new(
55            Arc::clone(&self.store),
56            node,
57            self.ttl_ms,
58        ))
59    }
60
61    /// Build a monotonic deadline that covers the mandatory full-TTL observation plus bounded
62    /// authority I/O. Callers need not duplicate the process-lease TTL as another runtime knob.
63    ///
64    /// # Errors
65    /// Rejects a zero I/O budget or monotonic-clock overflow.
66    pub fn fencing_deadline(
67        &self,
68        io_budget: Duration,
69    ) -> Result<tokio::time::Instant, ProcessLeaseError> {
70        if io_budget.is_zero() {
71            return Err(ProcessLeaseError::Invalid(
72                "process lease fencing I/O budget must be nonzero".into(),
73            ));
74        }
75        tokio::time::Instant::now()
76            .checked_add(self.ttl)
77            .and_then(|deadline| deadline.checked_add(io_budget))
78            .ok_or_else(|| {
79                ProcessLeaseError::Invalid("process lease fencing deadline overflow".into())
80            })
81    }
82
83    async fn bounded<T>(
84        deadline: tokio::time::Instant,
85        operation: &str,
86        future: impl Future<Output = Result<T, ProcessLeaseError>>,
87    ) -> Result<T, ProcessLeaseError> {
88        if tokio::time::Instant::now() >= deadline {
89            return Err(ProcessLeaseError::Deadline(format!(
90                "deadline expired before {operation}"
91            )));
92        }
93        tokio::time::timeout_at(deadline, future)
94            .await
95            .map_err(|_| {
96                ProcessLeaseError::Deadline(format!("deadline expired during {operation}"))
97            })?
98    }
99
100    async fn recover_won_fence(
101        &self,
102        participant: crate::checkpoint::CheckpointParticipant,
103        head: ProcessLease,
104        deadline: tokio::time::Instant,
105    ) -> Result<ProcessLeaseFence, ProcessLeaseError> {
106        let node = NodeId(participant.node_id);
107        if head.node != node || head.owner == participant.boot_incarnation {
108            return Err(ProcessLeaseError::Invalid(
109                "process lease head has not superseded the requested incarnation".into(),
110            ));
111        }
112        let store = self.store_for(node);
113        let fence = Self::bounded(
114            deadline,
115            "locating process lease takeover evidence",
116            store.find_takeover_from(participant.boot_incarnation),
117        )
118        .await?
119        .ok_or_else(|| {
120            ProcessLeaseError::Invalid(format!(
121                "process lease takeover evidence for {} is missing",
122                participant.boot_incarnation
123            ))
124        })?;
125        if !self.verify_fence(&fence, deadline).await? {
126            return Err(ProcessLeaseError::Invalid(
127                "process lease takeover is no longer durably verifiable".into(),
128            ));
129        }
130        Ok(fence)
131    }
132
133    /// Durably supersede an unchanged process incarnation after observing it for one full TTL.
134    ///
135    /// A retry after the create-only takeover won reconstructs the exact fence from the two
136    /// retained history records. It never waits a second TTL for that already-durable result.
137    ///
138    /// # Errors
139    /// Fails closed on renewal, deadline expiry, missing history, or any authority I/O failure.
140    pub async fn fence_incarnation(
141        &self,
142        participant: crate::checkpoint::CheckpointParticipant,
143        deadline: tokio::time::Instant,
144    ) -> Result<ProcessLeaseFence, ProcessLeaseError> {
145        let node = NodeId(participant.node_id);
146        if node.is_unassigned() || participant.boot_incarnation.is_nil() {
147            return Err(ProcessLeaseError::Invalid(
148                "process lease fence participant is not canonical".into(),
149            ));
150        }
151        let store = self.store_for(node);
152        let head = Self::bounded(deadline, "loading process lease fence head", store.load())
153            .await?
154            .ok_or_else(|| {
155                ProcessLeaseError::Invalid("process lease fence history is missing".into())
156            })?;
157        if head.owner != participant.boot_incarnation {
158            return self.recover_won_fence(participant, head, deadline).await;
159        }
160
161        let observation = store.observe_rival(&head)?;
162        let observation_until = tokio::time::Instant::now()
163            .checked_add(self.ttl)
164            .ok_or_else(|| ProcessLeaseError::Invalid("process lease TTL overflows time".into()))?;
165        if observation_until >= deadline {
166            return Err(ProcessLeaseError::Deadline(
167                "deadline does not cover one full process lease TTL".into(),
168            ));
169        }
170        tokio::time::sleep_until(observation_until).await;
171
172        let mut revoker = Uuid::new_v4();
173        while revoker.is_nil() || revoker == participant.boot_incarnation {
174            revoker = Uuid::new_v4();
175        }
176        let outcome = Self::bounded(
177            deadline,
178            "publishing process lease takeover",
179            store.try_takeover(revoker, &observation, now_millis()),
180        )
181        .await?;
182        match outcome {
183            ProcessLeaseOutcome::Acquired(successor) => {
184                let fence = ProcessLeaseFence::new(head, successor)?;
185                if !self.verify_fence(&fence, deadline).await? {
186                    return Err(ProcessLeaseError::Invalid(
187                        "won process lease takeover could not be verified".into(),
188                    ));
189                }
190                Ok(fence)
191            }
192            ProcessLeaseOutcome::Held(current) if current.owner == participant.boot_incarnation => {
193                Err(ProcessLeaseError::Invalid(
194                    "process incarnation renewed during the full-TTL observation".into(),
195                ))
196            }
197            ProcessLeaseOutcome::Held(current) => {
198                self.recover_won_fence(participant, current, deadline).await
199            }
200        }
201    }
202
203    /// Verify that an exact boot incarnation is the current durable owner of its stable node.
204    ///
205    /// # Errors
206    /// Fails closed on deadline expiry, missing takeover evidence, malformed state, or I/O.
207    pub async fn verify_current_participant(
208        &self,
209        participant: crate::checkpoint::CheckpointParticipant,
210        deadline: tokio::time::Instant,
211    ) -> Result<bool, ProcessLeaseError> {
212        self.verify_current_participant_identity(participant, None, deadline)
213            .await
214    }
215
216    /// Verify an exact boot and process term against the current durable stable-node authority.
217    ///
218    /// # Errors
219    /// Fails closed on deadline expiry, missing takeover evidence, malformed state, or I/O.
220    pub async fn verify_current_participant_term(
221        &self,
222        participant: crate::checkpoint::CheckpointParticipant,
223        process_term: u64,
224        deadline: tokio::time::Instant,
225    ) -> Result<bool, ProcessLeaseError> {
226        if process_term == 0 {
227            return Err(ProcessLeaseError::Invalid(
228                "process lease term must be nonzero".into(),
229            ));
230        }
231        self.verify_current_participant_identity(participant, Some(process_term), deadline)
232            .await
233    }
234
235    async fn verify_current_participant_identity(
236        &self,
237        participant: crate::checkpoint::CheckpointParticipant,
238        process_term: Option<u64>,
239        deadline: tokio::time::Instant,
240    ) -> Result<bool, ProcessLeaseError> {
241        let node = NodeId(participant.node_id);
242        if node.is_unassigned() || participant.boot_incarnation.is_nil() {
243            return Err(ProcessLeaseError::Invalid(
244                "process lease participant is not canonical".into(),
245            ));
246        }
247        let store = self.store_for(node);
248        let Some(head) =
249            Self::bounded(deadline, "loading current process lease", store.load()).await?
250        else {
251            return Ok(false);
252        };
253        if head.owner != participant.boot_incarnation
254            || process_term.is_some_and(|term| head.term != term)
255        {
256            return Ok(false);
257        }
258        Self::bounded(
259            deadline,
260            "verifying current process term evidence",
261            store.ensure_current_term_fence(&head),
262        )
263        .await?;
264        let Some(after) =
265            Self::bounded(deadline, "rechecking current process lease", store.load()).await?
266        else {
267            return Ok(false);
268        };
269        Ok(after.owner == head.owner
270            && after.term == head.term
271            && after.seq >= head.seq
272            && process_term.is_none_or(|term| after.term == term))
273    }
274
275    /// Verify that both exact fence records remain present and the fenced owner is not current.
276    ///
277    /// # Errors
278    /// Fails closed on deadline expiry, missing history, malformed records, or authority I/O.
279    pub async fn verify_fence(
280        &self,
281        fence: &ProcessLeaseFence,
282        deadline: tokio::time::Instant,
283    ) -> Result<bool, ProcessLeaseError> {
284        if !fence.is_canonical() {
285            return Err(ProcessLeaseError::Invalid(
286                "process lease fence is not canonical".into(),
287            ));
288        }
289        let store = self.store_for(fence.predecessor.node);
290        let durable_fence = Self::bounded(
291            deadline,
292            "verifying indexed process lease fence",
293            ProcessLeaseStore::load_fence(
294                &store.store,
295                fence.predecessor.node,
296                fence.predecessor.owner,
297            ),
298        )
299        .await?
300        .ok_or_else(|| {
301            ProcessLeaseError::Invalid("indexed process lease fence is missing".into())
302        })?;
303        let head = Self::bounded(deadline, "verifying process lease fence head", store.load())
304            .await?
305            .ok_or_else(|| {
306                ProcessLeaseError::Invalid("process lease fence durable head is missing".into())
307            })?;
308        Ok(durable_fence == *fence
309            && head.seq >= fence.successor.seq
310            && head.term >= fence.successor.term
311            && (head.term != fence.successor.term || head.owner == fence.successor.owner)
312            && head.owner != fence.predecessor.owner)
313    }
314}
laminar_core/cluster/control/process_lease/authority.rs

laminar_core/cluster/control/process_lease/
authority.rs
