confium_coordinator/coordinator/
reaper.rs

1//! Session reaper — background thread that expires stale sessions.
2//!
3//! Sessions have an unlock window (default 4 hours). After it elapses,
4//! the session transitions to Expired, fires an audit event, and
5//! decrements the active session counter. Without reaping, stale
6//! sessions accumulate indefinitely.
7
8use crate::coordinator::audit::AuditEvent;
9use crate::coordinator::coordinator::Coordinator;
10use crate::coordinator::session::SessionState;
11use chrono::Utc;
12use std::sync::{Arc, Mutex};
13use std::thread;
14use std::time::Duration;
15
16/// Configuration for the session reaper.
17#[derive(Debug, Clone)]
18pub struct ReaperConfig {
19    /// How often to scan for expired sessions.
20    pub scan_interval: Duration,
21}
22
23impl Default for ReaperConfig {
24    fn default() -> Self {
25        Self {
26            scan_interval: Duration::from_secs(60),
27        }
28    }
29}
30
31/// Background session reaper. Runs in its own thread, periodically
32/// scanning the coordinator for expired sessions.
33pub struct SessionReaper {
34    coordinator: Arc<Mutex<Coordinator>>,
35    config: ReaperConfig,
36    running: Arc<std::sync::atomic::AtomicBool>,
37}
38
39impl SessionReaper {
40    /// Create a new reaper for the given shared coordinator.
41    pub fn new(coordinator: Arc<Mutex<Coordinator>>, config: ReaperConfig) -> Self {
42        Self {
43            coordinator,
44            config,
45            running: Arc::new(std::sync::atomic::AtomicBool::new(false)),
46        }
47    }
48
49    /// Start the reaper in a background thread.
50    pub fn start(&self) {
51        self.running
52            .store(true, std::sync::atomic::Ordering::SeqCst);
53        let coordinator = Arc::clone(&self.coordinator);
54        let interval = self.config.scan_interval;
55        let running = Arc::clone(&self.running);
56
57        thread::spawn(move || {
58            while running.load(std::sync::atomic::Ordering::SeqCst) {
59                reap_expired_sessions(&coordinator);
60                thread::sleep(interval);
61            }
62        });
63    }
64
65    /// Stop the reaper.
66    pub fn stop(&self) {
67        self.running
68            .store(false, std::sync::atomic::Ordering::SeqCst);
69    }
70
71    /// Run one reaping pass synchronously. Returns the number of
72    /// sessions that were expired.
73    pub fn reap_once(&self) -> usize {
74        reap_expired_sessions(&self.coordinator)
75    }
76}
77
78fn reap_expired_sessions(coordinator: &Arc<Mutex<Coordinator>>) -> usize {
79    let mut coord = coordinator.lock().unwrap();
80    let now = Utc::now();
81
82    let pending: Vec<String> = coord
83        .session_ids()
84        .into_iter()
85        .filter(|sid| coord.session_state(sid) == Some(SessionState::Pending))
86        .collect();
87
88    let mut expired_count = 0;
89    for sid in pending {
90        let should_expire = coord
91            .session_mut(&sid)
92            .map(|s| !s.is_unlocked(now))
93            .unwrap_or(false);
94        if should_expire {
95            coord.set_session_state(&sid, SessionState::Expired);
96            coord
97                .audit_log_mut()
98                .append(sid.clone(), AuditEvent::Expired);
99            expired_count += 1;
100            tracing::info!(session = %sid, "session expired by reaper");
101        }
102    }
103    expired_count
104}
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109    use crate::coordinator::coordinator::Coordinator;
110    use crate::coordinator::session::SessionRequest;
111
112    fn make_request(unlock_minutes: i64) -> SessionRequest {
113        SessionRequest {
114            quorum_id: "q".into(),
115            scheme: "CMP20".into(),
116            message: vec![0; 32],
117            threshold: 2,
118            num_parties: 3,
119            unlock_window_minutes: unlock_minutes as u32,
120            requested_by: "test".into(),
121        }
122    }
123
124    #[test]
125    fn reap_expires_past_unlock_window() {
126        let coord = Arc::new(Mutex::new(Coordinator::new()));
127        coord
128            .lock()
129            .unwrap()
130            .create_session(make_request(0))
131            .unwrap();
132        std::thread::sleep(Duration::from_millis(1100));
133        let reaper = SessionReaper::new(Arc::clone(&coord), ReaperConfig::default());
134        let count = reaper.reap_once();
135        assert_eq!(count, 1);
136    }
137
138    #[test]
139    fn reap_preserves_active_sessions() {
140        let coord = Arc::new(Mutex::new(Coordinator::new()));
141        coord
142            .lock()
143            .unwrap()
144            .create_session(make_request(240))
145            .unwrap();
146        let reaper = SessionReaper::new(Arc::clone(&coord), ReaperConfig::default());
147        let count = reaper.reap_once();
148        assert_eq!(count, 0);
149    }
150
151    #[test]
152    fn reap_does_not_touch_completed_sessions() {
153        let coord = Arc::new(Mutex::new(Coordinator::new()));
154        let sid = coord
155            .lock()
156            .unwrap()
157            .create_session(make_request(0))
158            .unwrap();
159        coord
160            .lock()
161            .unwrap()
162            .set_session_state(&sid, SessionState::Completed);
163        std::thread::sleep(Duration::from_millis(1100));
164        let reaper = SessionReaper::new(Arc::clone(&coord), ReaperConfig::default());
165        let count = reaper.reap_once();
166        assert_eq!(count, 0);
167    }
168
169    #[test]
170    fn reap_multiple_expired() {
171        let coord = Arc::new(Mutex::new(Coordinator::new()));
172        coord
173            .lock()
174            .unwrap()
175            .create_session(make_request(0))
176            .unwrap();
177        coord
178            .lock()
179            .unwrap()
180            .create_session(make_request(0))
181            .unwrap();
182        coord
183            .lock()
184            .unwrap()
185            .create_session(make_request(240))
186            .unwrap();
187        std::thread::sleep(Duration::from_millis(1100));
188        let reaper = SessionReaper::new(Arc::clone(&coord), ReaperConfig::default());
189        let count = reaper.reap_once();
190        assert_eq!(count, 2);
191    }
192
193    #[test]
194    fn reap_empty_coordinator_returns_zero() {
195        let coord = Arc::new(Mutex::new(Coordinator::new()));
196        let reaper = SessionReaper::new(coord, ReaperConfig::default());
197        assert_eq!(reaper.reap_once(), 0);
198    }
199
200    #[test]
201    fn config_default_scan_interval_is_60s() {
202        let config = ReaperConfig::default();
203        assert_eq!(config.scan_interval, Duration::from_secs(60));
204    }
205}
confium_coordinator/coordinator/reaper.rs

confium_coordinator/coordinator/
reaper.rs
