laminar_core/cluster/control/lease_deadline/
mod.rs

1use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
2use std::time::{Duration, Instant};
3
4/// Process-local monotonic deadline for a renewable durable lease.
5pub struct LeaseDeadline {
6    origin: Instant,
7    valid_until_ns: AtomicU64,
8    terminal: AtomicBool,
9    transition: parking_lot::Mutex<()>,
10    changed: tokio::sync::Notify,
11}
12
13impl LeaseDeadline {
14    /// Create an inactive, renewable deadline for a lease manager before acquisition.
15    #[must_use]
16    pub(crate) fn uninitialized() -> Self {
17        Self {
18            origin: Instant::now(),
19            valid_until_ns: AtomicU64::new(0),
20            terminal: AtomicBool::new(false),
21            transition: parking_lot::Mutex::new(()),
22            changed: tokio::sync::Notify::new(),
23        }
24    }
25
26    /// Create an irreversibly fenced deadline.
27    #[must_use]
28    pub fn fenced() -> Self {
29        let deadline = Self::uninitialized();
30        deadline.fence();
31        deadline
32    }
33
34    /// Create a deadline live for `remaining`.
35    #[must_use]
36    pub fn live_for(remaining: Duration) -> Self {
37        let deadline = Self::uninitialized();
38        deadline.extend(remaining);
39        deadline
40    }
41
42    pub(crate) fn extend(&self, remaining: Duration) {
43        let until = self.origin.elapsed().saturating_add(remaining).as_nanos();
44        let until = u64::try_from(until).unwrap_or(u64::MAX).max(1);
45        self.update_valid_until(until);
46    }
47
48    pub(crate) fn extend_until(&self, valid_until: Instant) {
49        let Some(remaining_from_origin) = valid_until.checked_duration_since(self.origin) else {
50            self.fence();
51            return;
52        };
53        if remaining_from_origin.is_zero() {
54            self.fence();
55            return;
56        }
57        let until = u64::try_from(remaining_from_origin.as_nanos()).unwrap_or(u64::MAX);
58        self.update_valid_until(until);
59    }
60
61    fn update_valid_until(&self, valid_until_ns: u64) {
62        let _transition = self.transition.lock();
63        if self.terminal.load(Ordering::Acquire) {
64            return;
65        }
66        let current = self.valid_until_ns.load(Ordering::Acquire);
67        if current != 0 && self.origin.elapsed().as_nanos() >= u128::from(current) {
68            self.terminal.store(true, Ordering::Release);
69            self.valid_until_ns.store(0, Ordering::Release);
70            self.changed.notify_waiters();
71            return;
72        }
73        self.valid_until_ns.store(valid_until_ns, Ordering::Release);
74        self.changed.notify_waiters();
75    }
76
77    /// Irreversibly revoke the lease deadline and wake all waiters.
78    pub fn fence(&self) {
79        let _transition = self.transition.lock();
80        self.terminal.store(true, Ordering::Release);
81        self.valid_until_ns.store(0, Ordering::Release);
82        self.changed.notify_waiters();
83    }
84
85    /// Whether the holder remains inside its last successful renewal deadline.
86    #[must_use]
87    pub fn is_live(&self) -> bool {
88        if self.terminal.load(Ordering::Acquire) {
89            return false;
90        }
91        let deadline = self.valid_until_ns.load(Ordering::Acquire);
92        deadline != 0 && self.origin.elapsed().as_nanos() < u128::from(deadline)
93    }
94
95    /// Wait until the deadline expires naturally or is terminally fenced.
96    pub async fn wait_until_expired(&self) {
97        loop {
98            let changed = self.changed.notified();
99            tokio::pin!(changed);
100            changed.as_mut().enable();
101
102            if self.terminal.load(Ordering::Acquire) {
103                return;
104            }
105            let valid_until_ns = self.valid_until_ns.load(Ordering::Acquire);
106            let elapsed_ns = self.origin.elapsed().as_nanos();
107            if valid_until_ns == 0 {
108                return;
109            }
110            if elapsed_ns >= u128::from(valid_until_ns) {
111                let _transition = self.transition.lock();
112                if self.terminal.load(Ordering::Acquire) {
113                    return;
114                }
115                let current = self.valid_until_ns.load(Ordering::Acquire);
116                if current != 0 && self.origin.elapsed().as_nanos() < u128::from(current) {
117                    continue;
118                }
119                self.terminal.store(true, Ordering::Release);
120                self.valid_until_ns.store(0, Ordering::Release);
121                self.changed.notify_waiters();
122                return;
123            }
124            let remaining_ns = u128::from(valid_until_ns).saturating_sub(elapsed_ns);
125            let remaining = Duration::from_nanos(u64::try_from(remaining_ns).unwrap_or(u64::MAX));
126            tokio::select! {
127                () = tokio::time::sleep(remaining) => {}
128                () = &mut changed => {}
129            }
130        }
131    }
132}
133
134impl std::fmt::Debug for LeaseDeadline {
135    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
136        formatter
137            .debug_struct("LeaseDeadline")
138            .field("terminal", &self.terminal.load(Ordering::Acquire))
139            .field("live", &self.is_live())
140            .finish_non_exhaustive()
141    }
142}
143
144#[cfg(test)]
145mod tests;
laminar_core/cluster/control/lease_deadline/mod.rs

laminar_core/cluster/control/lease_deadline/
mod.rs
