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            ttl: Duration::from_secs(15),
28            renew_interval: Duration::from_secs(5),
29        }
30    }
31}
32
33/// Renews an already-acquired process lease and publishes terminal lease loss.
34pub struct ProcessLeaseManager {
35    store: Arc<ProcessLeaseStore>,
36    owner: Uuid,
37    config: ProcessLeaseConfig,
38    initial_valid_until: std::time::Instant,
39    live_tx: watch::Sender<bool>,
40    deadline: Arc<LeaseDeadline>,
41}
42
43impl ProcessLeaseManager {
44    /// Construct a renewal manager for an acquired lease.
45    ///
46    /// # Errors
47    /// Rejects a lease that does not match the store namespace or owner.
48    pub fn new(
49        store: Arc<ProcessLeaseStore>,
50        owner: Uuid,
51        config: ProcessLeaseConfig,
52        acquisition_started_at: Instant,
53        initial: &ProcessLease,
54    ) -> Result<Self, ProcessLeaseError> {
55        initial.validate(store.node)?;
56        let store_ttl = u64::try_from(store.ttl_ms)
57            .ok()
58            .filter(|ttl| *ttl > 0)
59            .map(Duration::from_millis)
60            .ok_or_else(|| {
61                ProcessLeaseError::Invalid("process lease TTL must be positive".into())
62            })?;
63        if initial.owner != owner
64            || config.ttl.is_zero()
65            || config.ttl != store_ttl
66            || config.renew_interval.is_zero()
67            || config.renew_interval >= config.ttl
68        {
69            return Err(ProcessLeaseError::Invalid(
70                "renewal manager requires this boot's lease, the exact store TTL, and a renewal interval below TTL".into(),
71            ));
72        }
73        let (live_tx, _live_rx) = watch::channel(true);
74        let ttl = config.ttl;
75        let now = Instant::now();
76        if acquisition_started_at > now {
77            return Err(ProcessLeaseError::Invalid(
78                "process lease acquisition start is in the future".into(),
79            ));
80        }
81        let initial_valid_until = acquisition_started_at
82            .checked_add(ttl)
83            .ok_or_else(|| ProcessLeaseError::Invalid("process lease TTL overflows time".into()))?;
84        if now >= initial_valid_until {
85            return Err(ProcessLeaseError::Invalid(
86                "process lease acquisition response arrived after its local deadline".into(),
87            ));
88        }
89        let deadline = Arc::new(LeaseDeadline::uninitialized());
90        deadline.extend_until(initial_valid_until);
91        if !deadline.is_live() {
92            return Err(ProcessLeaseError::Invalid(
93                "process lease acquisition response arrived after its local deadline".into(),
94            ));
95        }
96        Ok(Self {
97            store,
98            owner,
99            config,
100            initial_valid_until,
101            live_tx,
102            deadline,
103        })
104    }
105
106    /// Watch terminal ownership status.
107    #[must_use]
108    pub fn live_watch(&self) -> watch::Receiver<bool> {
109        self.live_tx.subscribe()
110    }
111
112    /// Shared monotonic deadline for hot-path fencing.
113    #[must_use]
114    pub fn deadline(&self) -> Arc<LeaseDeadline> {
115        Arc::clone(&self.deadline)
116    }
117
118    /// Spawn the renewal loop. Once ownership is uncertain past its expiry or a rival is observed,
119    /// the watch becomes false and this manager never reacquires.
120    #[cfg(feature = "cluster")]
121    #[must_use]
122    pub fn spawn(
123        self,
124        shutdown: tokio_util::sync::CancellationToken,
125    ) -> tokio::task::JoinHandle<()> {
126        tokio::spawn(async move {
127            let mut valid_until = self.initial_valid_until;
128            let mut ticker = tokio::time::interval(self.config.renew_interval);
129            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
130            // The acquisition itself is the first successful tick.
131            ticker.tick().await;
132            loop {
133                let now = std::time::Instant::now();
134                if now >= valid_until {
135                    self.deadline.fence();
136                    self.live_tx.send_replace(false);
137                    return;
138                }
139                tokio::select! {
140                    biased;
141                    () = shutdown.cancelled() => {
142                        self.deadline.fence();
143                        self.live_tx.send_replace(false);
144                        return;
145                    },
146                    () = tokio::time::sleep_until(tokio::time::Instant::from_std(valid_until)) => {
147                        self.deadline.fence();
148                        self.live_tx.send_replace(false);
149                        return;
150                    }
151                    _ = ticker.tick() => {}
152                }
153
154                let now = std::time::Instant::now();
155                if now >= valid_until {
156                    self.deadline.fence();
157                    self.live_tx.send_replace(false);
158                    return;
159                }
160                let attempt_started_at = Instant::now();
161                let Some(attempt_valid_until) = attempt_started_at.checked_add(self.config.ttl)
162                else {
163                    self.deadline.fence();
164                    self.live_tx.send_replace(false);
165                    return;
166                };
167                let renewal = tokio::time::timeout_at(
168                    tokio::time::Instant::from_std(valid_until),
169                    self.store.try_acquire(self.owner, now_millis()),
170                )
171                .await;
172                match renewal {
173                    Ok(Ok(ProcessLeaseOutcome::Acquired(_))) => {
174                        let response_at = Instant::now();
175                        if response_at >= valid_until || response_at >= attempt_valid_until {
176                            self.deadline.fence();
177                            self.live_tx.send_replace(false);
178                            return;
179                        }
180                        valid_until = attempt_valid_until;
181                        self.deadline.extend_until(attempt_valid_until);
182                    }
183                    Ok(Ok(ProcessLeaseOutcome::Held(rival))) => {
184                        tracing::error!(
185                            node = self.store.node.0,
186                            owner = %rival.owner,
187                            term = rival.term,
188                            "stable node identity lease was lost"
189                        );
190                        self.deadline.fence();
191                        self.live_tx.send_replace(false);
192                        return;
193                    }
194                    Ok(Err(error)) => {
195                        if matches!(&error, ProcessLeaseError::Io(_)) {
196                            // RECOVERY: Preserve the original absolute fence, but do not discard
197                            // its remaining retry budget after a completed transient I/O failure.
198                            ticker.reset_after(
199                                PROCESS_LEASE_RENEW_RETRY_DELAY.min(self.config.renew_interval),
200                            );
201                        }
202                        tracing::warn!(%error, "stable node identity lease renewal failed");
203                    }
204                    Err(_) => {
205                        self.deadline.fence();
206                        self.live_tx.send_replace(false);
207                        return;
208                    }
209                }
210            }
211        })
212    }
213}
laminar_core/cluster/control/process_lease/manager.rs

laminar_core/cluster/control/process_lease/
manager.rs
