oximedia_distributed/
compaction.rs

1//! Raft log compaction via snapshotting.
2//!
3//! Provides `LogCompactor` for truncating the Raft log up to a committed
4//! snapshot index, and `Snapshot` to carry the state-machine snapshot data.
5
6use crate::distributed_enhancements::LogEntry;
7
8/// A snapshot of the replicated state machine at a given log index.
9#[derive(Debug, Clone)]
10pub struct Snapshot {
11    /// The log index that this snapshot covers (inclusive).
12    pub last_included_index: u64,
13    /// The term of the last included log entry.
14    pub last_included_term: u64,
15    /// Encoded state-machine data at `last_included_index`.
16    pub data: Vec<u8>,
17}
18
19/// Handles compaction of the Raft log by replacing committed entries with a
20/// snapshot, thereby bounding log growth.
21pub struct LogCompactor;
22
23impl LogCompactor {
24    /// Create a new `LogCompactor`.
25    #[must_use]
26    pub fn new() -> Self {
27        Self
28    }
29
30    /// Compact the Raft log up to (and including) `snapshot_idx`.
31    ///
32    /// All log entries with `index <= snapshot_idx` are removed from `log`.
33    /// Returns a `Snapshot` that captures the term of the last removed entry
34    /// (or 0 if no entry was found at `snapshot_idx`) and an empty state-data
35    /// payload.  Callers are expected to replace the payload with the actual
36    /// serialised state-machine snapshot before persisting.
37    ///
38    /// # Arguments
39    ///
40    /// * `log`          - Mutable reference to the Raft log.
41    /// * `snapshot_idx` - The log index up to which compaction should proceed.
42    ///
43    /// # Returns
44    ///
45    /// A [`Snapshot`] representing the compaction boundary.
46    #[must_use]
47    pub fn compact(log: &mut Vec<LogEntry>, snapshot_idx: u64) -> Snapshot {
48        // Find the term of the last entry being snapshotted
49        let last_included_term = log
50            .iter()
51            .rev()
52            .find(|e| e.index <= snapshot_idx)
53            .map(|e| e.term)
54            .unwrap_or(0);
55
56        // Remove all entries covered by the snapshot
57        log.retain(|e| e.index > snapshot_idx);
58
59        Snapshot {
60            last_included_index: snapshot_idx,
61            last_included_term,
62            data: Vec::new(),
63        }
64    }
65
66    /// Apply a received snapshot to a log: truncate the entire log if the
67    /// snapshot is newer than all existing entries.
68    ///
69    /// Returns `true` if the snapshot was applied (log was truncated), `false`
70    /// if the log already has entries beyond the snapshot (snapshot is stale).
71    pub fn install_snapshot(log: &mut Vec<LogEntry>, snapshot: &Snapshot) -> bool {
72        let log_last_index = log.last().map(|e| e.index).unwrap_or(0);
73        if snapshot.last_included_index >= log_last_index {
74            log.clear();
75            true
76        } else {
77            // Partial install: remove only the covered prefix
78            log.retain(|e| e.index > snapshot.last_included_index);
79            false
80        }
81    }
82}
83
84impl Default for LogCompactor {
85    fn default() -> Self {
86        Self::new()
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93
94    fn make_log(count: u64) -> Vec<LogEntry> {
95        (1..=count)
96            .map(|i| LogEntry {
97                term: (i + 1) / 2,
98                index: i,
99                command: format!("cmd-{i}"),
100            })
101            .collect()
102    }
103
104    #[test]
105    fn test_compact_removes_entries_up_to_index() {
106        let mut log = make_log(10);
107        let snap = LogCompactor::compact(&mut log, 5);
108        assert_eq!(snap.last_included_index, 5);
109        assert!(log.iter().all(|e| e.index > 5));
110        assert_eq!(log.len(), 5);
111    }
112
113    #[test]
114    fn test_compact_all_entries() {
115        let mut log = make_log(5);
116        let snap = LogCompactor::compact(&mut log, 5);
117        assert_eq!(snap.last_included_index, 5);
118        assert!(log.is_empty());
119    }
120
121    #[test]
122    fn test_compact_no_matching_entries() {
123        let mut log = make_log(5);
124        // snapshot_idx 0 → nothing removed (all have index >= 1)
125        let snap = LogCompactor::compact(&mut log, 0);
126        assert_eq!(snap.last_included_index, 0);
127        assert_eq!(snap.last_included_term, 0);
128        assert_eq!(log.len(), 5);
129    }
130
131    #[test]
132    fn test_compact_empty_log() {
133        let mut log: Vec<LogEntry> = Vec::new();
134        let snap = LogCompactor::compact(&mut log, 10);
135        assert_eq!(snap.last_included_index, 10);
136        assert_eq!(snap.last_included_term, 0);
137        assert!(log.is_empty());
138    }
139
140    #[test]
141    fn test_install_snapshot_truncates_older_log() {
142        let mut log = make_log(5);
143        let snap = Snapshot {
144            last_included_index: 10,
145            last_included_term: 5,
146            data: Vec::new(),
147        };
148        let applied = LogCompactor::install_snapshot(&mut log, &snap);
149        assert!(applied);
150        assert!(log.is_empty());
151    }
152
153    #[test]
154    fn test_install_snapshot_stale_does_not_clear_all() {
155        let mut log = make_log(10);
156        let snap = Snapshot {
157            last_included_index: 5,
158            last_included_term: 3,
159            data: Vec::new(),
160        };
161        LogCompactor::install_snapshot(&mut log, &snap);
162        // Entries 6–10 remain
163        assert!(log.iter().all(|e| e.index > 5));
164    }
165}
oximedia_distributed/compaction.rs

oximedia_distributed/
compaction.rs
