storage/
chain_audit.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//! The whole chain proved from genesis, one segment after another,
7//! reusing what an earlier strict walk proved, the segments read and
8//! their recorded prefixes hashed in waves side by side; and the verdict
9//! the writer waits for or halts on (`crates/storage/spec/ChainAudit.lean` §8-30, §8-37). The proof
10//! is a query; setting the verdict is the caller's command.
11
12use std::path::Path;
13use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError};
14
15use kernel::{AxError, B3Hash};
16
17use crate::error::StorageError;
18use crate::jsonl::{JsonlLedger, LineCheck, segment_names};
19use crate::real_fs::RealFs;
20use crate::verified_prefix::{ProofRecords, SegmentRecord, line_check_version};
21use wave::{Hashed, PROOF_WAVE, Read, read_wave};
22
23mod wave;
24
25/// What walking the whole chain found.
26#[derive(Debug, Clone, PartialEq)]
27pub enum ChainAudit {
28    /// Every complete line continues the chain.
29    Whole { lines: u64 },
30    /// The first line that did not, as the reason a person reads.
31    Broken(AxError),
32}
33
34/// What one proof read and computed: counts rather than time, so a test
35/// can hold them at two sizes of history.
36#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
37pub struct ProofCount {
38    /// Lines checked one by one through `LineCheck`.
39    pub lines_checked: u64,
40    /// Segments whose record stood in for checking their prefix.
41    pub segments_by_digest: u64,
42    pub bytes_read: u64,
43    pub bytes_hashed: u64,
44    /// Waves the segments were read in, each wave read side by side
45    /// before the chain is walked through it (`crates/storage/spec/ChainAudit.lean` §8-37).
46    pub waves: u64,
47}
48
49/// A proof's verdict, what it cost, and the first record it could not
50/// write, which changes no verdict.
51#[derive(Debug, Clone, PartialEq)]
52pub struct Proven {
53    pub audit: ChainAudit,
54    pub counted: ProofCount,
55    pub unkept: Option<AxError>,
56}
57
58/// The verdict a writer shares with the proof of its history: once set it
59/// stays, and it carries the reason every refused write repeats.
60#[derive(Debug, Clone, Default)]
61pub struct ChainHalt {
62    verdict: Arc<(Mutex<Option<Verdict>>, Condvar)>,
63    before: Before,
64}
65
66#[derive(Debug, Clone)]
67enum Verdict {
68    Whole,
69    Broken(AxError),
70}
71
72/// What the writer does before there is a verdict.
73#[derive(Debug, Clone, Copy, Default)]
74enum Before {
75    /// The history was proved when the writer opened (`resume`, `fork`,
76    /// a one-shot command), so nothing waits for a background proof.
77    #[default]
78    Admit,
79    /// A served city opened from its snapshots: nothing is written after
80    /// history nobody has walked yet.
81    Refuse,
82}
83
84impl ChainHalt {
85    /// A halt that refuses every append until a verdict is set.
86    pub fn awaiting_proof() -> ChainHalt {
87        ChainHalt {
88            verdict: Arc::default(),
89            before: Before::Refuse,
90        }
91    }
92
93    /// The chain is whole. The first verdict wins.
94    pub fn prove(&self) {
95        self.settle(Verdict::Whole);
96    }
97
98    /// Stop taking new work. The first verdict wins; a later reason is
99    /// dropped, because the chain broke where the first proof said.
100    pub fn trip(&self, reason: AxError) {
101        self.settle(Verdict::Broken(reason));
102    }
103
104    /// Why new work is refused, when the chain is broken.
105    pub fn reason(&self) -> Option<AxError> {
106        match &*self.held() {
107            Some(Verdict::Broken(reason)) => Some(reason.clone()),
108            Some(Verdict::Whole) | None => None,
109        }
110    }
111
112    /// Whether a proof found the chain whole.
113    pub fn proved(&self) -> bool {
114        matches!(&*self.held(), Some(Verdict::Whole))
115    }
116
117    /// Blocks until there is a verdict; returns at once for a halt that
118    /// admits before one.
119    pub fn await_verdict(&self) {
120        match self.before {
121            Before::Admit => {}
122            Before::Refuse => {
123                let (lock, settled) = &*self.verdict;
124                let guard = lock.lock().unwrap_or_else(PoisonError::into_inner);
125                drop(
126                    settled
127                        .wait_while(guard, |verdict| verdict.is_none())
128                        .unwrap_or_else(PoisonError::into_inner),
129                );
130            }
131        }
132    }
133
134    pub(crate) fn admit(&self) -> Result<(), StorageError> {
135        match (&*self.held(), self.before) {
136            (Some(Verdict::Whole), _) | (None, Before::Admit) => Ok(()),
137            (Some(Verdict::Broken(reason)), _) => Err(StorageError::ChainHalted {
138                source: reason.clone(),
139            }),
140            (None, Before::Refuse) => Err(StorageError::Unproven),
141        }
142    }
143
144    fn settle(&self, verdict: Verdict) {
145        let (lock, settled) = &*self.verdict;
146        let mut held = lock.lock().unwrap_or_else(PoisonError::into_inner);
147        if held.is_none() {
148            *held = Some(verdict);
149            settled.notify_all();
150        }
151    }
152
153    /// The verdict as it stands. A panic while it was held cannot leave it
154    /// half-written: it is one assignment of a whole value.
155    fn held(&self) -> MutexGuard<'_, Option<Verdict>> {
156        self.verdict
157            .0
158            .lock()
159            .unwrap_or_else(PoisonError::into_inner)
160    }
161}
162
163impl JsonlLedger {
164    /// Share `halt` with this writer: every later `append_all` asks it
165    /// before framing a line.
166    pub fn halt_on(&mut self, halt: ChainHalt) {
167        self.halt = halt;
168    }
169
170    /// Blocks until the proof this writer waits for has a verdict.
171    pub fn await_verdict(&self) {
172        self.halt.await_verdict();
173    }
174
175    /// Whether this writer takes a line now, asked before a caller does
176    /// anything a refused line would leave behind.
177    ///
178    /// # Errors
179    /// `Unproven` before the verdict of a proof this writer awaits;
180    /// `ChainHalted` after a broken one.
181    pub fn admits(&self) -> Result<(), StorageError> {
182        self.halt.admit()
183    }
184}
185
186/// Walk every complete line of the ledger in `dir` from genesis, each
187/// through the one per-line check, reading no record and writing none.
188/// Read-only; holds one segment at a time.
189///
190/// # Errors
191/// `StorageError::Io` when a segment cannot be listed or read.
192pub fn audit_chain(dir: &Path) -> Result<ChainAudit, StorageError> {
193    walk(dir, None).map(|proven| proven.audit)
194}
195
196/// Prove the ledger in `dir` from genesis, taking a segment's record in
197/// place of checking its prefix line by line when the record's rule
198/// version, entry state and digest all match, and writing a record for
199/// every segment proved when `records` may be written.
200///
201/// # Errors
202/// `StorageError::Io` when a segment cannot be listed or read.
203pub fn prove_chain(dir: &Path, records: &ProofRecords) -> Result<Proven, StorageError> {
204    walk(dir, Some(records))
205}
206
207/// The one walk behind both faces: each wave is read whole before the
208/// chain is walked through its segments in order.
209fn walk(dir: &Path, records: Option<&ProofRecords>) -> Result<Proven, StorageError> {
210    let version = line_check_version();
211    let mut walked = Walked {
212        check: LineCheck::at_genesis(),
213        lines: 0,
214        counted: ProofCount::default(),
215        unkept: None,
216    };
217    for wave in segment_names(&RealFs::new(), dir)?.chunks(PROOF_WAVE) {
218        walked.counted.waves = walked.counted.waves.saturating_add(1);
219        for (name, read) in wave.iter().zip(read_wave(dir, wave, records, version)?) {
220            let proved = match walked.segment(read, version) {
221                Ok(proved) => proved,
222                Err(broken) => {
223                    return Ok(Proven {
224                        audit: ChainAudit::Broken(broken),
225                        counted: walked.counted,
226                        unkept: walked.unkept,
227                    });
228                }
229            };
230            if let Some(records) = records {
231                let kept = records.keep(name, &proved).map_err(StorageError::into_ax);
232                walked.unkept = walked.unkept.take().or(kept.err());
233            }
234        }
235    }
236    Ok(Proven {
237        audit: ChainAudit::Whole {
238            lines: walked.lines,
239        },
240        counted: walked.counted,
241        unkept: walked.unkept,
242    })
243}
244
245/// The chain state and the counts, between two segments.
246struct Walked {
247    check: LineCheck,
248    lines: u64,
249    counted: ProofCount,
250    unkept: Option<AxError>,
251}
252
253impl Walked {
254    /// Prove one segment's complete lines and answer the record of what
255    /// was proved. The bytes checked and the bytes hashed are the same
256    /// read (`crates/storage/spec/ChainAudit.lean` §8-30).
257    fn segment(&mut self, read: Read, version: u32) -> Result<SegmentRecord, AxError> {
258        let Read { bytes, hashed } = read;
259        self.counted.bytes_read = self.counted.bytes_read.saturating_add(len_of(&bytes));
260        let entry = self.check;
261        let complete = bytes
262            .iter()
263            .rposition(|byte| *byte == b'\n')
264            .map_or(0, |at| at.saturating_add(1));
265        let (from, mut in_segment, mut hasher) = hashed
266            .and_then(|hashed| self.reuse(hashed))
267            .unwrap_or_else(|| (0, 0, blake3::Hasher::new()));
268        let rest = bytes.get(from..complete).unwrap_or_default();
269        hasher.update(rest);
270        self.counted.bytes_hashed = self.counted.bytes_hashed.saturating_add(len_of(rest));
271        for line in rest.split(|byte| *byte == b'\n') {
272            if line.is_empty() {
273                continue;
274            }
275            self.lines = self.lines.saturating_add(1);
276            in_segment = in_segment.saturating_add(1);
277            self.counted.lines_checked = self.counted.lines_checked.saturating_add(1);
278            self.check
279                .advance(line)
280                .map_err(|fault| fault.into_ax(self.lines))?;
281        }
282        Ok(SegmentRecord {
283            version,
284            len: len_of(bytes.get(..complete).unwrap_or_default()),
285            lines: in_segment,
286            entry,
287            exit: self.check,
288            digest: B3Hash::from_bytes(*hasher.finalize().as_bytes()),
289        })
290    }
291
292    /// Takes the record in place of checking the prefix it names when it
293    /// stands for it (`SegmentRecord::stands`), answering where checking
294    /// goes on, how many lines the prefix held, and the hasher that has
295    /// hashed it. `None` leaves the chain state where it was; the prefix
296    /// was hashed either way, and is counted.
297    fn reuse(&mut self, hashed: Hashed) -> Option<(usize, u64, blake3::Hasher)> {
298        let Hashed {
299            record,
300            len,
301            hasher,
302        } = hashed;
303        self.counted.bytes_hashed = self
304            .counted
305            .bytes_hashed
306            .saturating_add(u64::try_from(len).unwrap_or(u64::MAX));
307        let (lines, exit) = record.stands(self.check, &hasher)?;
308        self.lines = self.lines.saturating_add(lines);
309        self.check = exit;
310        self.counted.segments_by_digest = self.counted.segments_by_digest.saturating_add(1);
311        Some((len, lines, hasher))
312    }
313}
314
315fn len_of(bytes: &[u8]) -> u64 {
316    u64::try_from(bytes.len()).unwrap_or(u64::MAX)
317}
318
319#[cfg(test)]
320mod tests;
storage/chain_audit.rs

storage/
chain_audit.rs
