ipfrs_storage/
write_journal.rs

1//! Storage Write Journal — append-only record of all storage mutations for
2//! crash recovery, audit, and replication purposes.
3
4// ---------------------------------------------------------------------------
5// FNV-1a 64-bit hash
6// ---------------------------------------------------------------------------
7
8/// Computes the FNV-1a 64-bit hash of `bytes`.
9pub fn fnv1a(bytes: &[u8]) -> u64 {
10    const OFFSET_BASIS: u64 = 14_695_981_039_346_656_037;
11    const PRIME: u64 = 1_099_511_628_211;
12
13    let mut hash = OFFSET_BASIS;
14    for &b in bytes {
15        hash ^= u64::from(b);
16        hash = hash.wrapping_mul(PRIME);
17    }
18    hash
19}
20
21// ---------------------------------------------------------------------------
22// JournalEntryKind
23// ---------------------------------------------------------------------------
24
25/// The kind of mutation recorded in a journal entry.
26#[derive(Clone, Debug, PartialEq, Eq)]
27pub enum JournalEntryKind {
28    /// A block was written to storage.
29    Put,
30    /// A block was removed from storage.
31    Delete,
32    /// A block was pinned (protected from GC).
33    Pin,
34    /// A block was unpinned.
35    Unpin,
36    /// A compaction event occurred (no CID; use empty string).
37    Compact,
38}
39
40// ---------------------------------------------------------------------------
41// JournalEntry
42// ---------------------------------------------------------------------------
43
44/// A single record in the write journal.
45#[derive(Clone, Debug, PartialEq, Eq)]
46pub struct JournalEntry {
47    /// Monotonically increasing sequence number, starting at 1.
48    pub sequence: u64,
49    /// The kind of storage mutation.
50    pub kind: JournalEntryKind,
51    /// Content identifier of the affected block (empty for `Compact`).
52    pub cid: String,
53    /// Size in bytes of the block; 0 for non-`Put` entries.
54    pub size_bytes: u64,
55    /// Wall-clock timestamp in seconds (Unix epoch).
56    pub timestamp_secs: u64,
57    /// FNV-1a checksum of `cid.as_bytes()` XOR `sequence`.
58    pub checksum: u64,
59}
60
61// ---------------------------------------------------------------------------
62// JournalCursor
63// ---------------------------------------------------------------------------
64
65/// Opaque cursor into the write journal.
66///
67/// Used to track the position of a consumer so that only new entries need
68/// to be delivered on subsequent calls to [`StorageWriteJournal::entries_since`].
69#[derive(Clone, Debug, PartialEq, Eq)]
70pub struct JournalCursor {
71    /// The sequence number of the last entry seen by this cursor.
72    /// `0` indicates the cursor is positioned before the very first entry.
73    pub last_sequence: u64,
74}
75
76impl JournalCursor {
77    /// Returns `true` when the cursor is positioned before any entry (i.e.,
78    /// `last_sequence == 0`).
79    pub fn is_at_start(&self) -> bool {
80        self.last_sequence == 0
81    }
82}
83
84// ---------------------------------------------------------------------------
85// JournalStats
86// ---------------------------------------------------------------------------
87
88/// Aggregate statistics derived from the current state of the journal.
89#[derive(Clone, Debug, PartialEq, Eq)]
90pub struct JournalStats {
91    /// Total number of entries currently held in the journal.
92    pub total_entries: usize,
93    /// Number of `Put` entries.
94    pub put_count: usize,
95    /// Number of `Delete` entries.
96    pub delete_count: usize,
97    /// Sum of `size_bytes` across all `Put` entries.
98    pub total_bytes_journaled: u64,
99    /// Sequence number of the oldest retained entry, or `None` if empty.
100    pub oldest_sequence: Option<u64>,
101    /// Sequence number of the newest retained entry, or `None` if empty.
102    pub newest_sequence: Option<u64>,
103}
104
105// ---------------------------------------------------------------------------
106// StorageWriteJournal
107// ---------------------------------------------------------------------------
108
109/// An append-only write journal that records all storage mutations.
110///
111/// The journal enforces a bounded capacity: when the number of stored entries
112/// exceeds `max_entries`, the oldest entry is evicted.  Sequence numbers are
113/// never reused — they continue to increment even after eviction, so consumers
114/// holding a [`JournalCursor`] can detect that they have fallen behind.
115pub struct StorageWriteJournal {
116    /// Retained entries, in ascending sequence order.  Never reordered.
117    pub entries: Vec<JournalEntry>,
118    /// The sequence number that will be assigned to the *next* appended entry.
119    pub next_sequence: u64,
120    /// Maximum number of entries to retain.  When exceeded, the oldest entry
121    /// is dropped.
122    pub max_entries: usize,
123}
124
125impl StorageWriteJournal {
126    /// Creates a new, empty journal with the given capacity limit.
127    ///
128    /// `next_sequence` is initialised to `1`.
129    pub fn new(max_entries: usize) -> Self {
130        Self {
131            entries: Vec::new(),
132            next_sequence: 1,
133            max_entries,
134        }
135    }
136
137    /// Appends a new entry to the journal.
138    ///
139    /// The checksum is computed as `fnv1a(cid.as_bytes()) ^ sequence`.
140    ///
141    /// If `entries.len() > max_entries` after appending, the oldest entry
142    /// (index 0) is removed.
143    ///
144    /// Returns the sequence number assigned to the new entry.
145    pub fn append(
146        &mut self,
147        kind: JournalEntryKind,
148        cid: &str,
149        size_bytes: u64,
150        timestamp_secs: u64,
151    ) -> u64 {
152        let sequence = self.next_sequence;
153        let checksum = fnv1a(cid.as_bytes()) ^ sequence;
154
155        let entry = JournalEntry {
156            sequence,
157            kind,
158            cid: cid.to_owned(),
159            size_bytes,
160            timestamp_secs,
161            checksum,
162        };
163
164        self.entries.push(entry);
165
166        if self.entries.len() > self.max_entries {
167            self.entries.remove(0);
168        }
169
170        self.next_sequence += 1;
171        sequence
172    }
173
174    /// Returns all entries with `sequence > cursor.last_sequence`, in order.
175    pub fn entries_since<'a>(&'a self, cursor: &JournalCursor) -> Vec<&'a JournalEntry> {
176        self.entries
177            .iter()
178            .filter(|e| e.sequence > cursor.last_sequence)
179            .collect()
180    }
181
182    /// Returns a cursor pointing at the newest retained entry.
183    ///
184    /// If the journal is empty the cursor's `last_sequence` is `0`.
185    pub fn cursor(&self) -> JournalCursor {
186        let last_sequence = self.entries.last().map(|e| e.sequence).unwrap_or(0);
187        JournalCursor { last_sequence }
188    }
189
190    /// Verifies the checksum of `entry`.
191    ///
192    /// Returns `true` when `fnv1a(entry.cid.as_bytes()) ^ entry.sequence`
193    /// equals `entry.checksum`.
194    pub fn verify_checksum(&self, entry: &JournalEntry) -> bool {
195        let expected = fnv1a(entry.cid.as_bytes()) ^ entry.sequence;
196        expected == entry.checksum
197    }
198
199    /// Returns aggregate statistics for the current journal contents.
200    pub fn stats(&self) -> JournalStats {
201        let mut put_count = 0usize;
202        let mut delete_count = 0usize;
203        let mut total_bytes_journaled = 0u64;
204
205        for entry in &self.entries {
206            match entry.kind {
207                JournalEntryKind::Put => {
208                    put_count += 1;
209                    total_bytes_journaled = total_bytes_journaled.saturating_add(entry.size_bytes);
210                }
211                JournalEntryKind::Delete => {
212                    delete_count += 1;
213                }
214                _ => {}
215            }
216        }
217
218        let oldest_sequence = self.entries.first().map(|e| e.sequence);
219        let newest_sequence = self.entries.last().map(|e| e.sequence);
220
221        JournalStats {
222            total_entries: self.entries.len(),
223            put_count,
224            delete_count,
225            total_bytes_journaled,
226            oldest_sequence,
227            newest_sequence,
228        }
229    }
230
231    /// Removes all entries whose sequence number is strictly less than
232    /// `sequence`.
233    pub fn truncate_before(&mut self, sequence: u64) {
234        self.entries.retain(|e| e.sequence >= sequence);
235    }
236}
237
238// ---------------------------------------------------------------------------
239// Tests
240// ---------------------------------------------------------------------------
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245
246    // --- fnv1a ---
247
248    #[test]
249    fn test_fnv1a_empty_is_offset_basis() {
250        assert_eq!(fnv1a(b""), 14_695_981_039_346_656_037);
251    }
252
253    #[test]
254    fn test_fnv1a_deterministic() {
255        assert_eq!(fnv1a(b"hello"), fnv1a(b"hello"));
256    }
257
258    #[test]
259    fn test_fnv1a_different_inputs_differ() {
260        assert_ne!(fnv1a(b"foo"), fnv1a(b"bar"));
261    }
262
263    // --- new() ---
264
265    #[test]
266    fn test_new_starts_empty() {
267        let journal = StorageWriteJournal::new(100);
268        assert!(journal.entries.is_empty());
269    }
270
271    #[test]
272    fn test_new_next_sequence_is_one() {
273        let journal = StorageWriteJournal::new(100);
274        assert_eq!(journal.next_sequence, 1);
275    }
276
277    // --- append() ---
278
279    #[test]
280    fn test_append_returns_correct_sequence() {
281        let mut journal = StorageWriteJournal::new(100);
282        let seq = journal.append(JournalEntryKind::Put, "bafyabc", 512, 1_000);
283        assert_eq!(seq, 1);
284    }
285
286    #[test]
287    fn test_append_sequence_monotonically_increasing() {
288        let mut journal = StorageWriteJournal::new(100);
289        let s1 = journal.append(JournalEntryKind::Put, "cid1", 100, 1);
290        let s2 = journal.append(JournalEntryKind::Delete, "cid2", 0, 2);
291        let s3 = journal.append(JournalEntryKind::Pin, "cid3", 0, 3);
292        assert!(s1 < s2 && s2 < s3);
293        assert_eq!(s1, 1);
294        assert_eq!(s2, 2);
295        assert_eq!(s3, 3);
296    }
297
298    #[test]
299    fn test_append_stores_entry_correctly() {
300        let mut journal = StorageWriteJournal::new(100);
301        journal.append(JournalEntryKind::Put, "bafytest", 1024, 999);
302        let entry = &journal.entries[0];
303        assert_eq!(entry.sequence, 1);
304        assert_eq!(entry.kind, JournalEntryKind::Put);
305        assert_eq!(entry.cid, "bafytest");
306        assert_eq!(entry.size_bytes, 1024);
307        assert_eq!(entry.timestamp_secs, 999);
308    }
309
310    #[test]
311    fn test_append_checksum_is_fnv1a_xor_sequence() {
312        let mut journal = StorageWriteJournal::new(100);
313        let seq = journal.append(JournalEntryKind::Put, "bafychecksum", 0, 0);
314        let entry = &journal.entries[0];
315        let expected = fnv1a(b"bafychecksum") ^ seq;
316        assert_eq!(entry.checksum, expected);
317    }
318
319    // --- verify_checksum() ---
320
321    #[test]
322    fn test_verify_checksum_true_for_valid_entry() {
323        let mut journal = StorageWriteJournal::new(100);
324        journal.append(JournalEntryKind::Put, "bafyvalid", 256, 42);
325        let entry = &journal.entries[0];
326        assert!(journal.verify_checksum(entry));
327    }
328
329    #[test]
330    fn test_verify_checksum_false_for_tampered_cid() {
331        let mut journal = StorageWriteJournal::new(100);
332        journal.append(JournalEntryKind::Put, "bafyoriginal", 256, 42);
333        let mut tampered = journal.entries[0].clone();
334        tampered.cid = "bafytampered".to_owned();
335        assert!(!journal.verify_checksum(&tampered));
336    }
337
338    #[test]
339    fn test_verify_checksum_false_for_tampered_checksum() {
340        let mut journal = StorageWriteJournal::new(100);
341        journal.append(JournalEntryKind::Put, "bafyoriginal2", 256, 42);
342        let mut tampered = journal.entries[0].clone();
343        tampered.checksum ^= 0xDEAD_BEEF;
344        assert!(!journal.verify_checksum(&tampered));
345    }
346
347    // --- entries_since() ---
348
349    #[test]
350    fn test_entries_since_zero_cursor_returns_all() {
351        let mut journal = StorageWriteJournal::new(100);
352        journal.append(JournalEntryKind::Put, "a", 1, 0);
353        journal.append(JournalEntryKind::Put, "b", 2, 0);
354        journal.append(JournalEntryKind::Delete, "c", 0, 0);
355        let cursor = JournalCursor { last_sequence: 0 };
356        let result = journal.entries_since(&cursor);
357        assert_eq!(result.len(), 3);
358    }
359
360    #[test]
361    fn test_entries_since_cursor_returns_only_newer() {
362        let mut journal = StorageWriteJournal::new(100);
363        journal.append(JournalEntryKind::Put, "a", 1, 0);
364        journal.append(JournalEntryKind::Put, "b", 2, 0);
365        journal.append(JournalEntryKind::Delete, "c", 0, 0);
366        let cursor = JournalCursor { last_sequence: 1 };
367        let result = journal.entries_since(&cursor);
368        assert_eq!(result.len(), 2);
369        assert_eq!(result[0].sequence, 2);
370        assert_eq!(result[1].sequence, 3);
371    }
372
373    #[test]
374    fn test_entries_since_preserves_order() {
375        let mut journal = StorageWriteJournal::new(100);
376        for i in 0..5u64 {
377            journal.append(JournalEntryKind::Put, &format!("cid{i}"), i * 10, i);
378        }
379        let cursor = JournalCursor { last_sequence: 0 };
380        let result = journal.entries_since(&cursor);
381        let seqs: Vec<u64> = result.iter().map(|e| e.sequence).collect();
382        let mut sorted = seqs.clone();
383        sorted.sort_unstable();
384        assert_eq!(seqs, sorted);
385    }
386
387    // --- cursor() ---
388
389    #[test]
390    fn test_cursor_returns_zero_when_empty() {
391        let journal = StorageWriteJournal::new(100);
392        assert_eq!(journal.cursor().last_sequence, 0);
393    }
394
395    #[test]
396    fn test_cursor_returns_newest_sequence() {
397        let mut journal = StorageWriteJournal::new(100);
398        journal.append(JournalEntryKind::Put, "x", 0, 0);
399        journal.append(JournalEntryKind::Put, "y", 0, 0);
400        journal.append(JournalEntryKind::Put, "z", 0, 0);
401        assert_eq!(journal.cursor().last_sequence, 3);
402    }
403
404    // --- is_at_start() ---
405
406    #[test]
407    fn test_is_at_start_true_when_last_sequence_zero() {
408        let cursor = JournalCursor { last_sequence: 0 };
409        assert!(cursor.is_at_start());
410    }
411
412    #[test]
413    fn test_is_at_start_false_when_last_sequence_nonzero() {
414        let cursor = JournalCursor { last_sequence: 5 };
415        assert!(!cursor.is_at_start());
416    }
417
418    // --- max_entries enforcement ---
419
420    #[test]
421    fn test_max_entries_evicts_oldest() {
422        let mut journal = StorageWriteJournal::new(3);
423        journal.append(JournalEntryKind::Put, "a", 0, 0); // seq 1
424        journal.append(JournalEntryKind::Put, "b", 0, 0); // seq 2
425        journal.append(JournalEntryKind::Put, "c", 0, 0); // seq 3
426        journal.append(JournalEntryKind::Put, "d", 0, 0); // seq 4 → evicts seq 1
427        assert_eq!(journal.entries.len(), 3);
428        assert_eq!(journal.entries[0].sequence, 2);
429        assert_eq!(journal.entries[2].sequence, 4);
430    }
431
432    #[test]
433    fn test_append_after_max_entries_has_correct_sequences() {
434        let mut journal = StorageWriteJournal::new(2);
435        for i in 0..5u64 {
436            journal.append(JournalEntryKind::Put, &format!("cid{i}"), 0, 0);
437        }
438        // Only the 2 newest entries remain, with sequences 4 and 5.
439        assert_eq!(journal.entries.len(), 2);
440        assert_eq!(journal.entries[0].sequence, 4);
441        assert_eq!(journal.entries[1].sequence, 5);
442    }
443
444    // --- truncate_before() ---
445
446    #[test]
447    fn test_truncate_before_removes_correct_entries() {
448        let mut journal = StorageWriteJournal::new(100);
449        for i in 0..5u64 {
450            journal.append(JournalEntryKind::Put, &format!("cid{i}"), 0, 0);
451        }
452        journal.truncate_before(3);
453        assert_eq!(journal.entries.len(), 3);
454        assert_eq!(journal.entries[0].sequence, 3);
455    }
456
457    // --- stats() ---
458
459    #[test]
460    fn test_stats_total_entries_correct() {
461        let mut journal = StorageWriteJournal::new(100);
462        journal.append(JournalEntryKind::Put, "a", 10, 0);
463        journal.append(JournalEntryKind::Delete, "b", 0, 0);
464        journal.append(JournalEntryKind::Pin, "c", 0, 0);
465        assert_eq!(journal.stats().total_entries, 3);
466    }
467
468    #[test]
469    fn test_stats_put_count_correct() {
470        let mut journal = StorageWriteJournal::new(100);
471        journal.append(JournalEntryKind::Put, "a", 10, 0);
472        journal.append(JournalEntryKind::Put, "b", 20, 0);
473        journal.append(JournalEntryKind::Delete, "c", 0, 0);
474        assert_eq!(journal.stats().put_count, 2);
475    }
476
477    #[test]
478    fn test_stats_delete_count_correct() {
479        let mut journal = StorageWriteJournal::new(100);
480        journal.append(JournalEntryKind::Delete, "a", 0, 0);
481        journal.append(JournalEntryKind::Put, "b", 50, 0);
482        journal.append(JournalEntryKind::Delete, "c", 0, 0);
483        assert_eq!(journal.stats().delete_count, 2);
484    }
485
486    #[test]
487    fn test_stats_total_bytes_journaled_sums_put_entries_only() {
488        let mut journal = StorageWriteJournal::new(100);
489        journal.append(JournalEntryKind::Put, "a", 100, 0);
490        journal.append(JournalEntryKind::Delete, "b", 999, 0); // size_bytes should be ignored
491        journal.append(JournalEntryKind::Put, "c", 200, 0);
492        // Only Put entries contribute → 100 + 200 = 300.
493        assert_eq!(journal.stats().total_bytes_journaled, 300);
494    }
495
496    #[test]
497    fn test_stats_oldest_newest_sequence() {
498        let mut journal = StorageWriteJournal::new(100);
499        journal.append(JournalEntryKind::Put, "a", 0, 0);
500        journal.append(JournalEntryKind::Put, "b", 0, 0);
501        journal.append(JournalEntryKind::Put, "c", 0, 0);
502        let stats = journal.stats();
503        assert_eq!(stats.oldest_sequence, Some(1));
504        assert_eq!(stats.newest_sequence, Some(3));
505    }
506
507    #[test]
508    fn test_stats_none_when_empty() {
509        let journal = StorageWriteJournal::new(100);
510        let stats = journal.stats();
511        assert_eq!(stats.oldest_sequence, None);
512        assert_eq!(stats.newest_sequence, None);
513    }
514
515    // --- JournalEntryKind::Compact ---
516
517    #[test]
518    fn test_compact_entry_appended_correctly() {
519        let mut journal = StorageWriteJournal::new(100);
520        let seq = journal.append(JournalEntryKind::Compact, "", 0, 77);
521        assert_eq!(seq, 1);
522        let entry = &journal.entries[0];
523        assert_eq!(entry.kind, JournalEntryKind::Compact);
524        assert_eq!(entry.cid, "");
525        assert_eq!(entry.size_bytes, 0);
526        assert_eq!(entry.timestamp_secs, 77);
527        // Checksum must verify correctly even for empty CID.
528        assert!(journal.verify_checksum(entry));
529    }
530}
ipfrs_storage/write_journal.rs

ipfrs_storage/
write_journal.rs
