storage/
verified_prefix.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//! What a strict walk of the chain proved, one record per segment, so the
7//! next proof reads and hashes those bytes instead of checking them line
8//! by line again (`crates/storage/spec/ChainAudit.lean` §8-30).
9//!
10//! A projection: a record that is missing, damaged, of another rule
11//! version, or out of step with the chain state entering its segment is
12//! not used, and that segment is checked line by line. That reusing a
13//! record gives the strict verdict is `cachedVerifyIsStrict` in
14//! `crates/storage/spec/Snapshot.lean`.
15
16use std::io;
17use std::path::{Path, PathBuf};
18
19use kernel::{B3Hash, Seq};
20
21use crate::chain_audit::ProofCount;
22use crate::error::StorageError;
23use crate::jsonl::{JsonlLedger, LineCheck};
24use crate::real_fs::RealFs;
25use crate::vfs::Vfs;
26
27/// Changed whenever the per-line check changes what it accepts within one
28/// version of this binary; a new version discards every record anyway,
29/// because [`line_check_version`] hashes the version in with this. The
30/// suffix is the digest of a fixed fixture's verdicts, which
31/// `chain_audit::tests` holds, so the rule cannot move alone.
32pub(crate) const LINE_CHECK_RULES: &str = "line-check-40e703f48b222ca7";
33
34const MAGIC: &[u8; 8] = b"SPRPRF01";
35const SUFFIX: &str = ".proof";
36const STAGED: &str = ".staged";
37
38/// The rule version a record is written and accepted under.
39pub fn line_check_version() -> u32 {
40    let rules = [env!("CARGO_PKG_VERSION"), LINE_CHECK_RULES].join("\n");
41    let [a, b, c, d, ..] = *B3Hash::digest(rules.as_bytes()).as_bytes();
42    u32::from_le_bytes([a, b, c, d])
43}
44
45/// Where a city's records live, and whether this holder may write them.
46#[derive(Debug, Clone)]
47pub struct ProofRecords {
48    dir: PathBuf,
49    may: May,
50}
51
52/// Only a handle that holds the writer lock writes records, so a
53/// read-only command never touches the disk (`crates/storage/spec/ChainAudit.lean` §8-30).
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55enum May {
56    Read,
57    ReadAndWrite,
58}
59
60impl ProofRecords {
61    /// Records read, never written: `ask`, `replay`, and any holder that
62    /// does not hold the writer lock.
63    pub fn read_only(dir: &Path) -> ProofRecords {
64        ProofRecords {
65            dir: dir.to_path_buf(),
66            may: May::Read,
67        }
68    }
69
70    /// The record of the segment named `segment`, when there is one this
71    /// build can read. Absent, unreadable and damaged all answer `None`:
72    /// the segment is then checked line by line, which is the answer this
73    /// projection gives to every doubt.
74    pub(crate) fn read(&self, segment: &str) -> Option<SegmentRecord> {
75        let path = self.path_of(segment);
76        match RealFs::new().read(&path) {
77            Ok(bytes) => SegmentRecord::decode(&bytes),
78            Err(_unreadable) => None,
79        }
80    }
81
82    /// Writes `record` for `segment`, replacing the one there; nothing
83    /// for a read-only holder.
84    ///
85    /// # Errors
86    /// `StorageError::Snapshot` naming the step that failed.
87    pub(crate) fn keep(&self, segment: &str, record: &SegmentRecord) -> Result<(), StorageError> {
88        match self.may {
89            May::Read => Ok(()),
90            May::ReadAndWrite => self.write(segment, record),
91        }
92    }
93
94    fn write(&self, segment: &str, record: &SegmentRecord) -> Result<(), StorageError> {
95        let mut vfs = RealFs::new();
96        let file = self.path_of(segment);
97        let staged = self.dir.join(format!("{segment}{SUFFIX}{STAGED}"));
98        vfs.create_dir_all(&self.dir)
99            .map_err(refused("create the proof records dir", &self.dir))?;
100        if vfs.exists(&staged) {
101            vfs.remove_file(&staged)
102                .map_err(refused("remove a staged proof record", &staged))?;
103        }
104        vfs.append(&staged, &record.encode())
105            .map_err(refused("write a staged proof record", &staged))?;
106        vfs.sync_data(&staged)
107            .map_err(refused("sync a staged proof record", &staged))?;
108        vfs.rename(&staged, &file)
109            .map_err(refused("rename a staged proof record", &file))?;
110        vfs.sync_dir(&self.dir)
111            .map_err(refused("sync the proof records dir", &self.dir))
112    }
113
114    fn path_of(&self, segment: &str) -> PathBuf {
115        self.dir.join(format!("{segment}{SUFFIX}"))
116    }
117}
118
119/// Where checking a ledger's last segment begins when the ledger is
120/// opened: the byte, the lines before it, the chain state there, and what
121/// was read and hashed to get there (`crates/storage/spec/Jsonl.lean` §8-34).
122pub(crate) struct TailStart {
123    pub(crate) from: usize,
124    pub(crate) lines: u64,
125    pub(crate) check: LineCheck,
126    pub(crate) counted: ProofCount,
127}
128
129impl TailStart {
130    /// At the segment entrance, before any bytes have been read.
131    pub(crate) fn strict(entry: LineCheck) -> TailStart {
132        TailStart {
133            from: 0,
134            lines: 0,
135            check: entry,
136            counted: ProofCount::default(),
137        }
138    }
139}
140
141impl ProofRecords {
142    /// Hashes the recorded prefix through bounded windows. A mismatch
143    /// starts a fresh strict scan; a match consumes the hashed prefix's
144    /// exit state. The caller holds the writer lock throughout both reads.
145    pub(crate) fn tail_start(
146        &self,
147        vfs: &dyn Vfs,
148        path: &Path,
149        entry: LineCheck,
150    ) -> io::Result<TailStart> {
151        let mut start = TailStart::strict(entry);
152        let Some(record) = path
153            .file_name()
154            .and_then(|name| name.to_str())
155            .and_then(|name| self.read(name))
156        else {
157            return Ok(start);
158        };
159        let Some(length) = record.prefix_length(vfs.size(path)?, line_check_version()) else {
160            return Ok(start);
161        };
162        if record.entry != entry {
163            return Ok(start);
164        }
165        let mut hasher = blake3::Hasher::new();
166        let mut offset = 0u64;
167        while offset < length {
168            let chunk = vfs.read_at(
169                path,
170                offset,
171                crate::jsonl::SCAN_WINDOW_BYTES.min(length.saturating_sub(offset)),
172            )?;
173            if chunk.is_empty() {
174                return Ok(start);
175            }
176            #[cfg(test)]
177            crate::jsonl::measure_scan(chunk.capacity(), 0);
178            hasher.update(&chunk);
179            let read = u64::try_from(chunk.len()).map_err(io::Error::other)?;
180            offset = offset
181                .checked_add(read)
182                .ok_or_else(|| io::Error::other("prefix offset overflow"))?;
183            start.counted.bytes_read = start.counted.bytes_read.saturating_add(read);
184            start.counted.bytes_hashed = start.counted.bytes_hashed.saturating_add(read);
185        }
186        if let Some((lines, exit)) = record.stands(entry, &hasher) {
187            start.from = usize::try_from(length).map_err(io::Error::other)?;
188            start.lines = lines;
189            start.check = exit;
190            start.counted.segments_by_digest = 1;
191        }
192        Ok(start)
193    }
194}
195
196impl JsonlLedger {
197    /// The records of this ledger's proofs, kept in `dir`: writable when
198    /// this handle holds the ledger's writer lock, read-only otherwise.
199    pub fn proof_records(&self, dir: &Path) -> ProofRecords {
200        ProofRecords {
201            dir: dir.to_path_buf(),
202            may: match self.lock {
203                Some(_) => May::ReadAndWrite,
204                None => May::Read,
205            },
206        }
207    }
208}
209
210/// What a strict walk proved about the first `len` bytes of one segment.
211#[derive(Debug, Clone, PartialEq, Eq)]
212pub(crate) struct SegmentRecord {
213    pub(crate) version: u32,
214    /// The length of the proved prefix: always the end of a complete line.
215    pub(crate) len: u64,
216    pub(crate) lines: u64,
217    pub(crate) entry: LineCheck,
218    pub(crate) exit: LineCheck,
219    pub(crate) digest: B3Hash,
220}
221
222impl SegmentRecord {
223    /// `MAGIC | version u32 | len u64 | lines u64 | entry | exit | digest |
224    /// blake3 of everything before it`, a chain state being its `prev`
225    /// hash and the seq it expects, little-endian.
226    fn encode(&self) -> Vec<u8> {
227        let body = [
228            MAGIC.as_slice(),
229            &self.version.to_le_bytes(),
230            &self.len.to_le_bytes(),
231            &self.lines.to_le_bytes(),
232            self.entry.prev().as_bytes(),
233            &self.entry.expected().value().to_le_bytes(),
234            self.exit.prev().as_bytes(),
235            &self.exit.expected().value().to_le_bytes(),
236            self.digest.as_bytes(),
237        ]
238        .concat();
239        let seal = B3Hash::digest(&body);
240        [body.as_slice(), seal.as_bytes()].concat()
241    }
242
243    /// The record `encode` wrote; `None` for bytes that are not one.
244    fn decode(bytes: &[u8]) -> Option<SegmentRecord> {
245        let (body, seal) = bytes.split_last_chunk::<32>()?;
246        if B3Hash::digest(body).as_bytes() != seal {
247            return None;
248        }
249        let (magic, rest) = body.split_first_chunk::<8>()?;
250        if magic != MAGIC {
251            return None;
252        }
253        let (version, rest) = rest.split_first_chunk::<4>()?;
254        let (len, rest) = rest.split_first_chunk::<8>()?;
255        let (lines, rest) = rest.split_first_chunk::<8>()?;
256        let (entry, rest) = chain_state(rest)?;
257        let (exit, rest) = chain_state(rest)?;
258        let digest = <[u8; 32]>::try_from(rest).ok()?;
259        Some(SegmentRecord {
260            version: u32::from_le_bytes(*version),
261            len: u64::from_le_bytes(*len),
262            lines: u64::from_le_bytes(*lines),
263            entry,
264            exit,
265            digest: B3Hash::from_bytes(digest),
266        })
267    }
268}
269
270/// The judgement of one record, in two halves a proof can take apart:
271/// what the segment's own bytes decide, which a wave hashes before the
272/// walk reaches the segment, and what the chain entering it decides
273/// (`crates/storage/Spec.lean` §8-30, §8-34, §8-37). Together they are `reuses` in
274/// `crates/storage/spec/Snapshot.lean`, whose `cachedVerifyIsStrict`
275/// says a record that stands gives the strict verdict.
276impl SegmentRecord {
277    /// The recorded prefix length when its rule version and the
278    /// segment's current length permit reuse, for both read strategies.
279    pub(crate) fn prefix_length(&self, total: u64, version: u32) -> Option<u64> {
280        (self.version == version && self.len <= total).then_some(self.len)
281    }
282
283    /// The first `len` bytes of `bytes` this record names, when it was
284    /// written under rule version `version` and the segment still holds
285    /// that much; `None` otherwise, and nothing is to be hashed.
286    pub(crate) fn prefix<'a>(&self, bytes: &'a [u8], version: u32) -> Option<&'a [u8]> {
287        self.prefix_length(u64::try_from(bytes.len()).ok()?, version)
288            .and_then(|len| usize::try_from(len).ok())
289            .and_then(|len| bytes.get(..len))
290    }
291
292    /// Whether this record stands in for the prefix `hashed` has hashed,
293    /// entered at `entry`: its entry and its digest must both match. The
294    /// answer is the lines the prefix held and the chain state after them.
295    pub(crate) fn stands(
296        &self,
297        entry: LineCheck,
298        hashed: &blake3::Hasher,
299    ) -> Option<(u64, LineCheck)> {
300        (self.entry == entry && B3Hash::from_bytes(*hashed.finalize().as_bytes()) == self.digest)
301            .then_some((self.lines, self.exit))
302    }
303}
304
305fn chain_state(bytes: &[u8]) -> Option<(LineCheck, &[u8])> {
306    let (prev, rest) = bytes.split_first_chunk::<32>()?;
307    let (expected, rest) = rest.split_first_chunk::<8>()?;
308    Some((
309        LineCheck::after(
310            B3Hash::from_bytes(*prev),
311            Seq::new(u64::from_le_bytes(*expected)),
312        ),
313        rest,
314    ))
315}
316
317/// The constructor of a record's I/O failure. A record is a projection
318/// like a snapshot, so it shares the snapshot's advice.
319fn refused(op: &'static str, path: &Path) -> impl FnOnce(io::Error) -> StorageError {
320    let path = path.to_path_buf();
321    move |source| StorageError::Snapshot { op, path, source }
322}
storage/verified_prefix.rs

storage/
verified_prefix.rs
