laminar_core/cluster/control/process_lease/
manager.rs

1//! Renewal-loop ownership and terminal lease-loss publication.
2
3use std::sync::Arc;
4use std::time::{Duration, Instant};
5
6use tokio::sync::watch;
7use uuid::Uuid;
8
9use super::super::lease_deadline::LeaseDeadline;
10use super::{
11    now_millis, ProcessLease, ProcessLeaseError, ProcessLeaseOutcome, ProcessLeaseStore,
12    PROCESS_LEASE_RENEW_RETRY_DELAY,
13};
14
15/// Internal renewal timings for the stable-node lease.
16#[derive(Debug, Clone, Copy)]
17pub struct ProcessLeaseConfig {
18    /// Lease lifetime.
19    pub ttl: Duration,
20    /// Renewal cadence.
21    pub renew_interval: Duration,
22}
23
24impl Default for ProcessLeaseConfig {
25    fn default() -> Self {
26        Self {
27            // Same one-tail margin as the leader lease; incarnation revocation waits a full TTL.
28            ttl: Duration::from_secs(10),
29            renew_interval: Duration::from_secs(2),
30        }
31    }
32}
33
34/// Renews an already-acquired process lease and publishes terminal lease loss.
35pub struct ProcessLeaseManager {
36    store: Arc<ProcessLeaseStore>,
37    owner: Uuid,
38    config: ProcessLeaseConfig,
39    initial_valid_until: std::time::Instant,
40    live_tx: watch::Sender<bool>,
41    deadline: Arc<LeaseDeadline>,
42}
43
44impl ProcessLeaseManager {
45    /// Construct a renewal manager for an acquired lease.
46    ///
47    /// # Errors
48    /// Rejects a lease that does not match the store namespace or owner.
49    pub fn new(
50        store: Arc<ProcessLeaseStore>,
51        owner: Uuid,
52        config: ProcessLeaseConfig,
53        acquisition_started_at: Instant,
54        initial: &ProcessLease,
55    ) -> Result<Self, ProcessLeaseError> {
56        initial.validate(store.node)?;
57        let store_ttl = u64::try_from(store.ttl_ms)
58            .ok()
59            .filter(|ttl| *ttl > 0)
60            .map(Duration::from_millis)
61            .ok_or_else(|| {
62                ProcessLeaseError::Invalid("process lease TTL must be positive".into())
63            })?;
64        if initial.owner != owner
65            || config.ttl.is_zero()
66            || config.ttl != store_ttl
67            || config.renew_interval.is_zero()
68            || config.renew_interval >= config.ttl
69        {
70            return Err(ProcessLeaseError::Invalid(
71                "renewal manager requires this boot's lease, the exact store TTL, and a renewal interval below TTL".into(),
72            ));
73        }
74        let (live_tx, _live_rx) = watch::channel(true);
75        let ttl = config.ttl;
76        let now = Instant::now();
77        if acquisition_started_at > now {
78            return Err(ProcessLeaseError::Invalid(
79                "process lease acquisition start is in the future".into(),
80            ));
81        }
82        let initial_valid_until = acquisition_started_at
83            .checked_add(ttl)
84            .ok_or_else(|| ProcessLeaseError::Invalid("process lease TTL overflows time".into()))?;
85        if now >= initial_valid_until {
86            return Err(ProcessLeaseError::Invalid(
87                "process lease acquisition response arrived after its local deadline".into(),
88            ));
89        }
90        let deadline = Arc::new(LeaseDeadline::uninitialized());
91        deadline.extend_until(initial_valid_until);
92        if !deadline.is_live() {
93            return Err(ProcessLeaseError::Invalid(
94                "process lease acquisition response arrived after its local deadline".into(),
95            ));
96        }
97        Ok(Self {
98            store,
99            owner,
100            config,
101            initial_valid_until,
102            live_tx,
103            deadline,
104        })
105    }
106
107    /// Watch terminal ownership status.
108    #[must_use]
109    pub fn live_watch(&self) -> watch::Receiver<bool> {
110        self.live_tx.subscribe()
111    }
112
113    /// Shared monotonic deadline for hot-path fencing.
114    #[must_use]
115    pub fn deadline(&self) -> Arc<LeaseDeadline> {
116        Arc::clone(&self.deadline)
117    }
118
119    /// Spawn the renewal loop. Once ownership is uncertain past its expiry or a rival is observed,
120    /// the watch becomes false and this manager never reacquires.
121    #[cfg(feature = "cluster")]
122    #[must_use]
123    pub fn spawn(
124        self,
125        shutdown: tokio_util::sync::CancellationToken,
126    ) -> tokio::task::JoinHandle<()> {
127        tokio::spawn(async move {
128            let mut valid_until = self.initial_valid_until;
129            let mut ticker = tokio::time::interval(self.config.renew_interval);
130            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
131            // The acquisition itself is the first successful tick.
132            ticker.tick().await;
133            loop {
134                if Instant::now() >= valid_until {
135                    self.expire("renewal tick ran past the lease deadline");
136                    return;
137                }
138                let lease_deadline = tokio::time::Instant::from_std(valid_until);
139                tokio::select! {
140                    biased;
141                    () = shutdown.cancelled() => {
142                        self.fence();
143                        return;
144                    },
145                    () = tokio::time::sleep_until(lease_deadline) => {
146                        self.expire("no renewal completed before the lease deadline");
147                        return;
148                    }
149                    _ = ticker.tick() => {}
150                }
151
152                if Instant::now() >= valid_until {
153                    self.expire("renewal tick ran past the lease deadline");
154                    return;
155                }
156                let attempt_started_at = Instant::now();
157                let Some(attempt_valid_until) = attempt_started_at.checked_add(self.config.ttl)
158                else {
159                    self.expire("lease deadline overflows the monotonic clock");
160                    return;
161                };
162                // A stalled store request must not hold the only attempt until the lease lapses.
163                // An abandoned create that still lands is read back as this owner's head on retry.
164                let attempt_deadline =
165                    lease_deadline.min(tokio::time::Instant::now() + self.config.renew_interval);
166                let renewal = tokio::time::timeout_at(
167                    attempt_deadline,
168                    self.store.try_acquire(self.owner, now_millis()),
169                )
170                .await;
171                match renewal {
172                    Ok(Ok(ProcessLeaseOutcome::Acquired(_))) => {
173                        let response_at = Instant::now();
174                        if response_at >= valid_until || response_at >= attempt_valid_until {
175                            self.expire("renewal response arrived after the lease deadline");
176                            return;
177                        }
178                        valid_until = attempt_valid_until;
179                        self.deadline.extend_until(attempt_valid_until);
180                    }
181                    Ok(Ok(ProcessLeaseOutcome::Held(rival))) => {
182                        tracing::error!(
183                            node = self.store.node.0,
184                            owner = %rival.owner,
185                            term = rival.term,
186                            "stable node identity lease was lost"
187                        );
188                        self.fence();
189                        return;
190                    }
191                    Ok(Err(error)) => {
192                        if matches!(&error, ProcessLeaseError::Io(_)) {
193                            // RECOVERY: Preserve the original absolute fence, but do not discard
194                            // its remaining retry budget after a completed transient I/O failure.
195                            ticker.reset_after(
196                                PROCESS_LEASE_RENEW_RETRY_DELAY.min(self.config.renew_interval),
197                            );
198                        }
199                        tracing::warn!(%error, "stable node identity lease renewal failed");
200                    }
201                    Err(_) if attempt_deadline < lease_deadline => {
202                        ticker.reset_after(
203                            PROCESS_LEASE_RENEW_RETRY_DELAY.min(self.config.renew_interval),
204                        );
205                        tracing::warn!(
206                            node = self.store.node.0,
207                            timeout = ?self.config.renew_interval,
208                            "stable node identity lease renewal timed out; retrying"
209                        );
210                    }
211                    Err(_) => {
212                        self.expire("renewal was still pending at the lease deadline");
213                        return;
214                    }
215                }
216            }
217        })
218    }
219
220    #[cfg(feature = "cluster")]
221    fn expire(&self, reason: &'static str) {
222        tracing::error!(
223            node = self.store.node.0,
224            reason,
225            "stable node identity lease expired locally"
226        );
227        self.fence();
228    }
229
230    #[cfg(feature = "cluster")]
231    fn fence(&self) {
232        self.deadline.fence();
233        self.live_tx.send_replace(false);
234    }
235}
laminar_core/cluster/control/process_lease/manager.rs

laminar_core/cluster/control/process_lease/
manager.rs
