confium_coordinator/
wal.rs

1//! Session write-ahead log — crash recovery via append-only log.
2//!
3//! Every session state transition is appended to a WAL before being
4//! applied to the in-memory state. On crash, the WAL is replayed to
5//! reconstruct the session state up to the last persisted transition.
6
7use serde::{Deserialize, Serialize};
8use std::fs::OpenOptions;
9use std::io::{BufRead, BufReader, Write};
10use std::path::{Path, PathBuf};
11use std::sync::Mutex;
12
13/// A single WAL entry recording a state transition.
14#[derive(Debug, Clone, Serialize, Deserialize)]
15pub struct WalEntry {
16    /// Monotonic sequence number.
17    pub seq: u64,
18    /// Session ID.
19    pub session_id: String,
20    /// The state transition.
21    pub transition: StateTransition,
22    /// When the entry was recorded.
23    pub timestamp: chrono::DateTime<chrono::Utc>,
24}
25
26/// A session state transition.
27#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
28#[serde(rename_all = "snake_case")]
29pub enum StateTransition {
30    /// Session created.
31    Created {
32        quorum_id: String,
33        scheme: String,
34        threshold: u32,
35    },
36    /// Commitment received from signer.
37    CommitmentReceived { signer_id: String },
38    /// Share received from signer.
39    ShareReceived { signer_id: String },
40    /// Session completed (signature aggregated).
41    Completed,
42    /// Session expired.
43    Expired,
44    /// Session aborted.
45    Aborted { reason: String },
46}
47
48/// Append-only write-ahead log backed by a file.
49pub struct SessionWal {
50    path: PathBuf,
51    next_seq: Mutex<u64>,
52}
53
54impl SessionWal {
55    /// Create or open a WAL at the given path.
56    pub fn open(path: impl AsRef<Path>) -> std::io::Result<Self> {
57        let path = path.as_ref().to_path_buf();
58        let next_seq = if path.exists() {
59            let max_seq = Self::read_all_from_path(&path)?
60                .into_iter()
61                .map(|e| e.seq)
62                .max()
63                .unwrap_or(0);
64            max_seq + 1
65        } else {
66            1
67        };
68        Ok(Self {
69            path,
70            next_seq: Mutex::new(next_seq),
71        })
72    }
73
74    /// Append a transition to the WAL. Returns the assigned sequence number.
75    pub fn append(&self, session_id: &str, transition: StateTransition) -> std::io::Result<u64> {
76        let seq = {
77            let mut next = self.next_seq.lock().unwrap();
78            let current = *next;
79            *next += 1;
80            current
81        };
82        let entry = WalEntry {
83            seq,
84            session_id: session_id.into(),
85            transition,
86            timestamp: chrono::Utc::now(),
87        };
88        let json = serde_json::to_string(&entry)
89            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
90        let mut file = OpenOptions::new()
91            .create(true)
92            .append(true)
93            .open(&self.path)?;
94        writeln!(file, "{json}")?;
95        file.sync_data()?;
96        Ok(seq)
97    }
98
99    /// Read all entries from the WAL (for replay).
100    pub fn read_all(&self) -> std::io::Result<Vec<WalEntry>> {
101        Self::read_all_from_path(&self.path)
102    }
103
104    /// Read all entries from a WAL file path.
105    fn read_all_from_path(path: &Path) -> std::io::Result<Vec<WalEntry>> {
106        if !path.exists() {
107            return Ok(Vec::new());
108        }
109        let file = OpenOptions::new().read(true).open(path)?;
110        let reader = BufReader::new(file);
111        let mut entries = Vec::new();
112        for line in reader.lines() {
113            let line = line?;
114            if line.trim().is_empty() {
115                continue;
116            }
117            match serde_json::from_str::<WalEntry>(&line) {
118                Ok(entry) => entries.push(entry),
119                Err(_) => continue,
120            }
121        }
122        Ok(entries)
123    }
124
125    /// Truncate (clear) the WAL. Called after a successful checkpoint.
126    pub fn truncate(&self) -> std::io::Result<()> {
127        OpenOptions::new()
128            .write(true)
129            .truncate(true)
130            .open(&self.path)?;
131        let mut next = self.next_seq.lock().unwrap();
132        *next = 1;
133        Ok(())
134    }
135
136    /// Current next-sequence number.
137    pub fn next_seq(&self) -> u64 {
138        *self.next_seq.lock().unwrap()
139    }
140}
141
142#[cfg(test)]
143mod tests {
144    use super::*;
145
146    #[test]
147    fn append_and_read() {
148        let tmp = tempfile::tempdir().unwrap();
149        let wal_path = tmp.path().join("wal.jsonl");
150        let wal = SessionWal::open(&wal_path).unwrap();
151        wal.append(
152            "s1",
153            StateTransition::Created {
154                quorum_id: "q".into(),
155                scheme: "CMP20".into(),
156                threshold: 2,
157            },
158        )
159        .unwrap();
160        wal.append(
161            "s1",
162            StateTransition::CommitmentReceived {
163                signer_id: "alice".into(),
164            },
165        )
166        .unwrap();
167        wal.append("s1", StateTransition::Completed).unwrap();
168
169        let entries = wal.read_all().unwrap();
170        assert_eq!(entries.len(), 3);
171        assert_eq!(entries[0].seq, 1);
172        assert_eq!(entries[1].seq, 2);
173        assert_eq!(entries[2].seq, 3);
174    }
175
176    #[test]
177    fn sequence_numbers_monotonic() {
178        let tmp = tempfile::tempdir().unwrap();
179        let wal = SessionWal::open(tmp.path().join("wal.jsonl")).unwrap();
180        let s1 = wal.append("s1", StateTransition::Completed).unwrap();
181        let s2 = wal.append("s1", StateTransition::Expired).unwrap();
182        let s3 = wal.append("s2", StateTransition::Completed).unwrap();
183        assert!(s1 < s2);
184        assert!(s2 < s3);
185    }
186
187    #[test]
188    fn reopen_continues_sequence() {
189        let tmp = tempfile::tempdir().unwrap();
190        let wal_path = tmp.path().join("wal.jsonl");
191        {
192            let wal = SessionWal::open(&wal_path).unwrap();
193            wal.append("s1", StateTransition::Completed).unwrap();
194            wal.append("s1", StateTransition::Expired).unwrap();
195        }
196        {
197            let wal = SessionWal::open(&wal_path).unwrap();
198            assert_eq!(wal.next_seq(), 3);
199            let seq = wal.append("s2", StateTransition::Completed).unwrap();
200            assert_eq!(seq, 3);
201        }
202    }
203
204    #[test]
205    fn truncate_clears_entries() {
206        let tmp = tempfile::tempdir().unwrap();
207        let wal = SessionWal::open(tmp.path().join("wal.jsonl")).unwrap();
208        wal.append("s1", StateTransition::Completed).unwrap();
209        wal.truncate().unwrap();
210        assert_eq!(wal.read_all().unwrap().len(), 0);
211        assert_eq!(wal.next_seq(), 1);
212    }
213
214    #[test]
215    fn empty_wal_returns_empty_vec() {
216        let tmp = tempfile::tempdir().unwrap();
217        let wal = SessionWal::open(tmp.path().join("nonexistent.jsonl")).unwrap();
218        assert!(wal.read_all().unwrap().is_empty());
219    }
220
221    #[test]
222    fn all_transition_types_serialize() {
223        let transitions = vec![
224            StateTransition::Created {
225                quorum_id: "q".into(),
226                scheme: "CMP20".into(),
227                threshold: 2,
228            },
229            StateTransition::CommitmentReceived {
230                signer_id: "a".into(),
231            },
232            StateTransition::ShareReceived {
233                signer_id: "a".into(),
234            },
235            StateTransition::Completed,
236            StateTransition::Expired,
237            StateTransition::Aborted {
238                reason: "test".into(),
239            },
240        ];
241        for t in &transitions {
242            let json = serde_json::to_string(t).unwrap();
243            let recovered: StateTransition = serde_json::from_str(&json).unwrap();
244            assert_eq!(t, &recovered);
245        }
246    }
247
248    #[test]
249    fn corrupt_lines_skipped() {
250        let tmp = tempfile::tempdir().unwrap();
251        let path = tmp.path().join("wal.jsonl");
252        std::fs::write(&path, "not json\n{\"seq\":1,\"session_id\":\"s\",\"transition\":{\"created\":{\"quorum_id\":\"q\",\"scheme\":\"CMP20\",\"threshold\":2}},\"timestamp\":\"2026-01-01T00:00:00Z\"}\n").unwrap();
253        let wal = SessionWal::open(&path).unwrap();
254        let entries = wal.read_all().unwrap();
255        assert_eq!(entries.len(), 1);
256        assert_eq!(entries[0].session_id, "s");
257    }
258}
confium_coordinator/wal.rs

confium_coordinator/
wal.rs
