storage/jsonl/
append.rs

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4// Copyright (c) 2026 2youg1 and the sprawling contributors
5
6//! Ledger appends: waves, reads, and the kernel Ledger face.
7
8use std::path::PathBuf;
9
10use kernel::ledger::chain_hash;
11use kernel::{
12    AxCode, AxError, EventDraft, EventKind, EventRecord, EventRef, Payload, RunId, Seq, TimeMs,
13};
14
15use crate::error::{StorageError, io_err};
16
17use super::barrier::Barrier;
18use super::ledger::{JsonlLedger, WriteObserver, complete_lines, is_segment, segment_file_name};
19
20impl JsonlLedger {
21    pub(crate) fn append_log_truncated(
22        &mut self,
23        now: TimeMs,
24        dropped: u64,
25    ) -> Result<(), StorageError> {
26        let data = Payload::of(&kernel::event::record::LogTruncated {
27            dropped_bytes: dropped,
28        })
29        .map_err(|source| StorageError::Draft { source })?;
30        let draft = EventDraft {
31            run: RunId::CITY,
32            t: now,
33            who: kernel::event::Who::City.to_string(),
34            addr: None,
35            kind: EventKind::LogTruncated,
36            data,
37            ig: false,
38        };
39        self.append_all(vec![draft]).map(|_| ())
40    }
41
42    /// Group commit: one durability barrier for the whole wave
43    /// (`crates/storage/Spec.lean` §3, item 1: the batch is what the wave delivered).
44    pub fn append_all(&mut self, drafts: Vec<EventDraft>) -> Result<Vec<EventRef>, StorageError> {
45        // A restore a failed wave left owed runs first: it is what mends
46        // the barrier, and it may remove the segment this wave would
47        // append to.
48        self.finish_unwind()?;
49        self.barrier.admit(&self.dir, self.next_seq)?;
50        self.halt.admit()?;
51        if drafts.is_empty() {
52            return Ok(Vec::new());
53        }
54        let mut seq = self.next_seq;
55        let mut prev = self.prev;
56        let mut cur_path = self.seg_path.clone();
57        let mut cur_len = self.seg_len;
58        let mut writes: Vec<(PathBuf, Vec<u8>)> = Vec::new();
59        let mut created: Vec<PathBuf> = Vec::new();
60        let mut records = Vec::with_capacity(drafts.len());
61
62        if cur_len == 0 && !self.vfs.exists(&cur_path) {
63            created.push(cur_path.clone());
64        }
65
66        for draft in drafts {
67            let record = EventRecord::from_draft(draft, seq, prev);
68            let line = record
69                .canonical_line()
70                .map_err(|source| StorageError::Draft { source })?;
71            let line_len = u64::try_from(line.len())
72                .unwrap_or(u64::MAX)
73                .saturating_add(1);
74            if cur_len > 0 && cur_len.saturating_add(line_len) > self.roll_bytes {
75                cur_path = self.dir.join(segment_file_name(seq));
76                cur_len = 0;
77                created.push(cur_path.clone());
78            }
79            prev = chain_hash(&line);
80            seq = seq
81                .next()
82                .map_err(|source| StorageError::Draft { source })?;
83            records.push(record);
84            let mut terminated = line;
85            terminated.push(b'\n');
86            match writes.last_mut() {
87                Some((path, buffer)) if *path == cur_path => buffer.extend_from_slice(&terminated),
88                _ => writes.push((cur_path.clone(), terminated)),
89            }
90            cur_len = cur_len.saturating_add(line_len);
91        }
92
93        self.barrier = Barrier::Broken;
94        self.write_wave(&writes, created)?;
95
96        self.barrier = Barrier::Whole;
97        self.seg_path = cur_path;
98        self.seg_len = cur_len;
99        self.next_seq = seq;
100        self.prev = prev;
101
102        // The per-room projection is laid down only now, after the wave
103        // is durable: the bytes it copies exist before it runs. A
104        // refusal is reported and skipped rather than returned - the
105        // history already has the record, and a disposable artifact
106        // must never fail history's caller (`crates/storage/spec/Sessions.lean` §8-24).
107        if let Some(sessions) = self.sessions.as_mut() {
108            for record in &records {
109                if let Err(error) = sessions.absorb(record) {
110                    eprintln!("a session slice was refused and skipped: {error}");
111                }
112            }
113        }
114
115        // Only now, with the bytes synced, does anyone else hear about
116        // them: an observer that saw an event the disk never got would be
117        // telling the interface something the history does not contain.
118        if let Some(observer) = self.observer.as_mut() {
119            for record in &records {
120                observer(record);
121            }
122        }
123        Ok(records.iter().map(EventRecord::to_ref).collect())
124    }
125
126    /// Hands this ledger's session slices to whoever will file them from
127    /// now on; this writer files none after it. `None` for a ledger that
128    /// is not a city's, and for one that already handed them off
129    /// (`crates/storage/spec/Sessions.lean` §8-24).
130    pub fn hand_off_session_slices(&mut self) -> Option<crate::sessions::Sessions> {
131        self.sessions.take()
132    }
133
134    /// The position a record written now would take.
135    ///
136    /// A position, never content: this is what anchors a diagnostic log
137    /// line to the only history (`docs/logging.md` section 4), and a
138    /// reader accessor that returned records would invite decision logic
139    /// to consult the ledger it is in the middle of writing.
140    #[must_use]
141    pub fn position(&self) -> Seq {
142        self.next_seq
143    }
144
145    /// Installs the write-path observer, replacing any earlier one. The
146    /// sink runs on the appending thread after durability, so it must not
147    /// block; a bounded send that drops on lag is the shape this expects.
148    pub fn observe(&mut self, sink: WriteObserver) {
149        self.observer = Some(sink);
150    }
151
152    /// The read face for replay, fixtures and inspection: every canonical
153    /// line (no terminators), segments flattened in order.
154    pub fn read_raw_lines(&self) -> Result<Vec<Vec<u8>>, StorageError> {
155        let mut out = Vec::new();
156        let segments: Vec<PathBuf> = self
157            .vfs
158            .list(&self.dir)
159            .map_err(io_err("list ledger dir", &self.dir))?
160            .into_iter()
161            .filter(|p| is_segment(p))
162            .collect();
163        for seg in segments {
164            let bytes = self.vfs.read(&seg).map_err(io_err("read segment", &seg))?;
165            let (lines, _) = complete_lines(&bytes);
166            for line in lines {
167                if !line.is_empty() {
168                    out.push(line.to_vec());
169                }
170            }
171        }
172        Ok(out)
173    }
174
175    #[cfg(test)]
176    pub(crate) fn set_roll_bytes_for_test(&mut self, roll_bytes: u64) {
177        self.roll_bytes = roll_bytes;
178    }
179}
180
181impl kernel::Ledger for JsonlLedger {
182    fn append(&mut self, draft: EventDraft) -> Result<EventRef, AxError> {
183        let refs = self
184            .append_all(vec![draft])
185            .map_err(StorageError::into_ax)?;
186        refs.into_iter().next().ok_or_else(|| {
187            AxError::failure(AxCode::InvalidArgs, "append event", "empty wave echo").with_recovery(
188                "report this against storage::jsonl::append with the city path: \
189                         append_all owes one echo per draft and returned none",
190            )
191        })
192    }
193
194    /// One buffer, one write and one barrier for the whole wave.
195    fn append_all(&mut self, drafts: Vec<EventDraft>) -> Result<Vec<EventRef>, AxError> {
196        JsonlLedger::append_all(self, drafts).map_err(StorageError::into_ax)
197    }
198}
199
200#[cfg(feature = "conformance")]
201impl kernel::ledger::conformance::LedgerInspect for JsonlLedger {
202    fn raw_lines(&self) -> Result<Vec<Vec<u8>>, AxError> {
203        self.read_raw_lines().map_err(StorageError::into_ax)
204    }
205}
206
207#[cfg(test)]
208#[allow(
209    clippy::unwrap_used,
210    clippy::expect_used,
211    clippy::panic,
212    clippy::indexing_slicing,
213    reason = "test code"
214)]
215mod tests {
216    use super::*;
217    use kernel::GENESIS_PREV;
218    use std::fs;
219    use std::path::Path;
220    fn draft(kind: EventKind, t: u64) -> EventDraft {
221        EventDraft {
222            run: RunId::CITY,
223            t: TimeMs::new(t),
224            who: "city".to_string(),
225            addr: None,
226            kind,
227            data: Payload::empty(),
228            ig: false,
229        }
230    }
231    fn verify_chain(lines: &[Vec<u8>]) {
232        let mut prev = GENESIS_PREV;
233        for (i, line) in lines.iter().enumerate() {
234            let record = EventRecord::parse_line(line).unwrap();
235            assert_eq!(record.prev(), prev, "prev broken at line {i}");
236            assert_eq!(record.seq(), Seq::new(u64::try_from(i).unwrap()));
237            prev = chain_hash(line);
238        }
239    }
240
241    #[cfg(feature = "conformance")]
242    #[test]
243    fn passes_the_kernel_conformance_suite() {
244        use kernel::ledger::conformance::assert_ledger_conformance;
245        let keep: std::cell::RefCell<Vec<tempfile::TempDir>> = std::cell::RefCell::new(Vec::new());
246        assert_ledger_conformance(|| {
247            let dir = tempfile::tempdir().unwrap();
248            let (ledger, _) = JsonlLedger::open(dir.path(), TimeMs::new(0)).unwrap();
249            keep.borrow_mut().push(dir);
250            ledger
251        });
252    }
253
254    #[test]
255    fn rolls_segments_without_breaking_seq_or_chain() {
256        let dir = tempfile::tempdir().unwrap();
257        let (mut ledger, _) = JsonlLedger::open(dir.path(), TimeMs::new(0)).unwrap();
258        ledger.set_roll_bytes_for_test(1);
259        for t in 0..4 {
260            ledger
261                .append_all(vec![draft(EventKind::GateChecked, t)])
262                .unwrap();
263        }
264        let segments = fs::read_dir(dir.path()).unwrap().count();
265        assert!(segments >= 4, "tiny roll budget must produce many segments");
266        let lines = ledger.read_raw_lines().unwrap();
267        assert_eq!(lines.len(), 4);
268        verify_chain(&lines);
269
270        drop(ledger);
271        let (mut reopened, report) = JsonlLedger::open(dir.path(), TimeMs::new(9)).unwrap();
272        assert!(report.recovered.is_none());
273        reopened
274            .append_all(vec![draft(EventKind::RunFrozen, 9)])
275            .unwrap();
276        verify_chain(&reopened.read_raw_lines().unwrap());
277    }
278
279    /// What one production append costs the disk, counted rather than
280    /// timed: `kernel::Ledger::append` is a batch of one, so every
281    /// record pays a whole durability barrier. The op count is the
282    /// budget reading; the wall clock belongs to the machine.
283    #[test]
284    fn one_append_costs_a_fixed_number_of_disk_operations() {
285        let dir = tempfile::tempdir().unwrap();
286        let fs = crate::fault_fs::FaultFs::new(crate::fault_fs::FaultPlan {
287            cut_at_op: None,
288            cut_on_write: None,
289            torn_tail: crate::fault_fs::TornTail::None,
290        });
291        let (mut ledger, _) =
292            JsonlLedger::open_faulty(fs.clone(), dir.path(), TimeMs::new(0)).unwrap();
293        let opened = fs.op_count();
294        kernel::Ledger::append(&mut ledger, draft(EventKind::GateChecked, 1)).unwrap();
295        let first = fs.op_count().saturating_sub(opened);
296        kernel::Ledger::append(&mut ledger, draft(EventKind::GateChecked, 2)).unwrap();
297        let second = fs.op_count().saturating_sub(opened).saturating_sub(first);
298        let bytes: u64 = ledger
299            .read_raw_lines()
300            .unwrap()
301            .iter()
302            .map(|line| u64::try_from(line.len().saturating_add(1)).unwrap())
303            .sum();
304        eprintln!(
305            "ledger_append_ops: {first} disk ops for the first record, {second} for \n             every one after it, {bytes} B on disk for two records"
306        );
307        assert!(
308            first > second,
309            "the first record also creates the segment and syncs its directory entry"
310        );
311    }
312
313    fn only_segment(dir: &Path) -> std::path::PathBuf {
314        let mut files: Vec<_> = fs::read_dir(dir)
315            .unwrap()
316            .map(|e| e.unwrap().path())
317            .collect();
318        files.sort();
319        assert_eq!(files.len(), 1);
320        files.remove(0)
321    }
322
323    proptest::proptest! {
324        #[test]
325        fn any_tail_damage_recovers_to_a_valid_chain(
326            n in 1usize..5,
327            cut_back in 1usize..40,
328            garbage in proptest::collection::vec(1u8..=255, 0..24),
329        ) {
330            let dir = tempfile::tempdir().unwrap();
331            let (mut ledger, _) = JsonlLedger::open(dir.path(), TimeMs::new(0)).unwrap();
332            let drafts: Vec<_> = (0..n)
333                .map(|i| draft(EventKind::GateChecked, u64::try_from(i).unwrap()))
334                .collect();
335            ledger.append_all(drafts).unwrap();
336            let baseline = ledger.read_raw_lines().unwrap();
337            let seg = only_segment(dir.path());
338            drop(ledger);
339
340            let mut bytes = fs::read(&seg).unwrap();
341            let keep = bytes.len().saturating_sub(cut_back);
342            bytes.truncate(keep);
343            bytes.extend_from_slice(&garbage);
344            fs::write(&seg, &bytes).unwrap();
345
346            let (reopened, _) = JsonlLedger::open(dir.path(), TimeMs::new(999)).unwrap();
347            let lines = reopened.read_raw_lines().unwrap();
348            verify_chain(&lines);
349            // Every surviving pre-damage line is a byte-exact prefix entry.
350            let survivors = lines
351                .iter()
352                .filter(|l| {
353                    EventRecord::parse_line(l).unwrap().kind() != EventKind::LogTruncated
354                })
355                .count();
356            proptest::prop_assert!(survivors <= baseline.len());
357            for (mine, original) in lines.iter().take(survivors).zip(baseline.iter()) {
358                proptest::prop_assert_eq!(mine, original);
359            }
360        }
361    }
362}
storage/jsonl/append.rs

storage/jsonl/
append.rs
