confium_coordinator/coordinator/
scheduler.rs

1//! Share rotation scheduler — triggers periodic Herzberg refresh.
2//!
3//! Proactive share refresh rotates shares without changing the joint
4//! public key, protecting against slow-share-compromise adversaries.
5//! The scheduler runs in a background thread and periodically invokes
6//! the refresh callback.
7
8use chrono::{DateTime, Duration, Utc};
9use std::sync::{Arc, Mutex};
10use std::thread;
11
12/// Configuration for the rotation scheduler.
13#[derive(Debug, Clone)]
14pub struct RotationConfig {
15    /// Human-readable name for this rotation schedule.
16    pub name: String,
17    /// Time between refresh triggers.
18    pub interval: Duration,
19    /// Next scheduled run time.
20    pub next_run: DateTime<Utc>,
21    /// Whether the scheduler is paused.
22    pub paused: bool,
23}
24
25impl RotationConfig {
26    /// Create a new schedule that runs every `interval`, starting now.
27    pub fn every(name: &str, interval: Duration) -> Self {
28        Self {
29            name: name.into(),
30            interval,
31            next_run: Utc::now() + interval,
32            paused: false,
33        }
34    }
35
36    /// Pause the scheduler (it will skip triggers until resumed).
37    pub fn pause(&mut self) {
38        self.paused = true;
39    }
40
41    /// Resume the scheduler.
42    pub fn resume(&mut self) {
43        self.paused = false;
44        self.next_run = Utc::now() + self.interval;
45    }
46}
47
48/// A refresh trigger callback. The scheduler calls this when it's
49/// time to rotate shares. The callback receives the schedule name.
50pub type RefreshCallback = Arc<dyn Fn(&str) -> Result<(), String> + Send + Sync>;
51
52/// The rotation scheduler. Runs in a background thread, periodically
53/// triggering the refresh callback.
54pub struct RotationScheduler {
55    config: Arc<Mutex<RotationConfig>>,
56    callback: RefreshCallback,
57    running: Arc<std::sync::atomic::AtomicBool>,
58}
59
60impl RotationScheduler {
61    /// Create a new scheduler with the given config and callback.
62    pub fn new(config: RotationConfig, callback: RefreshCallback) -> Self {
63        Self {
64            config: Arc::new(Mutex::new(config)),
65            callback,
66            running: Arc::new(std::sync::atomic::AtomicBool::new(false)),
67        }
68    }
69
70    /// Get a handle to the config for inspection or modification.
71    pub fn config_handle(&self) -> Arc<Mutex<RotationConfig>> {
72        Arc::clone(&self.config)
73    }
74
75    /// Start the scheduler in a background thread.
76    pub fn start(&self) {
77        self.running
78            .store(true, std::sync::atomic::Ordering::SeqCst);
79        let config = Arc::clone(&self.config);
80        let callback = Arc::clone(&self.callback);
81        let running = Arc::clone(&self.running);
82
83        thread::spawn(move || {
84            while running.load(std::sync::atomic::Ordering::SeqCst) {
85                let should_run = {
86                    let cfg = config.lock().unwrap();
87                    !cfg.paused && Utc::now() >= cfg.next_run
88                };
89
90                if should_run {
91                    let name = {
92                        let cfg = config.lock().unwrap();
93                        cfg.name.clone()
94                    };
95                    match callback(&name) {
96                        Ok(()) => tracing::info!(schedule = %name, "refresh completed"),
97                        Err(e) => tracing::error!(schedule = %name, error = %e, "refresh failed"),
98                    }
99                    let mut cfg = config.lock().unwrap();
100                    cfg.next_run = Utc::now() + cfg.interval;
101                }
102
103                thread::sleep(std::time::Duration::from_secs(1));
104            }
105        });
106    }
107
108    /// Stop the scheduler.
109    pub fn stop(&self) {
110        self.running
111            .store(false, std::sync::atomic::Ordering::SeqCst);
112    }
113
114    /// Trigger a manual refresh immediately (bypasses the schedule).
115    pub fn trigger_now(&self) -> Result<(), String> {
116        let name = self.config.lock().unwrap().name.clone();
117        (self.callback)(&name)
118    }
119}
120
121#[cfg(test)]
122mod tests {
123    use super::*;
124    use std::sync::atomic::{AtomicU32, Ordering};
125
126    #[test]
127    fn config_every_sets_interval() {
128        let config = RotationConfig::every("hourly", Duration::hours(1));
129        assert_eq!(config.name, "hourly");
130        assert_eq!(config.interval, Duration::hours(1));
131        assert!(!config.paused);
132    }
133
134    #[test]
135    fn pause_prevents_and_resume_reenables() {
136        let mut config = RotationConfig::every("test", Duration::minutes(5));
137        config.pause();
138        assert!(config.paused);
139        config.resume();
140        assert!(!config.paused);
141    }
142
143    #[test]
144    fn trigger_now_invokes_callback() {
145        let counter = Arc::new(AtomicU32::new(0));
146        let counter_clone = Arc::clone(&counter);
147        let callback: RefreshCallback = Arc::new(move |_name| {
148            counter_clone.fetch_add(1, Ordering::SeqCst);
149            Ok(())
150        });
151        let config = RotationConfig::every("test", Duration::hours(24));
152        let scheduler = RotationScheduler::new(config, callback);
153        scheduler.trigger_now().unwrap();
154        assert_eq!(counter.load(Ordering::SeqCst), 1);
155    }
156
157    #[test]
158    fn trigger_now_propagates_error() {
159        let callback: RefreshCallback = Arc::new(|_| Err("simulated failure".into()));
160        let config = RotationConfig::every("test", Duration::hours(24));
161        let scheduler = RotationScheduler::new(config, callback);
162        assert!(scheduler.trigger_now().is_err());
163    }
164
165    #[test]
166    fn config_handle_allows_modification() {
167        let config = RotationConfig::every("test", Duration::hours(1));
168        let scheduler = RotationScheduler::new(config, Arc::new(|_| Ok(())));
169        let handle = scheduler.config_handle();
170        {
171            let mut cfg = handle.lock().unwrap();
172            cfg.pause();
173        }
174        let cfg = handle.lock().unwrap();
175        assert!(cfg.paused);
176    }
177}
confium_coordinator/coordinator/scheduler.rs

confium_coordinator/coordinator/
scheduler.rs
