subetha_cxc/
failover.rs

1//! `FailoverWatchdog` - scans the heartbeat table and reclaims
2//! in-flight work whose owning process has stopped beating.
3//!
4//! Architectural contract: failover happens within ONE epoch of the
5//! peer's last beat. The watchdog advances the global epoch on each
6//! scan; any process whose `last_seen_epoch < global - grace_epochs`
7//! is presumed dead and its `in_flight_bitmap` is returned to the
8//! caller as a `ReclaimReport` so the caller (typically the
9//! scheduler) can reassign the work.
10
11use crate::heartbeat::{HeartbeatSnapshot, HeartbeatTable, IN_FLIGHT_SLOTS};
12
13/// Default grace window. A slot must miss more than this many epochs
14/// before it is reclaimed.
15pub const DEFAULT_GRACE_EPOCHS: u64 = 1;
16
17/// Report from one watchdog scan.
18#[derive(Debug, Clone)]
19pub struct ReclaimReport {
20    /// Slot index -> last snapshot of a dead process whose
21    /// in-flight bits should be reclaimed.
22    pub dead_slots: Vec<(usize, HeartbeatSnapshot)>,
23    /// New global epoch after the scan.
24    pub new_global_epoch: u64,
25}
26
27impl ReclaimReport {
28    pub fn is_empty(&self) -> bool { self.dead_slots.is_empty() }
29}
30
31/// Watchdog scanner; one per cooperating cluster.
32pub struct FailoverWatchdog<'a> {
33    pub table: &'a HeartbeatTable,
34    pub grace_epochs: u64,
35}
36
37impl<'a> FailoverWatchdog<'a> {
38    pub fn new(table: &'a HeartbeatTable) -> Self {
39        Self { table, grace_epochs: DEFAULT_GRACE_EPOCHS }
40    }
41
42    pub fn with_grace(table: &'a HeartbeatTable, grace_epochs: u64) -> Self {
43        Self { table, grace_epochs }
44    }
45
46    /// Advance global epoch and scan every slot. Returns a report
47    /// of slots whose last beat is more than `grace_epochs` behind
48    /// the new global epoch.
49    ///
50    /// Per-scan observation is pushed to the underlying
51    /// HeartbeatTable's sidecar ring rather than a separate
52    /// watchdog-owned ring: the watchdog borrows the table with
53    /// lifetime `'a`, which is incompatible with the
54    /// `AdaptiveInstance: 'static` trait bound. Routing the scan
55    /// observation through the table preserves visibility to any
56    /// policy attached at that level.
57    pub fn scan(&self) -> ReclaimReport {
58        let new_epoch = self.table.tick_global_epoch();
59        let mut dead = Vec::new();
60        for i in 0..self.table.capacity() {
61            if let Some(snap) = self.table.snapshot(i) {
62                let lag = new_epoch.saturating_sub(snap.last_seen_epoch);
63                if lag > self.grace_epochs && snap.in_flight_bitmap != 0 {
64                    dead.push((i, snap));
65                }
66            }
67        }
68        let dead_count = dead.len();
69        <HeartbeatTable as subetha_sidecar::AdaptiveInstance>::ring(self.table).push(
70            subetha_core::Observation {
71                op_kind: crate::sidecar_ops::liveness::OP_SCAN,
72                flags: if dead_count > 0 { 1 } else { 0 },  // 1 = reclaim required
73                ..subetha_core::Observation::ZERO
74            },
75        );
76        ReclaimReport { dead_slots: dead, new_global_epoch: new_epoch }
77    }
78
79    /// Iterate the set bits in `bitmap`, returning each bit's
80    /// position. Used by callers reclaiming an in_flight_bitmap.
81    pub fn iter_in_flight_bits(bitmap: u64) -> impl Iterator<Item = u8> {
82        (0u8..IN_FLIGHT_SLOTS as u8).filter(move |b| (bitmap >> b) & 1 == 1)
83    }
84
85    /// Clear the dead process's bitmap so subsequent scans don't
86    /// re-report it. Typically called by the caller after they have
87    /// reassigned the work.
88    pub fn clear_dead_bitmap(&self, slot_idx: usize) {
89        let slot = self.table_slot(slot_idx);
90        slot.in_flight_bitmap.store(0, std::sync::atomic::Ordering::Release);
91    }
92
93    fn table_slot(&self, idx: usize) -> &crate::heartbeat::HeartbeatSlot {
94        // Re-derive via the public snapshot path is awkward; reach
95        // into the table directly. (HeartbeatTable's private slot
96        // accessor is accessed via this crate-private helper.)
97        crate::heartbeat::__slot_for_watchdog(self.table, idx)
98    }
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104    use crate::heartbeat::HeartbeatTable;
105
106    fn tmp_path(name: &str) -> std::path::PathBuf {
107        let mut p = std::env::temp_dir();
108        let pid = std::process::id();
109        p.push(format!("subetha-failover-{name}-{pid}.bin"));
110        p
111    }
112
113    #[test]
114    fn watchdog_reports_no_dead_when_all_beat() {
115        let p = tmp_path("all-alive");
116        let t = HeartbeatTable::create(&p, 4).unwrap();
117        let s0 = t.register(1).unwrap();
118        let s1 = t.register(2).unwrap();
119        t.mark_in_flight(s0, 0);
120        t.mark_in_flight(s1, 1);
121
122        let w = FailoverWatchdog::new(&t);
123        // Beat both and then scan - should not be dead.
124        t.beat(s0); t.beat(s1);
125        let r = w.scan();
126        assert!(r.is_empty(),
127                "no dead processes expected; got {} dead", r.dead_slots.len());
128        std::fs::remove_file(&p).ok();
129    }
130
131    #[test]
132    fn watchdog_reports_dead_when_grace_exceeded() {
133        let p = tmp_path("dead-one");
134        let t = HeartbeatTable::create(&p, 4).unwrap();
135        let s_alive = t.register(1).unwrap();
136        let s_dead = t.register(2).unwrap();
137        t.mark_in_flight(s_alive, 0);
138        t.mark_in_flight(s_dead, 1);
139        t.beat(s_alive); t.beat(s_dead);
140
141        let w = FailoverWatchdog::with_grace(&t, 2);
142        // Scan: global=1, both slots lag=1, grace=2 -> not dead.
143        let r1 = w.scan();
144        assert!(r1.is_empty(), "first scan within grace; got {:?}", r1.dead_slots);
145        // Only alive beats.
146        t.beat(s_alive);
147        // Scan: global=2, alive lag=1, dead lag=2 == grace -> not dead.
148        let r2 = w.scan();
149        assert!(r2.is_empty(), "second scan equal to grace; got {:?}", r2.dead_slots);
150        // Only alive beats again.
151        t.beat(s_alive);
152        // Scan: global=3, alive lag=1, dead lag=3 > grace=2 -> dead.
153        let r3 = w.scan();
154        assert_eq!(r3.dead_slots.len(), 1);
155        let (idx, snap) = &r3.dead_slots[0];
156        assert_eq!(*idx, s_dead);
157        assert_eq!(snap.pid, 2);
158        assert_eq!(snap.in_flight_bitmap, 1u64 << 1);
159        std::fs::remove_file(&p).ok();
160    }
161
162    #[test]
163    fn iter_in_flight_bits_walks_set_positions() {
164        let bm = (1u64 << 0) | (1u64 << 3) | (1u64 << 5) | (1u64 << 63);
165        let bits: Vec<u8> = FailoverWatchdog::iter_in_flight_bits(bm).collect();
166        assert_eq!(bits, vec![0, 3, 5, 63]);
167    }
168
169    #[test]
170    fn clear_dead_bitmap_silences_reports() {
171        let p = tmp_path("clear-dead");
172        let t = HeartbeatTable::create(&p, 1).unwrap();
173        let s = t.register(7).unwrap();
174        t.mark_in_flight(s, 4);
175        t.beat(s);
176        let w = FailoverWatchdog::with_grace(&t, 0);
177        // grace=0 + tick=1 -> lag 1 > 0 -> reported as dead.
178        let r1 = w.scan();
179        assert_eq!(r1.dead_slots.len(), 1);
180        // Clear and rescan.
181        w.clear_dead_bitmap(s);
182        let r2 = w.scan();
183        assert!(r2.is_empty(), "after clear, no dead slots reported");
184        std::fs::remove_file(&p).ok();
185    }
186}
subetha_cxc/failover.rs

subetha_cxc/
failover.rs
