storage/jsonl/ledger.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 types: the struct, its reports, and segment grammar.
7
8use std::path::{Path, PathBuf};
9
10use kernel::{B3Hash, EventRecord, Seq};
11
12use crate::error::{StorageError, io_err};
13use crate::vfs::Vfs;
14
15/// Segment rolling threshold. Internal affair: changing it changes how
16/// files are cut, never any observable semantics (`crates/storage/Spec.lean` §14).
17pub(crate) const SEGMENT_ROLL_BYTES: u64 = 64 * 1024 * 1024;
18
19/// What open found and repaired, and what its tail recovery read and
20/// checked (`crates/storage/spec/Jsonl.lean` §8-34).
21pub struct OpenReport {
22 pub recovered: Option<TailTruncation>,
23 pub counted: crate::ProofCount,
24}
25
26pub struct TailTruncation {
27 pub dropped_bytes: u64,
28}
29
30/// The durable Ledger. See the module doc for the contract.
31pub struct JsonlLedger {
32 pub(crate) vfs: Box<dyn Vfs>,
33 pub(crate) dir: PathBuf,
34 pub(crate) seg_path: PathBuf,
35 pub(crate) seg_len: u64,
36 pub(crate) next_seq: Seq,
37 pub(crate) prev: B3Hash,
38 pub(crate) roll_bytes: u64,
39 /// Whether `seg_len`, `next_seq` and `prev` still name the end of
40 /// the segments; a failed wave leaves it broken until a reopen.
41 pub(crate) barrier: super::barrier::Barrier,
42 /// The per-room projection of what this ledger appends, laid down
43 /// after a wave is durable. `None` for a ledger opened directly in a
44 /// directory that is not a city's, which has no buildings to file
45 /// sessions under.
46 pub(crate) sessions: Option<crate::sessions::Sessions>,
47 /// The write-path observer, called only after the wave is durable.
48 pub(crate) observer: Option<WriteObserver>,
49 /// This process's hold on the city's writer lock. `None` for a
50 /// directory that is not a city's ledger, and on the fault model,
51 /// whose disk no other process can reach.
52 pub(crate) lock: Option<WriterLock>,
53 /// The stop a failed whole-chain audit trips (`crates/storage/spec/Worktree/Back.lean` §8-27).
54 pub(crate) halt: crate::chain_audit::ChainHalt,
55 /// The segments a failed wave created, while restoring the disk to
56 /// its length before that wave is still owed (`jsonl::unwind`).
57 pub(crate) pending_unwind: Option<Vec<PathBuf>>,
58}
59
60/// This process's exclusive hold on a city's Ledger, released when the
61/// ledger that took it is dropped (`crates/storage/spec/Jsonl.lean` §8-1).
62///
63/// The operating system keeps the lock on an open handle and refuses it
64/// to every other handle on the same file, in this process or in
65/// another, which is what makes "a city has one writer" hold across
66/// processes rather than only inside the one that owns the type. The
67/// file is never removed: a lock file deleted on release would let a
68/// latecomer lock a new file of the same name while the earlier holder
69/// still held the old one.
70pub(crate) struct WriterLock {
71 _held: std::fs::File,
72}
73
74impl WriterLock {
75 /// Takes the writer lock of the ledger directory `dir`: the file
76 /// `<name>.lock` beside it, named from `dir` alone so that this
77 /// module, not the city layout, owns where the lock lies
78 /// (storage D18).
79 ///
80 /// # Errors
81 /// `LedgerHeld` when another handle holds the lock; `Io` when `dir`
82 /// has no name to put a sibling beside (a root, `..`), or the lock
83 /// file cannot be made, or the lock cannot be asked for.
84 pub(crate) fn take(dir: &Path) -> Result<WriterLock, StorageError> {
85 let name = dir.file_name().ok_or_else(|| {
86 io_err("name the ledger lock", dir)(std::io::Error::new(
87 std::io::ErrorKind::InvalidInput,
88 "a ledger directory needs a name for its lock to sit beside it",
89 ))
90 })?;
91 let mut lock_name = name.to_os_string();
92 lock_name.push(".lock");
93 let path = dir.with_file_name(lock_name);
94 if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) {
95 std::fs::create_dir_all(parent)
96 .map_err(io_err("create the ledger's parent directory", parent))?;
97 }
98 let file = std::fs::OpenOptions::new()
99 .create(true)
100 .truncate(false)
101 .write(true)
102 .open(&path)
103 .map_err(io_err("open the ledger lock", &path))?;
104 match file.try_lock() {
105 Ok(()) => Ok(WriterLock { _held: file }),
106 Err(std::fs::TryLockError::WouldBlock) => Err(StorageError::LedgerHeld {
107 dir: dir.to_path_buf(),
108 }),
109 Err(std::fs::TryLockError::Error(source)) => {
110 Err(io_err("lock the ledger", &path)(source))
111 }
112 }
113 }
114}
115
116/// The write-path observer: what a live reader (the control surface's
117/// event stream) installs to hear each line once it is durable.
118pub type WriteObserver = Box<dyn FnMut(&EventRecord) + Send>;
119
120/// Chain state at the entrance of the last segment.
121pub(crate) struct TailBoundary {
122 pub(crate) prev: B3Hash,
123 pub(crate) next_seq: Seq,
124 pub(crate) prior: Option<PriorSegment>,
125}
126
127pub(crate) struct PriorSegment {
128 pub(crate) path: PathBuf,
129 pub(crate) len: u64,
130}
131
132/// The two names a segment is made of. One home for the grammar: the
133/// writer that names a segment and the readers that recognise one, or
134/// read a sequence back off one, agree because they read these.
135const SEGMENT_PREFIX: &str = "ledger-";
136const SEGMENT_SUFFIX: &str = ".jsonl";
137
138pub(crate) fn segment_file_name(first_seq: Seq) -> String {
139 format!("{SEGMENT_PREFIX}{:020}{SEGMENT_SUFFIX}", first_seq.value())
140}
141
142pub(crate) fn is_segment(path: &Path) -> bool {
143 match path.file_name().and_then(|n| n.to_str()) {
144 Some(name) => name.starts_with(SEGMENT_PREFIX) && name.ends_with(SEGMENT_SUFFIX),
145 None => false,
146 }
147}
148
149/// The first sequence a segment's name claims, for a reader that wants
150/// only the records after a sequence it already holds: it skips the
151/// segments that cannot contain them by reading the name, rather than by
152/// opening the file. `None` for a name that is not this grammar.
153///
154/// The inverse of [`segment_file_name`], and it lives beside it for the
155/// reason the prefix and suffix are constants here: one reader of a
156/// grammar, not two spellings of it.
157pub(crate) fn segment_first_seq(name: &str) -> Option<Seq> {
158 let digits = name
159 .strip_prefix(SEGMENT_PREFIX)?
160 .strip_suffix(SEGMENT_SUFFIX)?;
161 digits.parse::<u64>().ok().map(Seq::new)
162}
163
164/// The segment files of `dir`, in the ledger's own order.
165///
166/// `Vfs::list` answers files only, already sorted: zero-padded names
167/// sort lexically the way they sort numerically, so the order is the
168/// ledger's own rather than the filesystem's.
169pub(crate) fn segment_names(vfs: &dyn Vfs, dir: &Path) -> Result<Vec<String>, StorageError> {
170 let mut names = Vec::new();
171 for path in vfs.list(dir).map_err(io_err("list ledger dir", dir))? {
172 if !is_segment(&path) {
173 continue;
174 }
175 if let Some(name) = path.file_name().and_then(|n| n.to_str()) {
176 names.push(name.to_owned());
177 }
178 }
179 Ok(names)
180}
181
182/// Complete (`\n`-terminated) lines and the leftover tail bytes.
183pub(crate) fn complete_lines(bytes: &[u8]) -> (Vec<&[u8]>, usize) {
184 let mut lines = Vec::new();
185 let mut consumed = 0usize;
186 let mut start = 0usize;
187 for (index, byte) in bytes.iter().enumerate() {
188 if *byte == b'\n' {
189 if let Some(line) = bytes.get(start..index) {
190 lines.push(line);
191 }
192 start = index.saturating_add(1);
193 consumed = start;
194 }
195 }
196 (lines, consumed)
197}