hatchery_core/
journal.rs

1//! Node journal: persistent append-only log + C2 gap recovery.
2//!
3//! Core never talks to nodes. When RESUME asks for a seq older than the buffer,
4//! C2 fetches missing events from the node's journal over the uplink
5//! (`JournalReq` / `JournalReply`).
6
7use std::collections::BTreeMap;
8use std::fs::{File, OpenOptions};
9use std::io::{Read, Seek, SeekFrom, Write};
10use std::path::{Path, PathBuf};
11use std::sync::{Arc, Mutex};
12
13use crate::{JournalReplyBody, JournalWireEntry};
14
15use crate::buffer::BufferedEvent;
16
17/// Opaque journal entry as the node stores it for retransmission.
18#[derive(Clone, Debug, PartialEq, Eq)]
19pub struct JournalEntry {
20    pub seq: u64,
21    pub channel_id: u32,
22    pub payload: Vec<u8>,
23}
24
25/// C2-side handle to ask a connected node for a journal slice.
26pub trait NodeJournal: Send + Sync {
27    /// Return entries with `seq > after` up to `limit`, oldest first.
28    fn replay_after(
29        &self,
30        node_id: &str,
31        channel_id: u32,
32        after: u64,
33        limit: usize,
34    ) -> Result<Vec<JournalEntry>, JournalError>;
35}
36
37#[derive(Clone, Debug, PartialEq, Eq)]
38pub enum JournalError {
39    NodeUnknown,
40    NodeUnavailable,
41    Io(String),
42    Corrupt,
43    Other(String),
44}
45
46impl std::fmt::Display for JournalError {
47    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
48        match self {
49            Self::NodeUnknown => write!(f, "node unknown"),
50            Self::NodeUnavailable => write!(f, "node unavailable"),
51            Self::Io(msg) => write!(f, "journal io: {msg}"),
52            Self::Corrupt => write!(f, "journal corrupt"),
53            Self::Other(msg) => write!(f, "{msg}"),
54        }
55    }
56}
57
58impl std::error::Error for JournalError {}
59
60/// In-memory journal for tests.
61#[derive(Clone, Default)]
62pub struct MemoryJournal {
63    inner: Arc<Mutex<BTreeMap<String, Vec<JournalEntry>>>>,
64}
65
66impl MemoryJournal {
67    pub fn new() -> Self {
68        Self::default()
69    }
70
71    pub fn append(&self, node_id: &str, channel_id: u32, seq: u64, payload: Vec<u8>) {
72        let mut guard = self.inner.lock().expect("journal lock");
73        guard
74            .entry(node_id.to_owned())
75            .or_default()
76            .push(JournalEntry {
77                seq,
78                channel_id,
79                payload,
80            });
81    }
82}
83
84impl NodeJournal for MemoryJournal {
85    fn replay_after(
86        &self,
87        node_id: &str,
88        channel_id: u32,
89        after: u64,
90        limit: usize,
91    ) -> Result<Vec<JournalEntry>, JournalError> {
92        let guard = self.inner.lock().expect("journal lock");
93        let Some(entries) = guard.get(node_id) else {
94            return Err(JournalError::NodeUnknown);
95        };
96        Ok(entries
97            .iter()
98            .filter(|e| e.channel_id == channel_id && e.seq > after)
99            .take(limit)
100            .cloned()
101            .collect())
102    }
103}
104
105/// On-disk journal format (per node, one file per channel or a single multiplexed file).
106///
107/// Record: `u32 channel_id ‖ u64 seq ‖ u32 len ‖ [len] payload` (little-endian).
108/// Append-only; readers scan from the start (or from a cached offset index).
109#[derive(Clone)]
110pub struct FileJournal {
111    path: PathBuf,
112    state: Arc<Mutex<FileJournalState>>,
113}
114
115struct FileJournalState {
116    /// Next seq to assign per channel.
117    next_seq: BTreeMap<u32, u64>,
118}
119
120impl FileJournal {
121    pub fn open(path: impl Into<PathBuf>) -> Result<Self, JournalError> {
122        let path = path.into();
123        if let Some(parent) = path.parent() {
124            std::fs::create_dir_all(parent).map_err(|e| JournalError::Io(e.to_string()))?;
125        }
126        let mut next_seq = BTreeMap::new();
127        if path.exists() {
128            for entry in scan_file(&path)? {
129                let next = next_seq.entry(entry.channel_id).or_insert(1);
130                *next = (*next).max(entry.seq + 1);
131            }
132        } else {
133            File::create(&path).map_err(|e| JournalError::Io(e.to_string()))?;
134        }
135        Ok(Self {
136            path,
137            state: Arc::new(Mutex::new(FileJournalState { next_seq })),
138        })
139    }
140
141    pub fn path(&self) -> &Path {
142        &self.path
143    }
144
145    /// Append a payload; returns the assigned seq.
146    pub fn append(&self, channel_id: u32, payload: &[u8]) -> Result<u64, JournalError> {
147        if payload.len() > 1024 * 1024 {
148            return Err(JournalError::Other("payload too large".into()));
149        }
150        let mut guard = self.state.lock().expect("journal lock");
151        let seq = {
152            let entry = guard.next_seq.entry(channel_id).or_insert(1);
153            let seq = *entry;
154            *entry = seq + 1;
155            seq
156        };
157        let mut file = OpenOptions::new()
158            .append(true)
159            .open(&self.path)
160            .map_err(|e| JournalError::Io(e.to_string()))?;
161        write_record(&mut file, channel_id, seq, payload)?;
162        file.flush().map_err(|e| JournalError::Io(e.to_string()))?;
163        Ok(seq)
164    }
165
166    pub fn replay_after(
167        &self,
168        channel_id: u32,
169        after: u64,
170        limit: usize,
171    ) -> Result<Vec<JournalEntry>, JournalError> {
172        let mut out = Vec::new();
173        for entry in scan_file(&self.path)? {
174            if entry.channel_id == channel_id && entry.seq > after {
175                out.push(entry);
176                if out.len() >= limit {
177                    break;
178                }
179            }
180        }
181        Ok(out)
182    }
183
184    /// Build a wire reply body for an uplink JournalReq.
185    pub fn reply_body(
186        &self,
187        channel_id: u32,
188        after: u64,
189        limit: usize,
190    ) -> Result<JournalReplyBody, JournalError> {
191        let entries = self.replay_after(channel_id, after, limit)?;
192        Ok(JournalReplyBody {
193            after,
194            entries: entries
195                .into_iter()
196                .map(|e| JournalWireEntry {
197                    seq: e.seq,
198                    payload: e.payload,
199                })
200                .collect(),
201        })
202    }
203}
204
205/// Adapter: FileJournal as NodeJournal for a single node id (local / test).
206pub struct LocalFileJournal {
207    node_id: String,
208    journal: FileJournal,
209}
210
211impl LocalFileJournal {
212    pub fn new(node_id: impl Into<String>, journal: FileJournal) -> Self {
213        Self {
214            node_id: node_id.into(),
215            journal,
216        }
217    }
218
219    pub fn journal(&self) -> &FileJournal {
220        &self.journal
221    }
222}
223
224impl NodeJournal for LocalFileJournal {
225    fn replay_after(
226        &self,
227        node_id: &str,
228        channel_id: u32,
229        after: u64,
230        limit: usize,
231    ) -> Result<Vec<JournalEntry>, JournalError> {
232        if node_id != self.node_id {
233            return Err(JournalError::NodeUnknown);
234        }
235        self.journal.replay_after(channel_id, after, limit)
236    }
237}
238
239fn write_record(file: &mut File, channel_id: u32, seq: u64, payload: &[u8]) -> Result<(), JournalError> {
240    file.write_all(&channel_id.to_le_bytes())
241        .map_err(|e| JournalError::Io(e.to_string()))?;
242    file.write_all(&seq.to_le_bytes())
243        .map_err(|e| JournalError::Io(e.to_string()))?;
244    file.write_all(&(payload.len() as u32).to_le_bytes())
245        .map_err(|e| JournalError::Io(e.to_string()))?;
246    file.write_all(payload)
247        .map_err(|e| JournalError::Io(e.to_string()))?;
248    Ok(())
249}
250
251fn scan_file(path: &Path) -> Result<Vec<JournalEntry>, JournalError> {
252    let mut file = File::open(path).map_err(|e| JournalError::Io(e.to_string()))?;
253    let mut out = Vec::new();
254    loop {
255        let mut hdr = [0u8; 4 + 8 + 4];
256        match file.read_exact(&mut hdr) {
257            Ok(()) => {}
258            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
259            Err(e) => return Err(JournalError::Io(e.to_string())),
260        }
261        let channel_id = u32::from_le_bytes([hdr[0], hdr[1], hdr[2], hdr[3]]);
262        let seq = u64::from_le_bytes([
263            hdr[4], hdr[5], hdr[6], hdr[7], hdr[8], hdr[9], hdr[10], hdr[11],
264        ]);
265        let len = u32::from_le_bytes([hdr[12], hdr[13], hdr[14], hdr[15]]) as usize;
266        if len > 1024 * 1024 {
267            return Err(JournalError::Corrupt);
268        }
269        let mut payload = vec![0u8; len];
270        file.read_exact(&mut payload)
271            .map_err(|_| JournalError::Corrupt)?;
272        out.push(JournalEntry {
273            seq,
274            channel_id,
275            payload,
276        });
277    }
278    // Ensure we did not stop mid-record with leftover bytes that are not EOF-aligned.
279    let pos = file.stream_position().map_err(|e| JournalError::Io(e.to_string()))?;
280    let len = file.seek(SeekFrom::End(0)).map_err(|e| JournalError::Io(e.to_string()))?;
281    if pos != len && pos + (4 + 8 + 4) as u64 <= len {
282        // Trailing incomplete record — treat as corrupt rather than silent truncate.
283        return Err(JournalError::Corrupt);
284    }
285    Ok(out)
286}
287
288/// Recover events for a RESUME that falls before the in-memory buffer.
289pub fn recover_gap(
290    journal: &dyn NodeJournal,
291    node_id: &str,
292    channel_id: u32,
293    after: u64,
294    until_exclusive: Option<u64>,
295    limit: usize,
296) -> Result<Vec<BufferedEvent>, JournalError> {
297    let entries = journal.replay_after(node_id, channel_id, after, limit)?;
298    Ok(entries
299        .into_iter()
300        .filter(|e| until_exclusive.map(|u| e.seq < u).unwrap_or(true))
301        .map(|e| BufferedEvent {
302            seq: e.seq,
303            node_id: node_id.to_owned(),
304            channel_id: e.channel_id,
305            payload: e.payload,
306            enqueued_at: std::time::Instant::now(),
307        })
308        .collect())
309}
310
311#[cfg(test)]
312mod tests {
313    use super::*;
314
315    #[test]
316    fn memory_journal_replays_slice() {
317        let j = MemoryJournal::new();
318        for seq in 1..=5 {
319            j.append("n1", 1, seq, vec![seq as u8]);
320        }
321        let got = j.replay_after("n1", 1, 2, 10).unwrap();
322        assert_eq!(got.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![3, 4, 5]);
323    }
324
325    #[test]
326    fn recover_gap_stops_before_buffer_oldest() {
327        let j = MemoryJournal::new();
328        for seq in 1..=10 {
329            j.append("n1", 1, seq, vec![seq as u8]);
330        }
331        let recovered = recover_gap(&j, "n1", 1, 1, Some(5), 100).unwrap();
332        assert_eq!(
333            recovered.iter().map(|e| e.seq).collect::<Vec<_>>(),
334            vec![2, 3, 4]
335        );
336    }
337
338    #[test]
339    fn file_journal_persists_across_reopen() {
340        let dir = std::env::temp_dir().join(format!("fj-{}", std::process::id()));
341        let _ = std::fs::remove_dir_all(&dir);
342        std::fs::create_dir_all(&dir).unwrap();
343        let path = dir.join("node.journal");
344        {
345            let j = FileJournal::open(&path).unwrap();
346            assert_eq!(j.append(1, b"a").unwrap(), 1);
347            assert_eq!(j.append(1, b"b").unwrap(), 2);
348            assert_eq!(j.append(2, b"x").unwrap(), 1);
349        }
350        let j = FileJournal::open(&path).unwrap();
351        let got = j.replay_after(1, 0, 10).unwrap();
352        assert_eq!(got.len(), 2);
353        assert_eq!(got[0].payload, b"a");
354        assert_eq!(got[1].payload, b"b");
355        assert_eq!(j.append(1, b"c").unwrap(), 3);
356        let body = j.reply_body(1, 1, 10).unwrap();
357        assert_eq!(body.entries.len(), 2);
358        assert_eq!(body.entries[0].seq, 2);
359        assert_eq!(body.entries[1].seq, 3);
360        let _ = std::fs::remove_dir_all(&dir);
361    }
362
363    #[test]
364    fn local_file_journal_trait() {
365        let dir = std::env::temp_dir().join(format!("lfj-{}", std::process::id()));
366        let _ = std::fs::remove_dir_all(&dir);
367        let path = dir.join("j");
368        let fj = FileJournal::open(&path).unwrap();
369        fj.append(1, b"p").unwrap();
370        let local = LocalFileJournal::new("node-a", fj);
371        let got = NodeJournal::replay_after(&local, "node-a", 1, 0, 10).unwrap();
372        assert_eq!(got[0].payload, b"p");
373        assert!(NodeJournal::replay_after(&local, "other", 1, 0, 10).is_err());
374        let _ = std::fs::remove_dir_all(&dir);
375    }
376}
hatchery_core/journal.rs

hatchery_core/
journal.rs
