confium_coordinator/coordinator/
checkpoint.rs

1//! Session checkpoint manager — periodic WAL checkpointing.
2
3use std::sync::Mutex;
4use std::time::{Duration, Instant};
5
6/// Configuration for checkpointing.
7#[derive(Debug, Clone)]
8pub struct CheckpointConfig {
9    /// How often to checkpoint.
10    pub interval: Duration,
11    /// Maximum entries before forcing a checkpoint.
12    pub max_entries_before_checkpoint: usize,
13}
14
15impl Default for CheckpointConfig {
16    fn default() -> Self {
17        Self {
18            interval: Duration::from_secs(300),
19            max_entries_before_checkpoint: 1000,
20        }
21    }
22}
23
24/// Tracks when the next checkpoint should occur.
25pub struct CheckpointManager {
26    config: CheckpointConfig,
27    last_checkpoint: Mutex<Instant>,
28    entries_since_checkpoint: Mutex<usize>,
29    total_checkpoints: Mutex<u64>,
30}
31
32impl CheckpointManager {
33    pub fn new(config: CheckpointConfig) -> Self {
34        Self {
35            config,
36            last_checkpoint: Mutex::new(Instant::now()),
37            entries_since_checkpoint: Mutex::new(0),
38            total_checkpoints: Mutex::new(0),
39        }
40    }
41
42    /// Record that a WAL entry was appended.
43    pub fn record_entry(&self) {
44        *self.entries_since_checkpoint.lock().unwrap() += 1;
45    }
46
47    /// Should a checkpoint be taken now?
48    pub fn should_checkpoint(&self) -> bool {
49        let elapsed = self.last_checkpoint.lock().unwrap().elapsed();
50        let entries = *self.entries_since_checkpoint.lock().unwrap();
51        elapsed >= self.config.interval || entries >= self.config.max_entries_before_checkpoint
52    }
53
54    /// Record that a checkpoint was taken.
55    pub fn checkpoint_taken(&self) {
56        *self.last_checkpoint.lock().unwrap() = Instant::now();
57        *self.entries_since_checkpoint.lock().unwrap() = 0;
58        *self.total_checkpoints.lock().unwrap() += 1;
59    }
60
61    /// Total checkpoints taken since start.
62    pub fn total_checkpoints(&self) -> u64 {
63        *self.total_checkpoints.lock().unwrap()
64    }
65
66    /// Entries since last checkpoint.
67    pub fn entries_since_checkpoint(&self) -> usize {
68        *self.entries_since_checkpoint.lock().unwrap()
69    }
70}
71
72#[cfg(test)]
73mod tests {
74    use super::*;
75
76    #[test]
77    fn default_interval_5min() {
78        let config = CheckpointConfig::default();
79        assert_eq!(config.interval, Duration::from_secs(300));
80    }
81
82    #[test]
83    fn no_checkpoint_needed_initially() {
84        let mgr = CheckpointManager::new(CheckpointConfig::default());
85        assert!(!mgr.should_checkpoint());
86    }
87
88    #[test]
89    fn checkpoint_after_max_entries() {
90        let config = CheckpointConfig {
91            interval: Duration::from_secs(3600),
92            max_entries_before_checkpoint: 5,
93        };
94        let mgr = CheckpointManager::new(config);
95        for _ in 0..5 {
96            mgr.record_entry();
97        }
98        assert!(mgr.should_checkpoint());
99    }
100
101    #[test]
102    fn checkpoint_resets_counter() {
103        let config = CheckpointConfig {
104            interval: Duration::from_secs(3600),
105            max_entries_before_checkpoint: 3,
106        };
107        let mgr = CheckpointManager::new(config);
108        for _ in 0..3 {
109            mgr.record_entry();
110        }
111        assert!(mgr.should_checkpoint());
112        mgr.checkpoint_taken();
113        assert!(!mgr.should_checkpoint());
114        assert_eq!(mgr.entries_since_checkpoint(), 0);
115    }
116
117    #[test]
118    fn total_checkpoints_increments() {
119        let mgr = CheckpointManager::new(CheckpointConfig::default());
120        assert_eq!(mgr.total_checkpoints(), 0);
121        mgr.checkpoint_taken();
122        mgr.checkpoint_taken();
123        assert_eq!(mgr.total_checkpoints(), 2);
124    }
125
126    #[test]
127    fn entries_accumulate() {
128        let mgr = CheckpointManager::new(CheckpointConfig::default());
129        mgr.record_entry();
130        mgr.record_entry();
131        mgr.record_entry();
132        assert_eq!(mgr.entries_since_checkpoint(), 3);
133    }
134}
confium_coordinator/coordinator/checkpoint.rs

confium_coordinator/coordinator/
checkpoint.rs
