Skip to main content

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(&section);
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}