verit_core/file.rs
1//! `.verit` — the Veritate **file**: many messages in one self-contained,
2//! `mmap`-able, appendable file.
3//!
4//! Normative definition: `docs/Architecture/VERIT - File Format Specification.md`.
5//! The rationale is `docs/decisions/ADR-0002`. This module implements that
6//! document; where the two disagree, the document wins.
7//!
8//! ```text
9//! ┌─ header (32 B, written once) ───────────────────────────────────────────┐
10//! │ "VRTF" · version · required_features · optional_features │
11//! ├─ generation 1 ──────────────────────────────────────────────────────────┤
12//! │ records · schema section (VRSB) · index · footer (64 B) │
13//! ├─ generation 2 (append / removal — appended, nothing above is rewritten) ┤
14//! │ new records · schema section · index · footer (64 B) ← authoritative │
15//! └─────────────────────────────────────────────────────────────────────────┘
16//! ```
17//!
18//! Four properties fall out of that shape:
19//!
20//! - **Self-contained.** The schema section holds a schema for every id in the
21//! index, so a `.verit` file plus nothing else is readable. Records are
22//! hash-only, so each schema is stored once no matter how many records use it.
23//! - **Crash-safe without a journal.** A footer is authoritative only once fully
24//! written and CRC-valid. A crash mid-append leaves the previous footer intact,
25//! so the file reads exactly as it did before. [`FileView::open`] finds the
26//! newest valid footer by scanning back from the end.
27//! - **Snapshot reads with no locking.** Committed bytes are never rewritten, so
28//! a reader holding a footer has a stable view while a writer appends. One
29//! writer, many readers.
30//! - **Stable identity.** Every record carries a monotonic `u64` **record id**
31//! that is never reused and survives removal and compaction. A position is
32//! not an identity — it shifts under both — so anything that remembers a
33//! record across commits (a consumer checkpoint, a sync cursor) must remember
34//! its id. Because ids ascend with the index, [`FileView::find_by_id`] is a
35//! binary search over bytes already loaded, and
36//! [`FileView::records_after`] gives a tailing reader "everything since my
37//! checkpoint" for free.
38//!
39//! Two entry points, sharing one commit-image implementation so they cannot
40//! drift:
41//!
42//! - [`FileBuilder`] builds a complete generation-1 file in memory, returning
43//! `Vec<u8>`. No I/O, so it is what the golden-file corpus and the ports'
44//! conformance suites are written against.
45//! - [`FileWriter`] owns a [`std::fs::File`] and performs incremental,
46//! crash-safe commits — [`append`](FileWriter::append),
47//! [`remove_id`](FileWriter::remove_id), [`commit`](FileWriter::commit),
48//! [`compact`](FileWriter::compact).
49//!
50//! Reading stays zero-copy and dependency-free: [`FileView::open`] takes a
51//! `&[u8]`, so `mmap` the file with your platform's facility (or
52//! [`std::fs::read`] it) and pass the bytes. [`FileView::get`] returns a
53//! sub-slice you hand straight to [`Message::parse`].
54//!
55//! ```
56//! # use verit_core::{Dt, SchemaBuilder, Value};
57//! # use verit_core::file::{FileBuilder, FileView};
58//! let schema = SchemaBuilder::new()
59//! .add_struct("Point", vec![(1, "x", Dt::I32), (2, "y", Dt::I32)])
60//! .build("Point")
61//! .unwrap();
62//!
63//! let mut b = FileBuilder::new();
64//! let first = b.append(&schema, &Value::Struct(vec![(1, Value::I32(3)), (2, Value::I32(4))])).unwrap();
65//! b.append(&schema, &Value::Struct(vec![(1, Value::I32(-1)), (2, Value::I32(0))])).unwrap();
66//! let bytes = b.finish().unwrap();
67//!
68//! // From here on, pretend we know nothing but `bytes`.
69//! let f = FileView::open(&bytes).unwrap();
70//! assert_eq!(f.len(), 2);
71//! assert_eq!(f.dump_json(0).unwrap(), r#"{"x":3,"y":4}"#);
72//! // A record is found by id, not by position.
73//! assert_eq!(f.find_by_id(first), Some(0));
74//! ```
75//!
76//! ## What this file is not
77//!
78//! One file. Retention — "drop the oldest million records" — costs a full
79//! compaction here, and no layout inside a single file avoids that. The answer
80//! is **segmented files**: many `.verit` files under a naming convention, whole
81//! segments dropped. That belongs in a layer above the format, and is
82//! deliberately not built into it.
83
84use std::collections::{HashMap, HashSet};
85use std::fs::{File, OpenOptions};
86use std::io::{Read, Seek, SeekFrom, Write};
87use std::path::{Path, PathBuf};
88
89use crate::encode::{encode, SchemaMode};
90use crate::error::{Error, Result};
91use crate::hash::crc32;
92use crate::message::Message;
93use crate::registry::SchemaRegistry;
94use crate::resolve::Resolver;
95use crate::schema::Schema;
96use crate::value::Value;
97
98/// File magic: "VRTF". Deliberately not a `VRTC` version bump — a 0.1.0
99/// container is rejected here structurally rather than partially misread.
100pub const FILE_MAGIC: &[u8; 4] = b"VRTF";
101/// File format version this build reads and writes.
102pub const FILE_VERSION: u8 = 1;
103/// Fixed header length, and the alignment every region starts on.
104pub const FILE_HEADER_LEN: usize = 32;
105/// Fixed footer length — one cache line, with room reserved for one more field.
106pub const FOOTER_LEN: usize = 64;
107/// Index entry width: `u64 id`, `u64 offset`, `u64 length`, `u128 schema_id`.
108pub const INDEX_ENTRY_LEN: usize = 40;
109/// Record ids start at 1, so `0` is available as "no record".
110pub const FIRST_RECORD_ID: u64 = 1;
111/// `optional_features` bit 0 — the file carries a CRC-32 per record.
112///
113/// Optional in the strict sense of spec §3.1: a reader that does not implement
114/// it reads the file correctly and simply never checks the checksums, because
115/// they live in space it already skips. See [`FileView::verify_checksums`].
116pub const OPT_RECORD_CRC: u32 = 1;
117
118const ALIGN: u64 = 8;
119
120#[inline]
121fn align_up(x: u64) -> Result<u64> {
122 x.checked_add(ALIGN - 1)
123 .map(|v| v & !(ALIGN - 1))
124 .ok_or(Error::BadFile("offset overflow"))
125}
126
127/// One index entry: a record's identity, where it lives, and which schema
128/// interprets it.
129#[derive(Clone, Copy, Debug, PartialEq, Eq)]
130pub struct Record {
131 /// Monotonic record id. Never reused; survives removal and compaction.
132 /// **This is the record's identity** — its position is not.
133 pub id: u64,
134 /// Absolute offset of the record's first byte. Always 8-aligned.
135 pub offset: u64,
136 /// The record's length in bytes, unpadded.
137 pub length: u64,
138 /// The record's 128-bit schema id, mirroring the one in its own header.
139 pub schema_id: u128,
140}
141
142/// A parsed footer — the authoritative statement of a file's committed state.
143#[derive(Clone, Copy, Debug, PartialEq, Eq)]
144pub struct Footer {
145 /// Monotonic commit counter, starting at 1.
146 pub generation: u64,
147 pub index_offset: u64,
148 pub schema_offset: u64,
149 pub schema_len: u32,
150 pub record_count: u32,
151 /// Committed extent. Bytes at or beyond this are uncommitted debris.
152 pub file_len: u64,
153 /// The id the next appended record will take. Persisted rather than derived
154 /// from the index, so removing the highest-id record cannot cause an id to
155 /// be handed out twice.
156 pub next_record_id: u64,
157}
158
159impl Footer {
160 fn encode(&self) -> [u8; FOOTER_LEN] {
161 let mut out = [0u8; FOOTER_LEN];
162 out[0..8].copy_from_slice(&self.generation.to_le_bytes());
163 out[8..16].copy_from_slice(&self.index_offset.to_le_bytes());
164 out[16..24].copy_from_slice(&self.schema_offset.to_le_bytes());
165 out[24..28].copy_from_slice(&self.schema_len.to_le_bytes());
166 out[28..32].copy_from_slice(&self.record_count.to_le_bytes());
167 out[32..40].copy_from_slice(&self.file_len.to_le_bytes());
168 out[40..48].copy_from_slice(&self.next_record_id.to_le_bytes());
169 // out[48..56] is reserved and stays zero.
170 let crc = crc32(&out[0..56]);
171 out[56..60].copy_from_slice(&crc.to_le_bytes());
172 out[60..64].copy_from_slice(FILE_MAGIC);
173 out
174 }
175
176 /// Decode a footer candidate sitting at absolute offset `at`, accepting it
177 /// only if the trailing magic, the CRC, and the self-referential
178 /// `file_len == at + FOOTER_LEN` all agree (spec §7.2). That last check is
179 /// what pins a footer to its own position, so `VRTF` bytes occurring inside
180 /// record data cannot be mistaken for one.
181 fn decode_at(bytes: &[u8; FOOTER_LEN], at: u64) -> Option<Footer> {
182 if &bytes[60..64] != FILE_MAGIC {
183 return None;
184 }
185 if u32::from_le_bytes(bytes[56..60].try_into().ok()?) != crc32(&bytes[0..56]) {
186 return None;
187 }
188 // A nonzero reserved field means these are not v1 footer bytes. The
189 // header's version check runs first, so a genuinely newer format is
190 // refused there with a clear error rather than skipped here.
191 if u64::from_le_bytes(bytes[48..56].try_into().ok()?) != 0 {
192 return None;
193 }
194 let file_len = u64::from_le_bytes(bytes[32..40].try_into().ok()?);
195 if file_len != at.checked_add(FOOTER_LEN as u64)? {
196 return None;
197 }
198 Some(Footer {
199 generation: u64::from_le_bytes(bytes[0..8].try_into().ok()?),
200 index_offset: u64::from_le_bytes(bytes[8..16].try_into().ok()?),
201 schema_offset: u64::from_le_bytes(bytes[16..24].try_into().ok()?),
202 schema_len: u32::from_le_bytes(bytes[24..28].try_into().ok()?),
203 record_count: u32::from_le_bytes(bytes[28..32].try_into().ok()?),
204 file_len,
205 next_record_id: u64::from_le_bytes(bytes[40..48].try_into().ok()?),
206 })
207 }
208}
209
210fn header_bytes(optional_features: u32) -> [u8; FILE_HEADER_LEN] {
211 let mut out = [0u8; FILE_HEADER_LEN];
212 out[0..4].copy_from_slice(FILE_MAGIC);
213 out[4] = FILE_VERSION;
214 // required_features stays zero: nothing this version writes is mandatory
215 // for a reader to understand.
216 out[12..16].copy_from_slice(&optional_features.to_le_bytes());
217 out
218}
219
220/// Validate the 32-byte header (spec §3). Feature bits split critical from
221/// non-critical: an unknown *required* bit is fatal, an unknown *optional* bit
222/// is ignored, which is how the format grows without stranding old readers.
223fn check_header(buf: &[u8]) -> Result<()> {
224 if buf.len() < FILE_HEADER_LEN {
225 return Err(Error::Truncated);
226 }
227 if &buf[0..4] != FILE_MAGIC {
228 return Err(Error::BadFile("bad magic (not a .verit file)"));
229 }
230 if buf[4] != FILE_VERSION {
231 return Err(Error::BadFile("unsupported .verit file version"));
232 }
233 if buf[5] != 0 || u16::from_le_bytes(buf[6..8].try_into().unwrap()) != 0 {
234 return Err(Error::BadFile("nonzero reserved header field"));
235 }
236 let required = u32::from_le_bytes(buf[8..12].try_into().unwrap());
237 if required != 0 {
238 return Err(Error::UnsupportedFileFeature(required));
239 }
240 // buf[12..16] is `optional_features`: unknown bits are ignored by design.
241 if u64::from_le_bytes(buf[16..24].try_into().unwrap()) != 0
242 || u64::from_le_bytes(buf[24..32].try_into().unwrap()) != 0
243 {
244 return Err(Error::BadFile("nonzero reserved header field"));
245 }
246 Ok(())
247}
248
249/// Serialize the tail of a commit — schema section, index, footer — for a set
250/// of records whose bytes already sit below `tail_start`.
251///
252/// Shared by [`FileBuilder`] and [`FileWriter`] so the in-memory and on-disk
253/// writers cannot produce different bytes for the same logical file.
254/// `tail_start` must be 8-aligned. Returns `(schema section + index, footer)`
255/// kept separate because the commit protocol (§7.1) must durably flush the
256/// former *before* writing the latter.
257fn commit_tail(
258 index: &[Record],
259 registry: &SchemaRegistry,
260 tail_start: u64,
261 generation: u64,
262 next_record_id: u64,
263 checksums: Option<&[u32]>,
264) -> Result<(Vec<u8>, Footer)> {
265 let count = u32::try_from(index.len()).map_err(|_| Error::BadFile("too many records"))?;
266
267 // Only the schemas the live index actually references. A schema whose last
268 // record was removed drops out of the file on the next commit.
269 let live: HashSet<u128> = index.iter().map(|r| r.schema_id).collect();
270 let mut pruned = SchemaRegistry::new();
271 for id in live {
272 let schema = registry.get(id).ok_or(Error::MissingSchema(id))?;
273 pruned.register(schema.clone());
274 }
275
276 let section = pruned.to_bundle();
277 let schema_len =
278 u32::try_from(section.len()).map_err(|_| Error::BadFile("schema section too large"))?;
279
280 let schema_offset = tail_start;
281 // With per-record checksums, the array sits in the space between the schema
282 // section and the index — space a reader without the feature already skips,
283 // which is what makes the feature genuinely optional (spec §5.1).
284 let crc_bytes = match checksums {
285 Some(c) => {
286 debug_assert_eq!(c.len(), index.len());
287 (c.len() as u64) * 4
288 }
289 None => 0,
290 };
291 let index_offset = align_up(schema_offset + section.len() as u64 + crc_bytes)?;
292 let index_bytes = (count as u64)
293 .checked_mul(INDEX_ENTRY_LEN as u64)
294 .ok_or(Error::BadFile("index size overflow"))?;
295 let file_len = index_offset
296 .checked_add(index_bytes)
297 .and_then(|v| v.checked_add(FOOTER_LEN as u64))
298 .ok_or(Error::BadFile("file length overflow"))?;
299
300 let mut tail = Vec::with_capacity((file_len - tail_start) as usize);
301 tail.extend_from_slice(§ion);
302 if let Some(c) = checksums {
303 // Placed so the array ends exactly at `index_offset`, which is how a
304 // reader locates it without a new footer field.
305 tail.resize((index_offset - schema_offset) as usize - c.len() * 4, 0);
306 for crc in c {
307 tail.extend_from_slice(&crc.to_le_bytes());
308 }
309 }
310 tail.resize((index_offset - schema_offset) as usize, 0); // pad to index alignment
311 for r in index {
312 tail.extend_from_slice(&r.id.to_le_bytes());
313 tail.extend_from_slice(&r.offset.to_le_bytes());
314 tail.extend_from_slice(&r.length.to_le_bytes());
315 tail.extend_from_slice(&r.schema_id.to_le_bytes());
316 }
317
318 Ok((
319 tail,
320 Footer {
321 generation,
322 index_offset,
323 schema_offset,
324 schema_len,
325 record_count: count,
326 file_len,
327 next_record_id,
328 },
329 ))
330}
331
332/// Pull the writer schema out of an inline-schema message, or fail if it is
333/// hash-only (in which case the caller must supply the schema).
334fn schema_of(bytes: &[u8]) -> Result<Schema> {
335 Message::parse(bytes)?
336 .writer_schema()?
337 .ok_or(Error::NoInlineSchema)
338}
339
340// ---------------------------------------------------------------------------
341// Reading
342// ---------------------------------------------------------------------------
343
344/// A read-only, zero-copy view over a `.verit` file's bytes (typically an
345/// `mmap`). [`open`](FileView::open) validates the header, footer, schema
346/// section, and **every** index entry up front, so each later
347/// [`get`](FileView::get) is a bounds-free slice.
348#[derive(Clone, Debug)]
349pub struct FileView<'a> {
350 buf: &'a [u8],
351 footer: Footer,
352 registry: SchemaRegistry,
353}
354
355impl<'a> FileView<'a> {
356 /// Open a file image, recovering the newest valid commit.
357 ///
358 /// Follows spec §7.2: scan back from the end over 8-aligned offsets for a
359 /// footer whose trailing magic, CRC, and self-referential length all agree.
360 /// The first hit is the newest generation, so an intact file resolves on the
361 /// first candidate and a file torn mid-append transparently opens at the
362 /// previous generation — the crash-safety guarantee, exercised as a read.
363 pub fn open(buf: &'a [u8]) -> Result<FileView<'a>> {
364 check_header(buf)?;
365 if buf.len() < FILE_HEADER_LEN + FOOTER_LEN {
366 return Err(Error::Truncated);
367 }
368
369 let footer = Self::find_footer(buf)?;
370
371 // Region geometry. The index must end exactly where the footer begins:
372 // no slack is tolerated, which removes a class of ambiguous files.
373 if footer.file_len > buf.len() as u64 {
374 return Err(Error::BadFile(
375 "footer claims more bytes than the image has",
376 ));
377 }
378 if footer.schema_offset % ALIGN != 0 || footer.index_offset % ALIGN != 0 {
379 return Err(Error::BadFile("misaligned schema or index offset"));
380 }
381 if footer.schema_offset < FILE_HEADER_LEN as u64 {
382 return Err(Error::BadFile("schema section overlaps the header"));
383 }
384 let schema_end = footer
385 .schema_offset
386 .checked_add(footer.schema_len as u64)
387 .ok_or(Error::BadFile("schema section extent overflow"))?;
388 if schema_end > footer.index_offset {
389 return Err(Error::BadFile("schema section overlaps the index"));
390 }
391 // With per-record checksums, the array occupies the 4 × record_count
392 // bytes ending at `index_offset`; it must not run back into the schema
393 // section.
394 if u32::from_le_bytes(buf[12..16].try_into().unwrap()) & OPT_RECORD_CRC != 0 {
395 let crc_bytes = (footer.record_count as u64)
396 .checked_mul(4)
397 .ok_or(Error::BadFile("checksum array size overflow"))?;
398 if footer.index_offset < crc_bytes || footer.index_offset - crc_bytes < schema_end {
399 return Err(Error::BadFile("checksum array overlaps the schema section"));
400 }
401 }
402 // `record_count` is attacker-controlled, so this is proven by
403 // arithmetic before anything is allocated or indexed (spec §10).
404 let index_bytes = (footer.record_count as u64)
405 .checked_mul(INDEX_ENTRY_LEN as u64)
406 .ok_or(Error::BadFile("index size overflow"))?;
407 let index_end = footer
408 .index_offset
409 .checked_add(index_bytes)
410 .ok_or(Error::BadFile("index extent overflow"))?;
411 if index_end != footer.file_len - FOOTER_LEN as u64 {
412 return Err(Error::BadFile("index does not end at the footer"));
413 }
414
415 let registry =
416 SchemaRegistry::from_bundle(&buf[footer.schema_offset as usize..schema_end as usize])?;
417
418 let view = FileView {
419 buf,
420 footer,
421 registry,
422 };
423
424 // Validate every entry once, so `get` can never go out of bounds, the
425 // self-containment rule (§5) is proven before any record is served, and
426 // the strictly-ascending id invariant that `find_by_id`'s binary search
427 // relies on is established rather than assumed.
428 let mut prev_id = 0u64;
429 for i in 0..view.len() {
430 let r = view.record(i)?;
431 if r.id == 0 {
432 return Err(Error::BadFile("record id 0 is reserved"));
433 }
434 if r.id <= prev_id {
435 return Err(Error::BadFile("record ids are not strictly ascending"));
436 }
437 prev_id = r.id;
438 if r.offset % ALIGN != 0 {
439 return Err(Error::BadFile("misaligned record"));
440 }
441 if r.offset < FILE_HEADER_LEN as u64 {
442 return Err(Error::BadFile("record overlaps the header"));
443 }
444 let end = r
445 .offset
446 .checked_add(r.length)
447 .ok_or(Error::BadFile("record extent overflow"))?;
448 if end > footer.schema_offset {
449 return Err(Error::BadFile("record outside the record region"));
450 }
451 if !view.registry.contains(r.schema_id) {
452 return Err(Error::MissingSchema(r.schema_id));
453 }
454 }
455 if footer.next_record_id <= prev_id {
456 return Err(Error::BadFile(
457 "next_record_id does not exceed every record id",
458 ));
459 }
460
461 Ok(view)
462 }
463
464 fn find_footer(buf: &[u8]) -> Result<Footer> {
465 // Highest 8-aligned position a footer could start at. Descending, so
466 // the first valid candidate is the newest generation.
467 let mut pos = ((buf.len() - FOOTER_LEN) as u64) & !(ALIGN - 1);
468 loop {
469 let at = pos as usize;
470 let bytes: [u8; FOOTER_LEN] = buf[at..at + FOOTER_LEN]
471 .try_into()
472 .map_err(|_| Error::Internal("footer slice width"))?;
473 if let Some(f) = Footer::decode_at(&bytes, pos) {
474 return Ok(f);
475 }
476 if pos < FILE_HEADER_LEN as u64 + ALIGN {
477 return Err(Error::NoValidFooter);
478 }
479 pos -= ALIGN;
480 }
481 }
482
483 /// Number of live records.
484 pub fn len(&self) -> usize {
485 self.footer.record_count as usize
486 }
487
488 pub fn is_empty(&self) -> bool {
489 self.footer.record_count == 0
490 }
491
492 /// The commit counter of the generation this view resolved to. A value
493 /// lower than expected after a crash means the torn commit was rolled back.
494 pub fn generation(&self) -> u64 {
495 self.footer.generation
496 }
497
498 /// The authoritative footer.
499 pub fn footer(&self) -> Footer {
500 self.footer
501 }
502
503 /// The committed extent. Bytes at or beyond this are uncommitted debris and
504 /// carry no meaning.
505 pub fn file_len(&self) -> u64 {
506 self.footer.file_len
507 }
508
509 /// The id the next appended record will take. Every live record's id is
510 /// strictly below this, and no id at or above it has ever been used.
511 pub fn next_record_id(&self) -> u64 {
512 self.footer.next_record_id
513 }
514
515 /// Whether this file carries a CRC-32 per record ([`OPT_RECORD_CRC`]).
516 pub fn has_record_checksums(&self) -> bool {
517 u32::from_le_bytes(self.buf[12..16].try_into().unwrap()) & OPT_RECORD_CRC != 0
518 }
519
520 /// The stored CRC-32 for record `i`, or `None` when the file carries none.
521 ///
522 /// The array ends exactly at `index_offset`, in the space a reader without
523 /// the feature already skips — which is what makes the feature optional
524 /// rather than a format change.
525 pub fn record_checksum(&self, i: usize) -> Option<u32> {
526 if !self.has_record_checksums() || i >= self.len() {
527 return None;
528 }
529 let base = self.footer.index_offset as usize - self.len() * 4 + i * 4;
530 Some(u32::from_le_bytes(
531 self.buf[base..base + 4].try_into().ok()?,
532 ))
533 }
534
535 /// Verify every record against its stored checksum.
536 ///
537 /// Returns the number checked — `Ok(0)` for a file that carries none, which
538 /// is not an error: checksums are optional, and their absence is a property
539 /// of the file, not a fault. A mismatch is [`Error::ChecksumMismatch`],
540 /// naming the record's **id** rather than its position.
541 pub fn verify_checksums(&self) -> Result<usize> {
542 if !self.has_record_checksums() {
543 return Ok(0);
544 }
545 for i in 0..self.len() {
546 let want = self.record_checksum(i).ok_or(Error::BadFile(
547 "checksum array is shorter than the record count",
548 ))?;
549 let got = crc32(self.get(i)?);
550 if got != want {
551 return Err(Error::ChecksumMismatch {
552 id: self.record(i)?.id,
553 expected: want,
554 found: got,
555 });
556 }
557 }
558 Ok(self.len())
559 }
560
561 /// The file's schema section, decoded. Every schema needed to read every
562 /// record is here — this is what "self-contained" means concretely.
563 pub fn schemas(&self) -> &SchemaRegistry {
564 &self.registry
565 }
566
567 /// Index entry `i`.
568 pub fn record(&self, i: usize) -> Result<Record> {
569 if i >= self.len() {
570 return Err(Error::IndexOutOfBounds);
571 }
572 let base = self.footer.index_offset as usize + i * INDEX_ENTRY_LEN;
573 Ok(Record {
574 id: u64::from_le_bytes(self.buf[base..base + 8].try_into().unwrap()),
575 offset: u64::from_le_bytes(self.buf[base + 8..base + 16].try_into().unwrap()),
576 length: u64::from_le_bytes(self.buf[base + 16..base + 24].try_into().unwrap()),
577 schema_id: u128::from_le_bytes(self.buf[base + 24..base + 40].try_into().unwrap()),
578 })
579 }
580
581 /// The raw message bytes of record `i`, borrowing the file image. Hand the
582 /// result straight to [`Message::parse`].
583 pub fn get(&self, i: usize) -> Result<&'a [u8]> {
584 let r = self.record(i)?;
585 // Validated exhaustively in `open`; this cannot go out of bounds.
586 Ok(&self.buf[r.offset as usize..(r.offset + r.length) as usize])
587 }
588
589 /// The position of the record with this id, or `None` if it is not live.
590 ///
591 /// A binary search: ids ascend with the index (appends are monotonic and
592 /// neither removal nor compaction reorders), and `open` proved it. No
593 /// secondary structure, and nothing outside the index bytes is touched.
594 pub fn find_by_id(&self, id: u64) -> Option<usize> {
595 let (mut lo, mut hi) = (0usize, self.len());
596 while lo < hi {
597 let mid = lo + (hi - lo) / 2;
598 // `mid < len`, so this entry was validated in `open`.
599 let mid_id = self.record(mid).ok()?.id;
600 match mid_id.cmp(&id) {
601 std::cmp::Ordering::Equal => return Some(mid),
602 std::cmp::Ordering::Less => lo = mid + 1,
603 std::cmp::Ordering::Greater => hi = mid,
604 }
605 }
606 None
607 }
608
609 /// The bytes of the record with this id.
610 pub fn get_by_id(&self, id: u64) -> Result<&'a [u8]> {
611 self.get(self.find_by_id(id).ok_or(Error::IndexOutOfBounds)?)
612 }
613
614 /// Every record whose id is greater than `id`, in order — a tailing
615 /// reader's "everything since my checkpoint".
616 ///
617 /// Pass the last id you processed; pass `0` for the whole file. The search
618 /// for the starting position is logarithmic, so polling a large file is
619 /// cheap even when nothing has changed.
620 pub fn records_after(&self, id: u64) -> impl Iterator<Item = Record> + '_ {
621 // First position whose id exceeds `id`.
622 let (mut lo, mut hi) = (0usize, self.len());
623 while lo < hi {
624 let mid = lo + (hi - lo) / 2;
625 match self.record(mid) {
626 Ok(r) if r.id <= id => lo = mid + 1,
627 _ => hi = mid,
628 }
629 }
630 (lo..self.len()).map(move |i| self.record(i).expect("index validated in open"))
631 }
632
633 /// The schema id of record `i`, read from the index — no record bytes are
634 /// touched, so filtering a large mapped file by type stays in one
635 /// contiguous region instead of paging the whole file in.
636 pub fn schema_id(&self, i: usize) -> Result<u128> {
637 Ok(self.record(i)?.schema_id)
638 }
639
640 /// The writer schema of record `i`, from the file's own schema section.
641 pub fn schema(&self, i: usize) -> Result<&Schema> {
642 let id = self.schema_id(i)?;
643 self.registry.get(id).ok_or(Error::MissingSchema(id))
644 }
645
646 /// Parse record `i` into a [`Message`], zero-copy over the file image.
647 pub fn message(&self, i: usize) -> Result<Message<'a>> {
648 Message::parse(self.get(i)?)
649 }
650
651 /// A [`Resolver`] reading record `i`'s writer schema into `reader` — the
652 /// schema-evolution payoff at rest. A record written years ago under an
653 /// older schema resolves into today's type, because the file kept the
654 /// writer schema alongside it.
655 ///
656 /// **Builds a resolver on every call.** Resolution is meant to be paid once
657 /// per *schema pair*, not once per record, so calling this inside a loop
658 /// over a large file repeats identical work. Use
659 /// [`resolvers`](Self::resolvers) there — it resolves each distinct schema
660 /// in the file once and hands back a lookup.
661 pub fn resolver_for(&self, i: usize, reader: &Schema) -> Result<Resolver> {
662 self.registry.resolver_for(self.schema_id(i)?, reader)
663 }
664
665 /// Resolve every schema this file's records use into `reader`, **once
666 /// each**, and return the lookup to use across the whole file.
667 ///
668 /// This is the shape the "paid once per schema pair" promise actually needs:
669 /// hoist it out of the loop, then ask it per record.
670 ///
671 /// ```
672 /// # use verit_core::{Dt, SchemaBuilder, Value};
673 /// # use verit_core::file::{FileBuilder, FileView};
674 /// # let schema = SchemaBuilder::new()
675 /// # .add_struct("P", vec![(1, "x", Dt::I32)]).build("P").unwrap();
676 /// # let mut b = FileBuilder::new();
677 /// # b.append(&schema, &Value::Struct(vec![(1, Value::I32(7))])).unwrap();
678 /// # let bytes = b.finish().unwrap();
679 /// let file = FileView::open(&bytes).unwrap();
680 /// let resolvers = file.resolvers(&schema); // once
681 /// for i in 0..file.len() {
682 /// let resolver = resolvers.for_record(&file, i).unwrap();
683 /// let root = file.message(i).unwrap().root(resolver).unwrap();
684 /// assert_eq!(root.get_i32(1).unwrap(), Some(7));
685 /// }
686 /// ```
687 ///
688 /// A file may hold records this reader cannot interpret — a mixed-schema
689 /// file read by a type that only covers one of them. Those simply do not
690 /// appear in the lookup, and [`Resolvers::for_record`] reports them as
691 /// incompatible rather than the whole call failing, so a reader can walk a
692 /// mixed file and skip what is not for it.
693 pub fn resolvers(&self, reader: &Schema) -> Resolvers {
694 let mut by_schema: HashMap<u128, Resolver> = HashMap::new();
695 for r in self.records() {
696 if by_schema.contains_key(&r.schema_id) {
697 continue;
698 }
699 if let Ok(resolver) = self.registry.resolver_for(r.schema_id, reader) {
700 by_schema.insert(r.schema_id, resolver);
701 }
702 }
703 Resolvers {
704 by_schema,
705 reader_id: reader.id(),
706 }
707 }
708
709 /// Render record `i` as JSON using only this file's bytes.
710 pub fn dump_json(&self, i: usize) -> Result<String> {
711 crate::dump::dump_json_with(self.schema(i)?, self.get(i)?)
712 }
713
714 /// Iterate over each record's bytes in order.
715 pub fn iter(&self) -> impl Iterator<Item = &'a [u8]> + '_ {
716 (0..self.len()).map(move |i| self.get(i).expect("index validated in open"))
717 }
718
719 /// Iterate over the index entries in order.
720 pub fn records(&self) -> impl Iterator<Item = Record> + '_ {
721 (0..self.len()).map(move |i| self.record(i).expect("index validated in open"))
722 }
723}
724
725/// One [`Resolver`] per distinct schema in a file, built once by
726/// [`FileView::resolvers`].
727///
728/// The point is the arithmetic: a 10,000-record file written under one schema
729/// needs **one** resolution, not 10,000. Resolution walks both schemas and
730/// compiles an access plan, so doing it per record turns an O(1) cost into an
731/// O(n) one for no benefit.
732#[derive(Clone, Debug)]
733pub struct Resolvers {
734 by_schema: HashMap<u128, Resolver>,
735 reader_id: u128,
736}
737
738impl Resolvers {
739 /// The resolver for a writer schema id, if this reader can interpret it.
740 pub fn get(&self, schema_id: u128) -> Option<&Resolver> {
741 self.by_schema.get(&schema_id)
742 }
743
744 /// The resolver for record `i` of `view`.
745 ///
746 /// Errors with [`Error::Incompatible`] when the record's schema cannot be
747 /// read into this reader — the mixed-file case, where skipping is often the
748 /// right response.
749 pub fn for_record<'a>(&self, view: &FileView<'a>, i: usize) -> Result<&Resolver> {
750 let schema_id = view.schema_id(i)?;
751 self.by_schema.get(&schema_id).ok_or_else(|| {
752 Error::Incompatible(format!(
753 "record {i} was written under schema {schema_id:#034x}, \
754 which does not resolve into reader schema {:#034x}",
755 self.reader_id
756 ))
757 })
758 }
759
760 /// How many distinct writer schemas resolved successfully.
761 pub fn len(&self) -> usize {
762 self.by_schema.len()
763 }
764
765 pub fn is_empty(&self) -> bool {
766 self.by_schema.is_empty()
767 }
768}
769
770// ---------------------------------------------------------------------------
771// Reading — owning the bytes
772// ---------------------------------------------------------------------------
773
774/// A `.verit` file read into memory, owning its bytes.
775///
776/// [`FileView`] borrows the image it reads, which means it cannot be stored in
777/// the same struct as the buffer it points into — so every caller ends up
778/// writing the same `{ bytes: Vec<u8> }` wrapper. This is that wrapper, once.
779///
780/// ```no_run
781/// # use verit_core::file::FileReader;
782/// let file = FileReader::open("events.verit")?;
783/// let view = file.view()?;
784/// for i in 0..view.len() {
785/// println!("{}", view.dump_json(i)?);
786/// }
787/// # Ok::<(), verit_core::Error>(())
788/// ```
789///
790/// For a large file, `mmap` it with your platform's facility and hand the
791/// mapped `&[u8]` to [`FileView::open`] directly — the reader never copies a
792/// record either way, and a map avoids reading bytes nothing touches.
793#[derive(Clone, Debug)]
794pub struct FileReader {
795 bytes: Vec<u8>,
796}
797
798impl FileReader {
799 /// Read and validate a file. Structural errors surface here rather than at
800 /// first access.
801 pub fn open<P: AsRef<Path>>(path: P) -> Result<FileReader> {
802 FileReader::from_bytes(std::fs::read(path)?)
803 }
804
805 /// Take ownership of an already-read image, validating it.
806 pub fn from_bytes(bytes: Vec<u8>) -> Result<FileReader> {
807 FileView::open(&bytes)?;
808 Ok(FileReader { bytes })
809 }
810
811 /// A zero-copy view over the owned bytes.
812 pub fn view(&self) -> Result<FileView<'_>> {
813 FileView::open(&self.bytes)
814 }
815
816 /// The raw image.
817 pub fn bytes(&self) -> &[u8] {
818 &self.bytes
819 }
820}
821
822// ---------------------------------------------------------------------------
823// Writing — in memory
824// ---------------------------------------------------------------------------
825
826/// Build a complete generation-1 `.verit` file in memory.
827///
828/// No I/O, deterministic output: the same records added in the same order
829/// always produce the same bytes. That makes it the reference writer the golden
830/// file corpus and the ports' conformance suites are checked against
831/// (spec §11.5). For incremental, crash-safe mutation of a file on disk, use
832/// [`FileWriter`].
833pub struct FileBuilder {
834 buf: Vec<u8>,
835 index: Vec<Record>,
836 registry: SchemaRegistry,
837 next_id: u64,
838 /// One CRC-32 per record, when [`with_record_checksums`] was called.
839 ///
840 /// [`with_record_checksums`]: FileBuilder::with_record_checksums
841 checksums: Option<Vec<u32>>,
842}
843
844impl Default for FileBuilder {
845 fn default() -> FileBuilder {
846 FileBuilder::new()
847 }
848}
849
850impl FileBuilder {
851 pub fn new() -> FileBuilder {
852 FileBuilder {
853 buf: header_bytes(0).to_vec(),
854 index: Vec::new(),
855 registry: SchemaRegistry::new(),
856 next_id: FIRST_RECORD_ID,
857 checksums: None,
858 }
859 }
860
861 /// Record a CRC-32 per record, setting [`OPT_RECORD_CRC`] in the header.
862 ///
863 /// The footer's CRC proves a *commit* was not torn; it says nothing about
864 /// the record bytes. For a file meant to be read years from now, this is
865 /// the difference between detecting bit rot and trusting it. Costs four
866 /// bytes per record, and readers without the feature are unaffected.
867 ///
868 /// Must be called before the first record.
869 pub fn with_record_checksums(mut self) -> FileBuilder {
870 debug_assert!(self.index.is_empty(), "call before appending");
871 self.buf = header_bytes(OPT_RECORD_CRC).to_vec();
872 self.checksums = Some(Vec::new());
873 self
874 }
875
876 /// Encode `value` against `schema` and append it as the next record,
877 /// returning its **record id**. The message is written **hash-only** — the
878 /// schema goes to the file's schema section once, however many records use
879 /// it.
880 pub fn append(&mut self, schema: &Schema, value: &Value) -> Result<u64> {
881 let bytes = encode(schema, value, SchemaMode::HashOnly)?;
882 self.append_message(schema, &bytes)
883 }
884
885 /// Append an already-encoded message together with its writer schema,
886 /// returning its record id. Accepts hash-only and inline-schema messages
887 /// alike; the schema id in the message must match `schema`.
888 pub fn append_message(&mut self, schema: &Schema, bytes: &[u8]) -> Result<u64> {
889 let id = self.next_id;
890 self.append_message_with_id(schema, bytes, id)
891 }
892
893 /// Append a record under an explicit id — the compaction path, where ids
894 /// must be **preserved**, not reassigned. Ids must be handed over strictly
895 /// ascending.
896 pub fn append_message_with_id(
897 &mut self,
898 schema: &Schema,
899 bytes: &[u8],
900 id: u64,
901 ) -> Result<u64> {
902 if id < self.next_id {
903 return Err(Error::BadFile("record ids must be strictly ascending"));
904 }
905 let msg = Message::parse(bytes)?;
906 if msg.schema_id() != schema.id() {
907 return Err(Error::SchemaIdMismatch {
908 message: msg.schema_id(),
909 expected: schema.id(),
910 });
911 }
912 let schema_id = self.registry.register(schema.clone());
913 self.push_bytes(bytes, schema_id, id)
914 }
915
916 /// Append a self-describing (inline-schema) message, lifting its schema out
917 /// of the message itself. The record is stored verbatim, inline schema and
918 /// all — use [`append_message`](Self::append_message) to store it hash-only.
919 pub fn append_self_describing(&mut self, bytes: &[u8]) -> Result<u64> {
920 let schema = schema_of(bytes)?;
921 let schema_id = self.registry.register(schema);
922 let id = self.next_id;
923 self.push_bytes(bytes, schema_id, id)
924 }
925
926 fn push_bytes(&mut self, bytes: &[u8], schema_id: u128, id: u64) -> Result<u64> {
927 let offset = align_up(self.buf.len() as u64)?;
928 self.buf.resize(offset as usize, 0); // 8-align the record start
929 self.buf.extend_from_slice(bytes);
930 if let Some(c) = &mut self.checksums {
931 c.push(crc32(bytes));
932 }
933 self.index.push(Record {
934 id,
935 offset,
936 length: bytes.len() as u64,
937 schema_id,
938 });
939 self.next_id = id
940 .checked_add(1)
941 .ok_or(Error::BadFile("record id space exhausted"))?;
942 Ok(id)
943 }
944
945 /// Ensure the finished file's `next_record_id` is at least `n`.
946 ///
947 /// Compaction uses this to carry a writer's id counter across a rewrite:
948 /// removing the highest-id record must not let that id be handed out again.
949 pub fn reserve_next_record_id(&mut self, n: u64) -> &mut Self {
950 self.next_id = self.next_id.max(n);
951 self
952 }
953
954 /// Number of records added so far.
955 pub fn len(&self) -> usize {
956 self.index.len()
957 }
958
959 pub fn is_empty(&self) -> bool {
960 self.index.is_empty()
961 }
962
963 /// Finish the file: append the schema section, the index, and the
964 /// generation-1 footer, and return the complete image.
965 pub fn finish(mut self) -> Result<Vec<u8>> {
966 let tail_start = align_up(self.buf.len() as u64)?;
967 self.buf.resize(tail_start as usize, 0);
968 let (tail, footer) = commit_tail(
969 &self.index,
970 &self.registry,
971 tail_start,
972 1,
973 self.next_id,
974 self.checksums.as_deref(),
975 )?;
976 self.buf.extend_from_slice(&tail);
977 self.buf.extend_from_slice(&footer.encode());
978 debug_assert_eq!(self.buf.len() as u64, footer.file_len);
979 Ok(self.buf)
980 }
981}
982
983/// Synchronise the directory holding `path`, so a rename into it is durable.
984///
985/// A rename is atomic, but the *directory entry* it creates lives in the
986/// filesystem's own metadata and is not covered by `fsync` on the file. Without
987/// this, a power loss just after `compact` can leave the old file in place —
988/// never a torn mix, but a silently undone compaction.
989///
990/// Unix only. Windows has no equivalent of opening a directory as a file, and
991/// `ReplaceFile`-style durability is not reachable from portable `std`; there
992/// this is a no-op, which is why the guarantee is stated per-platform.
993fn sync_parent_dir(path: &Path) -> Result<()> {
994 #[cfg(unix)]
995 {
996 let dir = path.parent().unwrap_or_else(|| Path::new("."));
997 // A directory opened read-only is enough to fsync it on Unix.
998 File::open(dir)?.sync_all()?;
999 }
1000 #[cfg(not(unix))]
1001 {
1002 let _ = path;
1003 }
1004 Ok(())
1005}
1006
1007/// An advisory exclusive lock on a `.verit` file, held for a writer's lifetime.
1008///
1009/// The format specifies one writer and many readers (§7.3) and deliberately has
1010/// no in-format locking scheme — that would drag it toward the complexity
1011/// ADR-0002 §3 declined. This is the documented protocol instead: a sibling
1012/// `<path>.lock` file created exclusively, removed on drop.
1013///
1014/// **Advisory, not enforced.** It stops a second [`FileWriter`] that also asks
1015/// for the lock; it cannot stop a process that writes the file directly. A
1016/// stale lock left by a killed process must be removed by hand, which is the
1017/// honest trade — silently stealing a lock after a timeout would turn a visible
1018/// operational problem into a corrupted file.
1019#[derive(Debug)]
1020struct WriterLock {
1021 path: PathBuf,
1022}
1023
1024impl WriterLock {
1025 fn acquire(target: &Path) -> Result<WriterLock> {
1026 let mut lock = target.as_os_str().to_os_string();
1027 lock.push(".lock");
1028 let path = PathBuf::from(lock);
1029 match OpenOptions::new().write(true).create_new(true).open(&path) {
1030 Ok(_) => Ok(WriterLock { path }),
1031 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
1032 Err(Error::AlreadyLocked(path.display().to_string()))
1033 }
1034 Err(e) => Err(Error::Io(e.to_string())),
1035 }
1036 }
1037}
1038
1039impl Drop for WriterLock {
1040 fn drop(&mut self) {
1041 // Best effort: a failure here leaves a stale lock, which is visible and
1042 // fixable. Panicking in a destructor would be worse.
1043 let _ = std::fs::remove_file(&self.path);
1044 }
1045}
1046
1047// ---------------------------------------------------------------------------
1048// Writing — on disk, incrementally
1049// ---------------------------------------------------------------------------
1050
1051/// Where a record's bytes are: already committed at a known offset, or staged
1052/// in memory awaiting the next commit. Its id is assigned at staging time, so a
1053/// caller can checkpoint against it before the commit lands.
1054enum Slot {
1055 Committed {
1056 id: u64,
1057 offset: u64,
1058 length: u64,
1059 schema_id: u128,
1060 /// Carried in memory so a commit never re-reads committed records to
1061 /// recompute what it already knows.
1062 crc: u32,
1063 },
1064 Staged {
1065 id: u64,
1066 staged: usize,
1067 schema_id: u128,
1068 crc: u32,
1069 },
1070}
1071
1072impl Slot {
1073 fn id(&self) -> u64 {
1074 match self {
1075 Slot::Committed { id, .. } | Slot::Staged { id, .. } => *id,
1076 }
1077 }
1078
1079 fn crc(&self) -> u32 {
1080 match self {
1081 Slot::Committed { crc, .. } | Slot::Staged { crc, .. } => *crc,
1082 }
1083 }
1084
1085 fn schema_id(&self) -> u128 {
1086 match self {
1087 Slot::Committed { schema_id, .. } | Slot::Staged { schema_id, .. } => *schema_id,
1088 }
1089 }
1090}
1091
1092/// A `.verit` file open for reading and writing, mutated by crash-safe
1093/// append-only commits.
1094///
1095/// [`append`](Self::append) and the removal methods stage changes;
1096/// [`commit`](Self::commit) makes them durable and atomic. Nothing is visible to
1097/// a reader until a commit completes, and a crash mid-commit rolls the file back
1098/// to the previous generation — so a batch of appends and removals is a
1099/// transaction.
1100///
1101/// Records are addressed by **id**, not position: `append` returns the id it
1102/// assigned, and [`remove_id`](Self::remove_id) is the safe removal. Positional
1103/// [`remove`](Self::remove) exists but renumbers everything after it.
1104///
1105/// **One writer at a time.** Concurrent writers to the same path are undefined;
1106/// mutual exclusion is the caller's job. Concurrent *readers* need no
1107/// coordination at all.
1108pub struct FileWriter {
1109 file: File,
1110 path: PathBuf,
1111 /// Held for this writer's lifetime when the caller asked to lock. `None`
1112 /// keeps the historical behaviour: unlocked, and concurrent writers are the
1113 /// caller's problem exactly as §7.3 says.
1114 lock: Option<WriterLock>,
1115 generation: u64,
1116 committed_len: u64,
1117 next_record_id: u64,
1118 slots: Vec<Slot>,
1119 staged: Vec<Vec<u8>>,
1120 registry: SchemaRegistry,
1121 /// Whether this file carries a CRC-32 per record — fixed at creation, since
1122 /// the header is written once and never modified.
1123 checksums: bool,
1124}
1125
1126impl FileWriter {
1127 /// Create a new file, truncating any existing one, and commit an empty
1128 /// generation 1 so the path is immediately a valid `.verit` file.
1129 pub fn create<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
1130 FileWriter::create_with(path, 0)
1131 }
1132
1133 /// Create a file that records a CRC-32 per record ([`OPT_RECORD_CRC`]).
1134 ///
1135 /// The footer's CRC proves a *commit* was not torn; it says nothing about
1136 /// the record bytes. For a file meant to be read years from now, this is
1137 /// the difference between detecting bit rot and trusting it. Readers
1138 /// without the feature are unaffected.
1139 pub fn create_checksummed<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
1140 FileWriter::create_with(path, OPT_RECORD_CRC)
1141 }
1142
1143 fn create_with<P: AsRef<Path>>(path: P, optional_features: u32) -> Result<FileWriter> {
1144 let path = path.as_ref().to_path_buf();
1145 let mut file = OpenOptions::new()
1146 .read(true)
1147 .write(true)
1148 .create(true)
1149 .truncate(true)
1150 .open(&path)?;
1151 file.write_all(&header_bytes(optional_features))?;
1152
1153 let mut w = FileWriter {
1154 file,
1155 path,
1156 lock: None,
1157 generation: 0,
1158 committed_len: FILE_HEADER_LEN as u64,
1159 next_record_id: FIRST_RECORD_ID,
1160 slots: Vec::new(),
1161 staged: Vec::new(),
1162 registry: SchemaRegistry::new(),
1163 checksums: optional_features & OPT_RECORD_CRC != 0,
1164 };
1165 w.commit()?;
1166 Ok(w)
1167 }
1168
1169 /// Open an existing file, recovering the newest valid commit (spec §7.2).
1170 /// Any uncommitted tail left by a crash is left in place until the next
1171 /// commit truncates it, so opening never destroys evidence.
1172 pub fn open<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
1173 let path = path.as_ref().to_path_buf();
1174 let mut file = OpenOptions::new().read(true).write(true).open(&path)?;
1175
1176 let mut image = Vec::new();
1177 file.read_to_end(&mut image)?;
1178 let view = FileView::open(&image)?;
1179
1180 let checksums = view.has_record_checksums();
1181 let slots = (0..view.len())
1182 .map(|i| {
1183 let r = view.record(i).expect("index validated in open");
1184 Slot::Committed {
1185 id: r.id,
1186 offset: r.offset,
1187 length: r.length,
1188 schema_id: r.schema_id,
1189 // Recovered from the file rather than recomputed, so
1190 // reopening never re-reads every record.
1191 crc: view.record_checksum(i).unwrap_or(0),
1192 }
1193 })
1194 .collect();
1195
1196 Ok(FileWriter {
1197 lock: None,
1198 generation: view.generation(),
1199 committed_len: view.file_len(),
1200 next_record_id: view.next_record_id(),
1201 registry: view.schemas().clone(),
1202 slots,
1203 staged: Vec::new(),
1204 checksums,
1205 file,
1206 path,
1207 })
1208 }
1209
1210 /// Open a file **and take an advisory exclusive lock** on it, refusing if
1211 /// another locking writer already holds it.
1212 ///
1213 /// The format allows one writer and many readers (§7.3), and deliberately
1214 /// has no in-format locking scheme. This is the documented protocol
1215 /// instead: a sibling `<path>.lock`, held until this writer is dropped.
1216 ///
1217 /// It is **advisory**. It stops another `FileWriter` that also locks; it
1218 /// cannot stop a process that ignores the convention. A lock left behind by
1219 /// a killed process must be removed by hand — stealing it after a timeout
1220 /// would turn a visible operational problem into a corrupted file.
1221 ///
1222 /// Readers never need this, and never block.
1223 pub fn open_locked<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
1224 let path = path.as_ref();
1225 let lock = WriterLock::acquire(path)?;
1226 let mut w = FileWriter::open(path)?;
1227 w.lock = Some(lock);
1228 Ok(w)
1229 }
1230
1231 /// [`open_locked`](Self::open_locked), creating the file if it is absent.
1232 pub fn open_or_create_locked<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
1233 let path = path.as_ref();
1234 let lock = WriterLock::acquire(path)?;
1235 let mut w = FileWriter::open_or_create(path)?;
1236 w.lock = Some(lock);
1237 Ok(w)
1238 }
1239
1240 /// Whether this writer holds the advisory lock.
1241 pub fn is_locked(&self) -> bool {
1242 self.lock.is_some()
1243 }
1244
1245 /// Open the file if it exists, otherwise create it.
1246 pub fn open_or_create<P: AsRef<Path>>(path: P) -> Result<FileWriter> {
1247 let path = path.as_ref();
1248 if path.exists() {
1249 FileWriter::open(path)
1250 } else {
1251 FileWriter::create(path)
1252 }
1253 }
1254
1255 /// Stage `value` as a new record at the end, returning the **record id** it
1256 /// was assigned. Takes effect on [`commit`](Self::commit); the id is stable
1257 /// from this moment and is what a consumer should checkpoint against.
1258 pub fn append(&mut self, schema: &Schema, value: &Value) -> Result<u64> {
1259 let bytes = encode(schema, value, SchemaMode::HashOnly)?;
1260 self.append_message(schema, &bytes)
1261 }
1262
1263 /// Stage an already-encoded message with its writer schema.
1264 pub fn append_message(&mut self, schema: &Schema, bytes: &[u8]) -> Result<u64> {
1265 let msg = Message::parse(bytes)?;
1266 if msg.schema_id() != schema.id() {
1267 return Err(Error::SchemaIdMismatch {
1268 message: msg.schema_id(),
1269 expected: schema.id(),
1270 });
1271 }
1272 let schema_id = self.registry.register(schema.clone());
1273 self.stage(bytes.to_vec(), schema_id)
1274 }
1275
1276 /// Stage a self-describing (inline-schema) message, lifting its schema out
1277 /// of the message itself.
1278 pub fn append_self_describing(&mut self, bytes: &[u8]) -> Result<u64> {
1279 let schema = schema_of(bytes)?;
1280 let schema_id = self.registry.register(schema);
1281 self.stage(bytes.to_vec(), schema_id)
1282 }
1283
1284 fn stage(&mut self, bytes: Vec<u8>, schema_id: u128) -> Result<u64> {
1285 let id = self.next_record_id;
1286 self.next_record_id = id
1287 .checked_add(1)
1288 .ok_or(Error::BadFile("record id space exhausted"))?;
1289 let crc = if self.checksums { crc32(&bytes) } else { 0 };
1290 self.staged.push(bytes);
1291 self.slots.push(Slot::Staged {
1292 id,
1293 staged: self.staged.len() - 1,
1294 schema_id,
1295 crc,
1296 });
1297 Ok(id)
1298 }
1299
1300 /// Stage the removal of the record with this id — the safe removal, since
1301 /// an id does not shift when its neighbours go away.
1302 ///
1303 /// **Removal unlinks; it does not erase** (spec §8.2). The record's bytes
1304 /// stay in the file and remain fully recoverable with a hex editor until
1305 /// [`compact`](Self::compact) rewrites it. Use [`purge_ids`](Self::purge_ids)
1306 /// when the data actually has to go.
1307 pub fn remove_id(&mut self, id: u64) -> Result<&mut Self> {
1308 let at = self
1309 .slots
1310 .iter()
1311 .position(|s| s.id() == id)
1312 .ok_or(Error::IndexOutOfBounds)?;
1313 self.slots.remove(at);
1314 Ok(self)
1315 }
1316
1317 /// Stage the removal of every listed id, returning how many were live.
1318 /// Unknown ids are ignored, so this is idempotent and safe to retry.
1319 pub fn remove_ids(&mut self, ids: &[u64]) -> usize {
1320 let doomed: HashSet<u64> = ids.iter().copied().collect();
1321 let before = self.slots.len();
1322 self.slots.retain(|s| !doomed.contains(&s.id()));
1323 before - self.slots.len()
1324 }
1325
1326 /// Stage the removal of the record at position `i`.
1327 ///
1328 /// Prefer [`remove_id`](Self::remove_id): positions shift, so removing
1329 /// record `i` renumbers everything after it, and a loop over positions is
1330 /// an off-by-one waiting to happen. Same non-erasure caveat as
1331 /// [`remove_id`](Self::remove_id).
1332 pub fn remove(&mut self, i: usize) -> Result<&mut Self> {
1333 if i >= self.slots.len() {
1334 return Err(Error::IndexOutOfBounds);
1335 }
1336 self.slots.remove(i);
1337 Ok(self)
1338 }
1339
1340 /// Keep only the records for which `keep(id, schema_id)` is true, returning
1341 /// how many were removed. The bulk removal that cannot drift by one.
1342 ///
1343 /// Deciding from a record's *contents* needs its bytes, which this does not
1344 /// hand you: read what you need with [`read_record`](Self::read_record)
1345 /// first, collect the doomed ids, then call
1346 /// [`remove_ids`](Self::remove_ids).
1347 pub fn retain<F: FnMut(u64, u128) -> bool>(&mut self, mut keep: F) -> usize {
1348 let before = self.slots.len();
1349 self.slots.retain(|s| keep(s.id(), s.schema_id()));
1350 before - self.slots.len()
1351 }
1352
1353 /// Remove every listed id **and erase it** — one commit, then one
1354 /// compaction. Returns how many records were removed.
1355 ///
1356 /// This is the call to reach for when the data genuinely has to go
1357 /// (regulated or secret content), because plain removal only unlinks.
1358 /// Batched on purpose: compaction rewrites the whole file, so doing it per
1359 /// removal would be `O(file)` each time.
1360 ///
1361 /// Erasure covers *this* file only. Backups, snapshots, and unallocated
1362 /// disk blocks from before the compaction are outside its reach.
1363 pub fn purge_ids(&mut self, ids: &[u64]) -> Result<usize> {
1364 let removed = self.remove_ids(ids);
1365 self.commit()?;
1366 self.compact()?;
1367 Ok(removed)
1368 }
1369
1370 /// Live record count, including staged-but-uncommitted changes.
1371 pub fn len(&self) -> usize {
1372 self.slots.len()
1373 }
1374
1375 pub fn is_empty(&self) -> bool {
1376 self.slots.is_empty()
1377 }
1378
1379 /// Number of staged records not yet committed.
1380 pub fn pending(&self) -> usize {
1381 self.staged.len()
1382 }
1383
1384 /// The generation of the last completed commit.
1385 pub fn generation(&self) -> u64 {
1386 self.generation
1387 }
1388
1389 /// The id the next appended record will take.
1390 pub fn next_record_id(&self) -> u64 {
1391 self.next_record_id
1392 }
1393
1394 /// The ids of every live record, in order.
1395 pub fn ids(&self) -> impl Iterator<Item = u64> + '_ {
1396 self.slots.iter().map(|s| s.id())
1397 }
1398
1399 pub fn path(&self) -> &Path {
1400 &self.path
1401 }
1402
1403 /// Make every staged change durable and atomic, per spec §7.1.
1404 ///
1405 /// Truncate the uncommitted tail, append the new records, append the schema
1406 /// section and index, **synchronise**, then append the footer and
1407 /// synchronise again. The intermediate sync is load-bearing: it is what
1408 /// guarantees that a durable footer never points at records that never
1409 /// reached the disk.
1410 pub fn commit(&mut self) -> Result<u64> {
1411 // 1. Discard any uncommitted tail from an earlier crash.
1412 self.file.set_len(self.committed_len)?;
1413 self.file.seek(SeekFrom::Start(self.committed_len))?;
1414
1415 // 2. Place staged records, 8-aligned, after the last committed byte.
1416 let mut cursor = self.committed_len;
1417 let mut body = Vec::new();
1418 let mut index = Vec::with_capacity(self.slots.len());
1419 for slot in &self.slots {
1420 match slot {
1421 Slot::Committed {
1422 id,
1423 offset,
1424 length,
1425 schema_id,
1426 ..
1427 } => index.push(Record {
1428 id: *id,
1429 offset: *offset,
1430 length: *length,
1431 schema_id: *schema_id,
1432 }),
1433 Slot::Staged {
1434 id,
1435 staged,
1436 schema_id,
1437 ..
1438 } => {
1439 let bytes = &self.staged[*staged];
1440 let offset = align_up(cursor)?;
1441 body.resize((offset - self.committed_len) as usize, 0);
1442 body.extend_from_slice(bytes);
1443 cursor = offset + bytes.len() as u64;
1444 index.push(Record {
1445 id: *id,
1446 offset,
1447 length: bytes.len() as u64,
1448 schema_id: *schema_id,
1449 });
1450 }
1451 }
1452 }
1453
1454 // 3-4. Schema section and index, both rewritten in full.
1455 let tail_start = align_up(cursor)?;
1456 body.resize((tail_start - self.committed_len) as usize, 0);
1457 let generation = self.generation + 1;
1458 let crcs: Option<Vec<u32>> = if self.checksums {
1459 Some(self.slots.iter().map(|s| s.crc()).collect())
1460 } else {
1461 None
1462 };
1463 let (tail, footer) = commit_tail(
1464 &index,
1465 &self.registry,
1466 tail_start,
1467 generation,
1468 self.next_record_id,
1469 crcs.as_deref(),
1470 )?;
1471 body.extend_from_slice(&tail);
1472
1473 self.file.write_all(&body)?;
1474 // 5. Everything the footer will point at is durable *before* the footer
1475 // exists. Skipping this is the one way this design corrupts a file.
1476 self.file.sync_all()?;
1477
1478 // 6-7. The footer is the commit point.
1479 self.file.write_all(&footer.encode())?;
1480 self.file.sync_all()?;
1481
1482 // The commit succeeded: staged records are now committed at known
1483 // offsets, and the pruned registry is the file's live schema set.
1484 let kept: Vec<u32> = self.slots.iter().map(|s| s.crc()).collect();
1485 self.slots = index
1486 .iter()
1487 .zip(kept)
1488 .map(|(r, crc)| Slot::Committed {
1489 id: r.id,
1490 offset: r.offset,
1491 length: r.length,
1492 schema_id: r.schema_id,
1493 crc,
1494 })
1495 .collect();
1496 self.staged.clear();
1497 self.generation = generation;
1498 self.committed_len = footer.file_len;
1499 Ok(generation)
1500 }
1501
1502 /// Rewrite the file with only its live records, at generation 1 — the only
1503 /// operation that reclaims space, and the only one that actually **erases**
1504 /// removed records (spec §8.3).
1505 ///
1506 /// Record **ids are preserved**, so a consumer's checkpoint stays valid
1507 /// across a compaction; positions are not. The writer's id counter is
1508 /// carried over too, so an id belonging to a purged record is never reissued.
1509 ///
1510 /// Performed out of place: a fresh file is written and synchronised, then
1511 /// atomically renamed over the original, so a crash during compaction leaves
1512 /// the original intact and readable. Staged-but-uncommitted changes are
1513 /// committed first.
1514 ///
1515 /// The rename is atomic, and on Unix the containing directory is
1516 /// synchronised afterwards so the rename itself survives a power loss. On
1517 /// other platforms that step is a no-op and the rename may be lost — the
1518 /// old file survives in that case, never a torn mix of the two.
1519 pub fn compact(&mut self) -> Result<()> {
1520 if !self.staged.is_empty() {
1521 self.commit()?;
1522 }
1523
1524 let mut builder = if self.checksums {
1525 FileBuilder::new().with_record_checksums()
1526 } else {
1527 FileBuilder::new()
1528 };
1529 for i in 0..self.slots.len() {
1530 let bytes = self.read_record(i)?;
1531 let id = self.slots[i].id();
1532 let schema_id = self.slots[i].schema_id();
1533 let schema = self
1534 .registry
1535 .get(schema_id)
1536 .ok_or(Error::MissingSchema(schema_id))?
1537 .clone();
1538 builder.append_message_with_id(&schema, &bytes, id)?;
1539 }
1540 // Carry the counter across, so a purged record's id is never reissued.
1541 builder.reserve_next_record_id(self.next_record_id);
1542 let image = builder.finish()?;
1543
1544 let mut tmp = self.path.clone().into_os_string();
1545 tmp.push(".compact");
1546 let tmp = PathBuf::from(tmp);
1547 {
1548 let mut f = OpenOptions::new()
1549 .read(true)
1550 .write(true)
1551 .create(true)
1552 .truncate(true)
1553 .open(&tmp)?;
1554 f.write_all(&image)?;
1555 f.sync_all()?;
1556 }
1557 std::fs::rename(&tmp, &self.path)?;
1558 sync_parent_dir(&self.path)?;
1559
1560 // Reopening rebuilds our state from the new file; carry the lock across
1561 // rather than dropping it mid-compaction and letting another writer in.
1562 let held = self.lock.take();
1563 *self = FileWriter::open(&self.path)?;
1564 self.lock = held;
1565 Ok(())
1566 }
1567
1568 /// Read record `i`'s bytes from disk. Committed records are read at their
1569 /// offset; staged ones come straight from memory.
1570 pub fn read_record(&mut self, i: usize) -> Result<Vec<u8>> {
1571 match self.slots.get(i).ok_or(Error::IndexOutOfBounds)? {
1572 Slot::Staged { staged, .. } => Ok(self.staged[*staged].clone()),
1573 Slot::Committed { offset, length, .. } => {
1574 let (offset, length) = (*offset, *length as usize);
1575 let mut buf = vec![0u8; length];
1576 self.file.seek(SeekFrom::Start(offset))?;
1577 self.file.read_exact(&mut buf)?;
1578 Ok(buf)
1579 }
1580 }
1581 }
1582
1583 /// Read the bytes of the record with this id.
1584 pub fn read_record_by_id(&mut self, id: u64) -> Result<Vec<u8>> {
1585 let at = self
1586 .slots
1587 .iter()
1588 .position(|s| s.id() == id)
1589 .ok_or(Error::IndexOutOfBounds)?;
1590 self.read_record(at)
1591 }
1592
1593 /// Read the whole file image back, for handing to [`FileView`].
1594 pub fn image(&mut self) -> Result<Vec<u8>> {
1595 let mut buf = Vec::new();
1596 self.file.seek(SeekFrom::Start(0))?;
1597 self.file.read_to_end(&mut buf)?;
1598 Ok(buf)
1599 }
1600}