laminar_core/cluster/control/process_lease/
mod.rs

1//! Durable ownership of one stable cluster node identity.
2//!
3//! Each renewal appends a create-only sequence object. This gives local filesystems and object
4//! stores the same compare-and-set boundary without relying on backend-specific entity tags.
5
6mod authority;
7mod manager;
8mod store;
9
10use object_store::path::Path as OsPath;
11use serde::{Deserialize, Serialize};
12use uuid::Uuid;
13
14use crate::cluster::discovery::NodeId;
15
16pub use authority::ProcessLeaseAuthority;
17pub use manager::{ProcessLeaseConfig, ProcessLeaseManager};
18pub use store::ProcessLeaseStore;
19
20use std::time::Duration;
21
22const MAX_PROCESS_LEASE_RECORD_BYTES: u64 = 1024;
23const MAX_PROCESS_LEASE_FENCE_BYTES: u64 = 2048;
24const PROCESS_LEASE_HEAD_READ_ATTEMPTS: usize = 4;
25const PROCESS_LEASE_HISTORY_TO_RETAIN: usize = 2;
26const PROCESS_LEASE_MAX_LIST_RECORDS: usize = 4096;
27const PROCESS_LEASE_PRUNE_BATCH_RECORDS: usize = 256;
28const PROCESS_LEASE_PRUNE_READ_CONCURRENCY: usize = 32;
29const PROCESS_LEASE_WRITES_PER_PRUNE: u64 = 64;
30const PROCESS_LEASE_MAX_PRUNE_BATCHES: usize = 4;
31const PROCESS_LEASE_PRUNE_IO_TIMEOUT: Duration = Duration::from_secs(5);
32const PROCESS_LEASE_RENEW_RETRY_DELAY: Duration = Duration::from_millis(250);
33
34fn lease_prefix(node: NodeId) -> String {
35    format!("control/process-lease/node={}/", node.0)
36}
37
38fn lease_path(node: NodeId, seq: u64) -> OsPath {
39    OsPath::from(format!("{}v{seq:016}.json", lease_prefix(node)))
40}
41
42fn fence_path(node: NodeId, predecessor: Uuid) -> OsPath {
43    OsPath::from(format!(
44        "control/process-lease-fences/v1/node={}/predecessor={predecessor}.json",
45        node.0
46    ))
47}
48
49fn successor_fence_path(node: NodeId, successor: Uuid, term: u64) -> OsPath {
50    OsPath::from(format!(
51        "control/process-lease-fences/v1/node={}/successor={successor}/term={term:016}.json",
52        node.0
53    ))
54}
55
56fn sequence_from_path(node: NodeId, path: &OsPath) -> Result<u64, ProcessLeaseError> {
57    let prefix = lease_prefix(node);
58    let raw = path
59        .as_ref()
60        .strip_prefix(&prefix)
61        .and_then(|file| file.strip_prefix('v'))
62        .and_then(|file| file.strip_suffix(".json"))
63        .ok_or_else(|| {
64            ProcessLeaseError::Invalid(format!("invalid process lease record path {path}"))
65        })?;
66    if raw.is_empty() || !raw.bytes().all(|byte| byte.is_ascii_digit()) {
67        return Err(ProcessLeaseError::Invalid(format!(
68            "invalid process lease sequence in {path}"
69        )));
70    }
71    let sequence = raw.parse::<u64>().map_err(|error| {
72        ProcessLeaseError::Invalid(format!("invalid process lease sequence in {path}: {error}"))
73    })?;
74    if sequence == 0 || lease_path(node, sequence) != *path {
75        return Err(ProcessLeaseError::Invalid(format!(
76            "noncanonical process lease record path {path}"
77        )));
78    }
79    Ok(sequence)
80}
81
82fn now_millis() -> i64 {
83    std::time::SystemTime::now()
84        .duration_since(std::time::UNIX_EPOCH)
85        .map_or(0, |duration| {
86            i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
87        })
88}
89
90/// Durable owner of one stable node identity.
91#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
92pub struct ProcessLease {
93    /// Stable node identity protected by this lease.
94    pub node: NodeId,
95    /// Boot-unique owner identity.
96    pub owner: Uuid,
97    /// Monotonic process term. It advances after every lease lapse.
98    pub term: u64,
99    /// Append-only compare-and-set sequence.
100    pub seq: u64,
101    /// Owner-written advisory expiry for diagnostics; takeover never compares client clocks.
102    pub expires_at_ms: i64,
103}
104
105impl ProcessLease {
106    pub(crate) fn validate(&self, expected_node: NodeId) -> Result<(), ProcessLeaseError> {
107        if self.node != expected_node || self.node.is_unassigned() {
108            return Err(ProcessLeaseError::Invalid(
109                "lease node does not match its durable namespace".into(),
110            ));
111        }
112        if self.owner.is_nil() || self.term == 0 || self.seq == 0 {
113            return Err(ProcessLeaseError::Invalid(
114                "lease owner, term, and sequence must be nonzero".into(),
115            ));
116        }
117        Ok(())
118    }
119}
120
121/// Durable proof that one exact stable-node process term was superseded after a full lease
122/// observation. The successor is the immediate create-only takeover record, not a wall-clock
123/// expiry estimate.
124#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
125#[serde(deny_unknown_fields)]
126pub struct ProcessLeaseFence {
127    /// Last lease record owned by the process being fenced.
128    pub predecessor: ProcessLease,
129    /// Immediate different-owner term that revoked the predecessor.
130    pub successor: ProcessLease,
131}
132
133impl ProcessLeaseFence {
134    /// Build an exact process-term transition.
135    ///
136    /// # Errors
137    /// Rejects different node namespaces, same-owner renewals, or a non-adjacent sequence/term.
138    pub fn new(
139        predecessor: ProcessLease,
140        successor: ProcessLease,
141    ) -> Result<Self, ProcessLeaseError> {
142        let fence = Self {
143            predecessor,
144            successor,
145        };
146        if !fence.is_canonical() {
147            return Err(ProcessLeaseError::Invalid(
148                "process lease fence must bind an immediate different-owner takeover".into(),
149            ));
150        }
151        Ok(fence)
152    }
153
154    /// Whether this is one exact create-only owner transition.
155    #[must_use]
156    pub fn is_canonical(&self) -> bool {
157        self.predecessor.validate(self.predecessor.node).is_ok()
158            && self.successor.validate(self.predecessor.node).is_ok()
159            && self.predecessor.owner != self.successor.owner
160            && self.predecessor.seq.checked_add(1) == Some(self.successor.seq)
161            && self.predecessor.term.checked_add(1) == Some(self.successor.term)
162    }
163}
164
165/// Result of one process-lease acquisition attempt.
166#[derive(Debug, Clone, PartialEq, Eq)]
167pub enum ProcessLeaseOutcome {
168    /// This boot incarnation owns the stable node identity.
169    Acquired(ProcessLease),
170    /// A different live boot incarnation owns it.
171    Held(ProcessLease),
172}
173
174/// Candidate-local proof that one exact durable lease record stayed current for a full TTL.
175#[derive(Debug)]
176pub struct ProcessLeaseObservation {
177    lease: ProcessLease,
178    started: std::time::Instant,
179}
180
181/// Process lease storage failure.
182#[derive(Debug, thiserror::Error)]
183pub enum ProcessLeaseError {
184    /// Underlying object-store failure.
185    #[error("object store I/O: {0}")]
186    Io(String),
187    /// Invalid durable record.
188    #[error("invalid process lease: {0}")]
189    Invalid(String),
190    /// JSON encoding or decoding failure.
191    #[error("JSON: {0}")]
192    Json(#[from] serde_json::Error),
193    /// Caller-provided monotonic deadline expired before fencing could be proven.
194    #[error("process lease fencing deadline: {0}")]
195    Deadline(String),
196}
197
198#[cfg(test)]
199mod tests;
laminar_core/cluster/control/process_lease/mod.rs

laminar_core/cluster/control/process_lease/
mod.rs
