plugins/native/znippy-plugin-git/src/archive_write.rs
//! The push path's **archive writer**: one trait, three durability contracts.
//!
//! A git client pushes a packfile. Its bytes are already zlib-deflated and often
//! delta-encoded, so they go into the znippy Apache Arrow IPC archive
//! **verbatim** — never re-compressed, never re-encoded (the reason is measured
//! in this crate's own module docs: re-inflating per oid cost 3.4× the input on
//! 60 000 objects). What *does* vary is how much durability the server buys
//! before it sends the client its ack, and that is the only axis this module
//! exposes.
//!
//! The index tables are **not** built here. [`append`](ArchiveWrite::append)
//! returns a byte extent; the extent goes over a channel to the per-account
//! indexer in [`crate::indexer`], which builds the Arrow index tables *after*
//! the bytes are down. See that module for why a read arriving before the index
//! is ready still cannot be wrong.
//!
//! # The three arms, and what each one actually promises
//!
//! | impl | on return, the bytes are… | a crash right after return |
//! |---|---|---|
//! | [`FastWriter`] | in the **page cache** | **loses them** |
//! | [`SafeWriter`] | on the platter, and a journal row points at them | keeps them |
//! | [`UringWriter`] | on the platter, and a journal row points at them | keeps them |
//!
//! [`FastWriter`] **bounds** the other two: it is the ceiling that the cost of
//! durability is measured against. It is also a **selectable arm** — see
//! [`crate::arms::WriterArm`] — because there are workloads whose contract is
//! not git's (a rebuildable mirror, a bulk import that is re-run on failure, a
//! benchmark), and the operator who picks it is choosing the row above. Nothing
//! here refuses it; the table is what it promises.
//!
//! # The ordering the two durable arms both obey
//!
//! Taken from `znippy-common/src/hot.rs` (:92, :310-327) and not re-invented:
//!
//! ```text
//! blob bytes -> fsync(blobs) -> journal row -> fsync(journal)
//! ```
//!
//! The blob bytes are durable **before** any row references them. A crash
//! between the two leaves **orphan bytes nobody points at** — dead payload the
//! seal drops. The reverse order would leave an index row pointing into a hole,
//! which is a corrupt archive rather than a lost append.
//!
//! # The BufWriter trap
//!
//! The journal is written through a `std::io::BufWriter`. `flush()` moves
//! userspace → kernel; `sync_all()` moves kernel → platter. **`sync_all()`
//! without a prior `flush()` syncs nothing you just wrote and looks perfectly
//! durable** — the call succeeds, the fsync is real, and the bytes are still
//! sitting in a userspace `Vec`. [`SafeWriter`] deliberately keeps the userspace
//! buffer so that this ordering is load-bearing and a test can see it fail; the
//! guard `journal_flush_before_sync_is_load_bearing` in `tests/archive_write.rs`
//! is the one that watched it.
//!
//! # The journal is a LOG, and a reopen appends to it
//!
//! One Arrow IPC schema message at the head of the file, then one batch message
//! per acked pack, for the life of the archive. A writer opened over an archive
//! that already has a journal **appends** — it writes no second schema and it
//! truncates nothing.
//!
//! There is exactly one other kind of row and it is appended the same way: a
//! **tombstone** ([`retire_packs`]), written by a `gc` that found every object of
//! a pack dead. It retires that pack's extent without removing its row, because
//! removing a row is the one thing this file's contract forbids. See
//! [`JOURNAL_TOMBSTONE`] for the encoding and [`JournalRow`] for why a pack's
//! ordinal counts pack rows rather than raw rows.
//!
//! That is not a style choice, it is the crash-recovery contract. The extents in
//! this file are one half of §13.12's `indexed` bit: a pack is unabsorbed *iff*
//! its extent is here and its rows are not in the index, and
//! [`GitStore::open_with`](crate::git_ops::GitStore::open_with) diffs the two on
//! every open. A journal truncated by the reopen would erase the evidence that a
//! durable pack was ever acked, so the pack's bytes would stay on disk with
//! nothing pointing at them and nothing able to re-queue them — the exact failure
//! the ordering above exists to prevent, arriving one restart later. Emitting a
//! second schema message mid-file would be no better: arrow's `StreamReader`
//! stops at it, so [`read_journal()`] would silently return only the rows written
//! before this process started.
use std::fs::{File, OpenOptions};
use std::io::{BufWriter, Write};
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use anyhow::{Result, anyhow, bail};
use znippy_common::arrow;
use znippy_common::arrow::array::{ArrayRef, UInt64Array};
use znippy_common::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use znippy_common::arrow::ipc::MetadataVersion;
use znippy_common::arrow::ipc::writer::{
DictionaryTracker, IpcDataGenerator, IpcWriteOptions, write_message,
};
use znippy_common::arrow::record_batch::RecordBatch;
/// A byte range inside the archive: `(offset, len)`. Same shape as
/// [`crate::store::Extent`].
pub type Extent = (u64, u64);
/// Append pushed packfile bytes to an archive, verbatim.
///
/// One trait, three durability contracts. The implementation decides **when**
/// `append` is allowed to return; it never decides what the bytes are.
pub trait ArchiveWrite: Send + Sync {
/// Append these bytes and return when this implementation's durability
/// contract is met. Returns the byte extent written.
///
/// The bytes are stored exactly as handed over: no compression, no framing,
/// no re-encoding. The returned `(offset, len)` addresses them inside the
/// archive's blob region and is the *only* thing that goes to the indexer.
fn append(&self, bytes: &[u8]) -> Result<Extent>;
/// Name it on a bench row.
fn name(&self) -> &'static str;
/// One-line statement of what a crash immediately after `append` returns
/// does to the bytes. On the bench table next to the throughput, because a
/// throughput without this is not a comparison.
fn durability(&self) -> &'static str;
/// **This arm's open file description on the blob file**, for the readers
/// in the same process.
///
/// Every arm opens the file once, read/write, through
/// [`open_blobs`]; a reader `pread`s at an explicit offset and shares no
/// cursor with `append`, so handing the same description out is a
/// descriptor saved and nothing else changed. It is on the trait rather
/// than on the three concrete types because the one caller that matters —
/// [`crate::indexer::PushPath::with_absorber`] — holds a
/// `Box<dyn ArchiveWrite>` and would otherwise have to open its own.
///
/// `None` is reserved for an arm that has no such file. There is none
/// today, and an arm that returns it gets its own handle rather than an
/// error.
fn blobs_shared(&self) -> Option<Arc<File>>;
}
// ── the journal, shared by every durable arm ────────────────────────────────
/// Alignment of an IPC message in the journal. 8 is what
/// `znippy_common::hot`'s journal segments use.
pub const JOURNAL_ALIGNMENT: u8 = 8;
/// The journal's schema: the reference from a row to the blob bytes.
///
/// Deliberately two `u64`s and nothing else. The journal's job is to say *these
/// bytes exist at this extent*; everything derived from the bytes (object count,
/// pack version, checksum) is the indexer's job and is built later, off this
/// path.
pub fn journal_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new("blob_offset", DataType::UInt64, false),
Field::new("blob_size", DataType::UInt64, false),
]))
}
/// The IPC write options every arm's journal uses, so the three arms' journals
/// are byte-comparable.
pub fn journal_options() -> Result<IpcWriteOptions> {
IpcWriteOptions::try_new(JOURNAL_ALIGNMENT as usize, false, MetadataVersion::V5)
.map_err(|e| anyhow!("journal write options: {e}"))
}
/// One journal row for one extent.
pub fn journal_batch(offset: u64, len: u64) -> Result<RecordBatch> {
let off: ArrayRef = Arc::new(UInt64Array::from(vec![offset]));
let size: ArrayRef = Arc::new(UInt64Array::from(vec![len]));
RecordBatch::try_new(journal_schema(), vec![off, size])
.map_err(|e| anyhow!("journal batch: {e}"))
}
/// Serialize the journal's *schema* message — the header an Arrow IPC stream
/// opens with, emitted once per segment.
///
/// Both durable arms call it, and each writes it **once per journal file**:
/// [`SafeWriter`] when it finds the file empty,
/// [`UringWriter`](crate::uring_write::UringWriter) at offset 0 of the one it
/// creates. One encoder, one format, two transports (LAW 5) — and a
/// second schema message inside one file is what would make [`read_journal()`]
/// stop early, which is why neither arm can emit it per writer.
pub fn encode_journal_schema() -> Result<Vec<u8>> {
let opts = journal_options()?;
let dg = IpcDataGenerator {};
let mut tracker = DictionaryTracker::new(false);
let encoded =
dg.schema_to_bytes_with_dictionary_tracker(journal_schema().as_ref(), &mut tracker, &opts);
let mut out = Vec::with_capacity(512);
write_message(&mut out, encoded, &opts).map_err(|e| anyhow!("journal schema encode: {e}"))?;
Ok(out)
}
/// Serialize one journal *batch* message for `(offset, len)`.
pub fn encode_journal_row(offset: u64, len: u64) -> Result<Vec<u8>> {
let opts = journal_options()?;
let dg = IpcDataGenerator {};
let mut tracker = DictionaryTracker::new(false);
let batch = journal_batch(offset, len)?;
let (dicts, msg) = dg
.encode(&batch, &mut tracker, &opts, &mut Default::default())
.map_err(|e| anyhow!("journal batch encode: {e}"))?;
let mut out = Vec::with_capacity(512);
for d in dicts {
write_message(&mut out, d, &opts).map_err(|e| anyhow!("journal dict encode: {e}"))?;
}
write_message(&mut out, msg, &opts).map_err(|e| anyhow!("journal row encode: {e}"))?;
Ok(out)
}
/// Read every complete journal row back, tolerating a torn tail.
///
/// A journal is a **log**, not a document: a process killed mid-write leaves a
/// partial final message, and every complete message before it is real. Same
/// rule `znippy_common::hot`'s segment reader follows. Used by the guards to
/// assert what a crash actually left behind rather than what the writer claimed.
pub fn read_journal(path: &Path) -> Result<Vec<Extent>> {
let bytes = std::fs::read(path).map_err(|e| anyhow!("journal open {}: {e}", path.display()))?;
if bytes.is_empty() {
return Ok(Vec::new());
}
// Only the whole, well-formed messages reach arrow (the IPC guard's valid
// prefix): a hostile message reads as the torn tail it is treated as, never
// as an arrow-ipc panic or abort.
let whole = znippy_common::ipc_guard::valid_prefix(&bytes);
let reader = match arrow::ipc::reader::StreamReader::try_new(std::io::Cursor::new(whole), None) {
Ok(r) => r,
// A journal with a torn *schema* message carries no rows at all.
Err(_) => return Ok(Vec::new()),
};
let mut out = Vec::new();
for batch in reader {
let Ok(batch) = batch else { break }; // torn tail: stop, keep what is whole
let offs = batch
.columns()
.first()
.ok_or_else(|| anyhow!("journal has no column 0"))?
.as_any()
.downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("journal column 0 is not u64"))?;
let sizes = batch
.columns()
.get(1)
.ok_or_else(|| anyhow!("journal has no column 1"))?
.as_any()
.downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("journal column 1 is not u64"))?;
for i in 0..batch.num_rows() {
out.push((offs.value(i), sizes.value(i)));
}
}
Ok(out)
}
// ── retirement: the one row a `gc` appends ──────────────────────────────────
/// The `blob_size` of a journal row that is **not** a pack: a **tombstone**,
/// whose `blob_offset` names a pack this archive has retired.
///
/// `u64::MAX` rather than `0`, and that is the whole of the encoding decision.
/// The journal's schema is two `u64`s and a third column cannot be added without
/// making every existing journal file unreadable by the writer that appends to
/// it, so the marker has to live inside a value that no real pack can take. A
/// zero-length append is at least *conceivable* — a caller handing `append` an
/// empty slice gets `(cursor, 0)` — while an extent of 2^64-1 bytes cannot
/// exist on any filesystem this will ever run on. The unreachable value is the
/// safe one.
pub const JOURNAL_TOMBSTONE: u64 = u64::MAX;
/// One journal row, read back and interpreted.
///
/// **The ordinal of a pack is its position among the [`Pack`](JournalRow::Pack)
/// rows, not its row index.** Tombstones are appended to the same log — that is
/// what keeps the journal append-only — so counting raw rows would renumber
/// every pack acked before a `gc` and make the ordinals a live process holds
/// disagree with the ones the next open derives. Counting pack rows cannot: a
/// pack row is never removed and never reordered, so a pack's ordinal is fixed
/// for the life of the archive.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JournalRow {
/// A pack was acked at this extent.
Pack(Extent),
/// The pack that starts at this offset was retired: a `gc` found **every**
/// one of its objects dead and dropped every row it had. Its bytes are still
/// in the blob file (the blob file is append-only and nothing truncates it),
/// but no open may ever re-queue it for indexing again.
Retired(u64),
}
/// Interpret what [`read_journal`] returned.
pub fn journal_rows(rows: &[Extent]) -> Vec<JournalRow> {
rows.iter()
.map(|&(offset, len)| {
if len == JOURNAL_TOMBSTONE {
JournalRow::Retired(offset)
} else {
JournalRow::Pack((offset, len))
}
})
.collect()
}
/// The acked pack extents, in **ordinal order** — index `i` is pack `i`.
pub fn acked_packs(rows: &[Extent]) -> Vec<Extent> {
journal_rows(rows)
.into_iter()
.filter_map(|r| match r {
JournalRow::Pack(e) => Some(e),
JournalRow::Retired(_) => None,
})
.collect()
}
/// The offsets of the packs a `gc` has retired.
pub fn retired_offsets(rows: &[Extent]) -> std::collections::HashSet<u64> {
journal_rows(rows)
.into_iter()
.filter_map(|r| match r {
JournalRow::Retired(o) => Some(o),
JournalRow::Pack(_) => None,
})
.collect()
}
/// **Retire these packs: one tombstone row each, durable before this returns.**
///
/// Called by `GitOps::gc` *before* it drops a single index row, and that order
/// is the crash contract. The two possible interruptions are not symmetric:
///
/// * killed **before** the tombstones are durable — the rows are still in the
/// index, so the next open sees packs that have rows, calls them absorbed, and
/// nothing is lost or resurrected. The GC simply did not happen.
/// * killed **after** — the rows may or may not have gone, and either way the
/// next open reads a tombstone and refuses to re-queue the pack. The dead
/// objects cannot come back.
///
/// The reverse order has a window in which the rows are gone and the journal
/// still claims an unabsorbed pack, which is exactly the resurrection this
/// function exists to close.
///
/// A torn write leaves a partial final message, which [`read_journal`] discards
/// with the rest of a torn tail — the same tolerance a torn *pack* row gets, and
/// safe for the same reason: a tombstone that did not land is a `gc` that did not
/// happen.
///
/// All the rows go out in **one** `write(2)` on an `O_APPEND` fd, so a push
/// appending through [`SafeWriter`]'s own fd at the same instant cannot land
/// inside one of them.
pub fn retire_packs(journal: &Path, offsets: &[u64]) -> Result<()> {
if offsets.is_empty() {
return Ok(());
}
let mut buf = Vec::with_capacity(offsets.len() * 256);
for &offset in offsets {
buf.extend_from_slice(&encode_journal_row(offset, JOURNAL_TOMBSTONE)?);
}
let f = OpenOptions::new()
.append(true)
.open(journal)
.map_err(|e| anyhow!("open journal {} to retire a pack: {e}", journal.display()))?;
if f.metadata()?.len() == 0 {
// No schema message, therefore no pack row, therefore nothing that could
// have been retired. Writing a tombstone into an empty stream would
// produce a journal whose first message is a batch, which reads back as
// no rows at all.
bail!(
"{} is empty — nothing was ever acked here, so there is no pack to retire",
journal.display()
);
}
(&f).write_all(&buf)
.map_err(|e| anyhow!("journal tombstone write: {e}"))?;
f.sync_all()
.map_err(|e| anyhow!("journal tombstone fsync: {e}"))?;
Ok(())
}
/// **The writer lock** — `P-014`. One writer per blob file, enforced by the
/// kernel, held for as long as the writer's handle lives.
///
/// # What it is protecting against, measured
///
/// Every arm reserves its extent from an `AtomicU64` seeded here from the file's
/// length **at open time**. Two processes that both open before either appends
/// therefore hold two cursors with the same value and hand out *the same
/// offsets*. Driven on t14s 2026-08-11, two processes × 16 one-blob packs on
/// `SafeWriter`, released together: 32 packs acked, `objects.pack` never grew
/// past one writer's worth, and **16 of 32 acked, fsynced, journalled objects
/// were destroyed with two zero exit codes**. Neither writer errored, which is
/// the whole problem.
///
/// # Why `flock` and not an `O_EXCL` lock file
///
/// `S-019`: *"the `O_EXCL` store lock does not self-clear; a killed appliance
/// will not restart."* A lock that survives a crash is worse than the race it
/// prevents. `flock(2)` is released by the kernel when the last descriptor for
/// the open file description closes — including on `SIGKILL`, including on a
/// panic — so a killed server comes back up. Verified rather than recited: a
/// store held by one process, `kill -9`, and the next open succeeds.
///
/// [`std::fs::File::try_lock`] *is* `flock(LOCK_EX | LOCK_NB)` on unix, so this
/// costs no dependency, and it is **the same mechanism redb already uses** on
/// `objects.tail` (`redb-2.6.3/src/tree_store/page_store/file_backend/unix.rs:37`).
/// One locking discipline in this store, not two.
///
/// # Why at open, and NOT around `append`
///
/// The ack path is the latency path: `append` returns to the client and the
/// indexer runs after it. A lock taken per append would put a syscall on that
/// path and would still not help, because the *cursor* is what is stale, not the
/// write. Taken once here, the ack path takes **no lock at all** and nothing
/// about `append`'s cost changes.
///
/// It is here, in `open_blobs`, because that is the one function
/// [`FastWriter`], [`SafeWriter`] and
/// [`UringWriter`](crate::uring_write::UringWriter) all call — one lock for
/// three arms, rather than one arm fixed and two left. It also runs *before*
/// `UringWriter::create`'s `File::create(journal)`, so a second process can no
/// longer truncate the journal on its way to being refused.
///
/// # What it deliberately does NOT cover
///
/// **Two threads sharing one open handle.** An advisory lock is per open file
/// *description*; a second `try_lock` on the same handle returns `Ok`. That case
/// is `tests/concurrent_push.rs`'s second test, and it needs no lock: the
/// `fetch_add` cursor every arm shares makes two concurrent appends touch
/// provably disjoint ranges. A mutex there would serialise the ack path to fix a
/// race that cannot happen.
///
/// # Refuse, do not wait
///
/// `LOCK_NB`, matching redb. A server that blocked here would hang on startup
/// behind a process it cannot see, with no error and no timeout; a refusal names
/// the file and the caller already handles one, because redb's has been throwing
/// it all along.
fn lock_for_writing(f: &File, path: &Path) -> Result<()> {
match f.try_lock() {
Ok(()) => Ok(()),
Err(std::fs::TryLockError::WouldBlock) => bail!(
"another writer already holds {} — one process at a time appends to a store's \
blob file. Two would share a stale append cursor and overwrite each other's \
acked packs without either one erroring. The lock is an advisory flock and the \
kernel drops it when that process dies, so nothing has to be cleaned up by hand.",
path.display()
),
Err(std::fs::TryLockError::Error(e)) => Err(anyhow!(
"locking {} for writing: {e} — the store is not opened without the writer lock, \
because an unlocked open is the P-014 race",
path.display()
)),
}
}
/// Open (creating) the blob file an archive's payload region lives in, and
/// report the cursor to append at.
///
/// **Takes the writer lock** before reading the cursor — see
/// [`lock_for_writing`] for why the lock is here and not around `append`. The
/// order matters: the length this returns is only meaningful to a writer that
/// owns the file, and it is read after the lock so it cannot be stale by the
/// time it is used.
/// What a journal file's head says an opening writer must do about it.
pub(crate) struct JournalHead {
/// The schema message this build writes.
pub schema: Vec<u8>,
/// Write it (again). False for a file that already opens with it.
pub rewrite: bool,
}
/// ── AN EMPTY STORE IS NOT MADE DURABLE, IT IS MADE AGAIN ────────────────────
///
/// Both durable arms used to `fsync` the schema message they wrote into a fresh
/// journal. On gunnar's create path that barrier was 8.9 ms of a 68.5 ms
/// push-create (lane V's per-step medians, 2 vCPU rig), spent making durable a
/// file that references nothing: a store with no journal row has acked no pack,
/// and gunnar's create is transactional, so a create that does not finish leaves
/// no catalog row pointing at the store at all. The first row a writer claims
/// fsyncs the same file and takes the schema with it, so from the first ack
/// onward nothing about durability differs.
///
/// The one thing the barrier bought was that a schema torn by a crash could not
/// be inherited by the next open as if it were whole. That is bought here
/// instead, by **checking** — a short read of a file the open is about to read
/// anyway:
///
/// * an empty file gets the schema;
/// * a file that begins with exactly this build's schema message, and is at
/// least as long as one, is a live journal and is left alone;
/// * anything else **no longer than the schema** was interrupted before a row
/// could exist behind it (rows are appended after it), so it is written again
/// and nothing is lost;
/// * anything else **longer** than the schema did carry rows, so it is corrupt
/// rather than interrupted, and it is said out loud instead of truncated.
///
/// One function for both arms on purpose: LAW 5 already claims they are
/// indistinguishable on disk, and two copies of this reasoning is how that stops
/// being true.
pub(crate) fn journal_head(f: &File, path: &Path) -> Result<JournalHead> {
let schema = encode_journal_schema()?;
let already = f
.metadata()
.map_err(|e| anyhow!("stat journal {}: {e}", path.display()))?
.len();
if already == 0 {
return Ok(JournalHead { schema, rewrite: true });
}
let mut got = vec![0u8; schema.len().min(already as usize)];
f.read_exact_at(&mut got, 0)
.map_err(|e| anyhow!("reading the journal header of {}: {e}", path.display()))?;
let head_matches = got == schema[..got.len()];
if head_matches && already >= schema.len() as u64 {
return Ok(JournalHead { schema, rewrite: false });
}
if already <= schema.len() as u64 {
return Ok(JournalHead { schema, rewrite: true });
}
bail!(
"the journal {} does not begin with this build's schema message and is longer than one, \
so it carried acked packs: it is corrupt, and truncating it would unreference their bytes",
path.display()
)
}
/// **The one open file description on a store's blob file** — read/write,
/// `flock`ed, and shared.
///
/// `Arc<File>` rather than `File` because this handle serves three callers in
/// one process and used to be opened three times: the writer appends through
/// it, [`crate::git_ops::Absorber`] `pread`s extents back through it, and the
/// index workers `read_exact_at` through it. All three address the file
/// positionally, the file is append-only and `gc` truncates nothing (§14), so
/// there is no shared cursor and nothing to observe between them — three
/// descriptors bought three kernel objects and three page-cache aliases per
/// open repository and no isolation at all. See `tests/fd_per_store.rs`.
///
/// The `flock` therefore lives as long as the last of the three, which is the
/// store: a reader outliving the writer can no longer drop the writer lock out
/// from under a store that is still open.
pub(crate) fn open_blobs(path: &Path) -> Result<(Arc<File>, u64)> {
let f = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(path)
.map_err(|e| anyhow!("open blobs {}: {e}", path.display()))?;
lock_for_writing(&f, path)?;
let end = reclaim_orphan_tail(&f, path)?;
Ok((Arc::new(f), end))
}
/// **Give back the bytes past the last journaled extent** — the append-side gc,
/// run once at open while this process holds the writer `flock` and before any
/// append can race it.
///
/// # The leak it closes, which is older than the caller that made it obvious
///
/// An append is two durable acts in order — the bytes, then the journal row
/// naming them. A crash between them leaves bytes in the blob file that no
/// extent addresses, and the cursor is taken from the file's LENGTH, so the next
/// append starts past them and they are never reached again.
/// [`SafeWriter::create_inner`]'s own comment already names the shape ("a reopen
/// that dropped them would leave their bytes unreferenced for ever") — it is
/// talking about the journal rows, and this is the other half.
///
/// Nothing referenced them, so nothing is lost by taking them back: a push whose
/// journal row never landed was never acked, and the client will send it again.
///
/// # It refuses every ambiguous case rather than guessing
///
/// Truncation is the one operation here that cannot be undone, so it happens
/// only when the journal positively says where the data ends:
///
/// * **no journal file at all** — [`FastWriter`] keeps none, so an absent
/// journal is a normal arm and not evidence of anything. Left alone.
/// * **a journal that reads as empty** — [`read_journal`] answers `Ok(vec![])`
/// for a torn *schema* message as well as for a genuinely empty log, and those
/// two are not distinguishable from here. Truncating on that reading would
/// turn one bad message into the loss of every pack in the archive. Left
/// alone, loudly.
/// * **tombstones** ([`JOURNAL_TOMBSTONE`]) carry `u64::MAX` as their length and
/// are skipped: their `blob_offset` names a retired pack, and adding that
/// length would compute a high-water mark past the end of the universe.
fn reclaim_orphan_tail(f: &File, archive: &Path) -> Result<u64> {
let len = f.metadata()?.len();
if len == 0 {
return Ok(0);
}
let journal = journal_path(archive);
if !journal.exists() {
return Ok(len);
}
let rows = read_journal(&journal)?;
let high_water = rows
.iter()
.filter(|(_, l)| *l != JOURNAL_TOMBSTONE)
.map(|(o, l)| o.saturating_add(*l))
.max();
let Some(high_water) = high_water else {
if len > 0 {
tracing::warn!(
target: "znippy.git",
archive = %archive.display(),
blob_bytes = len,
"the journal names no extent but the blob file is not empty — leaving it \
alone: an unreadable journal and an empty one look the same from here, and \
truncating on that reading would drop every pack in the archive"
);
}
return Ok(len);
};
if len <= high_water {
return Ok(len);
}
let orphan = len - high_water;
f.set_len(high_water)
.map_err(|e| anyhow!("reclaiming {orphan} orphan byte(s) of {}: {e}", archive.display()))?;
tracing::warn!(
target: "znippy.git",
archive = %archive.display(),
reclaimed = orphan,
blob_bytes = high_water,
"gave back blob bytes past the last journaled extent — a crash between an append \
and its journal row. Nothing referenced them; the push they belonged to was never \
acked"
);
Ok(high_water)
}
// ── impl 1 — FastWriter ─────────────────────────────────────────────────────
/// **The cheating arm.** Returns before the bytes are on disk.
///
/// # What it does NOT guarantee
///
/// `append` returns as soon as one `pwrite(2)` has handed the bytes to the
/// kernel's page cache. There is **no `fsync`**, **no journal row**, and **no
/// barrier of any kind**. Concretely:
///
/// * **A crash (power loss, kernel panic, `SIGKILL` of the box) after `append`
/// returns loses the data.** Not "may lose" — the bytes exist only in volatile
/// page cache and nothing has told the device about them.
/// * Nothing on disk references the extent, so even bytes that *did* reach the
/// platter are unreachable after a restart until something scans for them.
/// * A process crash alone (not a machine crash) does keep them: page cache
/// survives `exit`. That is the only crash it survives.
///
/// It is here to **bound** [`SafeWriter`] and [`UringWriter`]: it is the ceiling
/// the price of durability is measured against — at 8 KiB it acks in 3.9 µs
/// against `SafeWriter`'s 132 µs (`examples/push_path_bench.rs`, oden
/// 2026-08-07).
///
/// # Selecting it
///
/// It is a first-class arm of [`crate::arms::WriterArm`] and
/// [`GitStore`](crate::git_ops::GitStore) will be built on it if it is asked
/// for. Nothing refuses it, because "durable before ack" is git's contract and
/// not every store on this code is serving git pushes: a mirror that can be
/// re-cloned, an import that is re-run on failure and a benchmark all have a
/// weaker requirement, and for them the four bullets above are a price they are
/// not obliged to pay. A store that *is* serving pushes and picks this one loses
/// acked data on a machine crash, and that is the whole of what the choice
/// means.
///
/// There is one consequence beyond durability, because it follows from the same
/// absence: with no journal there is no durable record that a pack was ever
/// acked, so a reopened store re-queues nothing and pack ordinals restart at 0
/// (`indexer::packs_already_acked`).
///
/// # Zero-copy
///
/// The caller's `bytes` are handed to `pwrite` at their own address. There is no
/// `to_vec`, no staging buffer and no `BufWriter` — one userspace→kernel copy,
/// which is the syscall itself and cannot be removed without io_uring registered
/// buffers (see [`UringWriter`]).
pub struct FastWriter {
blobs: Arc<File>,
cursor: AtomicU64,
}
impl FastWriter {
/// Open `archive` for verbatim appends.
pub fn create(archive: &Path) -> Result<Self> {
let (blobs, end) = open_blobs(archive)?;
Ok(Self {
blobs,
cursor: AtomicU64::new(end),
})
}
/// The blob file, for a reader that wants to `pread` an extent back.
pub fn blobs(&self) -> &File {
&self.blobs
}
}
impl ArchiveWrite for FastWriter {
fn append(&self, bytes: &[u8]) -> Result<Extent> {
let len = bytes.len() as u64;
// Reserve the extent atomically, then write it positionally. Two
// concurrent appends touch disjoint ranges and never share a file
// offset, so no lock is needed and none is taken.
let offset = self.cursor.fetch_add(len, Ordering::SeqCst);
self.blobs.write_all_at(bytes, offset)?;
// …and return. The bytes are in the page cache. That is the whole point.
Ok((offset, len))
}
fn name(&self) -> &'static str {
"FastWriter"
}
fn durability(&self) -> &'static str {
"none — page cache only; a machine crash after return loses the bytes"
}
fn blobs_shared(&self) -> Option<Arc<File>> {
Some(Arc::clone(&self.blobs))
}
}
// ── impl 2 — SafeWriter ─────────────────────────────────────────────────────
struct SafeJournal {
/// The userspace buffer the trap above is about. Raw Arrow IPC messages go
/// into it — [`encode_journal_schema`] once, [`encode_journal_row`] per
/// append — the same two calls
/// [`UringWriter`](crate::uring_write::UringWriter) builds its journal from
/// (LAW 5: one journal format, one encoder, two transports). An arrow
/// `StreamWriter` cannot be used here because it emits
/// its schema on construction, and a second schema message is what an
/// appending reopen must not write.
writer: BufWriter<File>,
path: PathBuf,
}
/// The ordering znippy's own hot path already uses, followed rather than
/// re-invented.
///
/// Per `append`, in this order and no other:
///
/// 1. `pwrite` the blob bytes at the reserved extent;
/// 2. `fsync` the **blobs** — the bytes are on the platter;
/// 3. write one Arrow IPC `RecordBatch` naming the extent into the journal;
/// 4. `flush()` the journal's `BufWriter` — userspace → kernel;
/// 5. `sync_all()` the journal file — kernel → platter.
///
/// Steps 4 and 5 are not interchangeable and 5 alone is not enough: see the
/// module docs' "BufWriter trap". Steps 2 and 3 are not interchangeable either
/// — that is the crash-ordering contract, and
/// `crash_between_fsyncs_leaves_orphan_bytes_not_a_dangling_reference` asserts
/// it on real files.
///
/// # Durability on return
///
/// The bytes are on the device and a journal row on the device points at them.
/// A crash after return keeps both.
pub struct SafeWriter {
blobs: Arc<File>,
cursor: AtomicU64,
journal: Mutex<SafeJournal>,
faults: Faults,
}
/// Injected faults, so the crash-ordering guard exercises the **real** `append`
/// rather than a hand-rolled twin of it (LAW 5: one writer, not two copies that
/// a guard then watches agree).
///
/// Never set outside a test. [`SafeWriter::create`] leaves both clear.
#[derive(Debug, Clone, Copy, Default)]
pub struct Faults {
/// Skip step (2), the blob `fsync`.
pub skip_blob_fsync: bool,
/// Stop after step (2) and before step (3) — the exact on-disk state a
/// machine that dies between the two fsyncs leaves behind. `append` returns
/// `Err` because a Rust function has to return something; a real crash
/// simply would not return, and the bytes on the platter are identical
/// either way, which is what the guard asserts on.
pub die_between_fsyncs: bool,
}
impl SafeWriter {
/// Open `archive` and start a journal segment beside it at
/// `<archive>.journal`.
pub fn create(archive: &Path) -> Result<Self> {
Self::create_inner(archive, Faults::default())
}
/// Same, with an injected fault. **Tests only.**
pub fn create_with_faults(archive: &Path, faults: Faults) -> Result<Self> {
Self::create_inner(archive, faults)
}
fn create_inner(archive: &Path, faults: Faults) -> Result<Self> {
let (blobs, end) = open_blobs(archive)?;
let path = journal_path(archive);
// **Append, never truncate** (module docs): the rows already here are the
// durable record that those packs were acked, and a reopen that dropped
// them would leave their bytes unreferenced for ever.
let f = OpenOptions::new()
.read(true)
.append(true)
.create(true)
.open(&path)
.map_err(|e| anyhow!("open journal {}: {e}", path.display()))?;
let head = journal_head(&f, &path)?;
if head.rewrite && f.metadata()?.len() != 0 {
f.set_len(0)
.map_err(|e| anyhow!("resetting the interrupted journal {}: {e}", path.display()))?;
}
// DELIBERATELY buffered. See the module docs: the flush→sync ordering is
// only load-bearing when there is a userspace buffer to lose.
let mut j = SafeJournal {
writer: BufWriter::new(f),
path,
};
// The schema message opens the stream and is written **once per file**,
// not once per writer.
if head.rewrite {
j.writer
.write_all(&head.schema)
.map_err(|e| anyhow!("journal schema: {e}"))?;
// Out of the userspace buffer, so the next `sync_all` — the one the
// first acked pack pays — has something to make durable.
j.writer.flush().map_err(|e| anyhow!("journal flush: {e}"))?;
}
Ok(Self {
blobs,
cursor: AtomicU64::new(end),
journal: Mutex::new(j),
faults,
})
}
/// Path of the journal segment beside `archive`.
pub fn journal_path(archive: &Path) -> PathBuf {
journal_path(archive)
}
/// The blob file, for a reader that wants to `pread` an extent back.
pub fn blobs(&self) -> &File {
&self.blobs
}
/// The journal segment this writer is appending rows to.
pub fn journal_file(&self) -> PathBuf {
self.journal.lock().expect("journal mutex").path.clone()
}
}
pub(crate) fn journal_path(archive: &Path) -> PathBuf {
let mut s = archive.as_os_str().to_os_string();
s.push(".journal");
PathBuf::from(s)
}
impl ArchiveWrite for SafeWriter {
fn append(&self, bytes: &[u8]) -> Result<Extent> {
let len = bytes.len() as u64;
let offset = self.cursor.fetch_add(len, Ordering::SeqCst);
// (1) blob bytes, positional, no userspace copy of `bytes`.
self.blobs.write_all_at(bytes, offset)?;
// (2) blob bytes DURABLE — before anything references them. A crash from
// here to the end of this function leaves orphan payload nobody
// points at, which the seal drops. The reverse order would leave a
// row pointing into a hole.
if !self.faults.skip_blob_fsync {
self.blobs.sync_all()?;
}
if self.faults.die_between_fsyncs {
// The machine is gone. On-disk state: the blob bytes, fsynced, and a
// journal that has never heard of them. Orphan payload, not a
// dangling reference.
bail!("injected crash between the two fsyncs");
}
let mut j = self
.journal
.lock()
.map_err(|_| anyhow!("journal mutex poisoned"))?;
// (3) the row that claims the extent — one IPC batch message appended
// after every row this archive has ever acked.
let row = encode_journal_row(offset, len)?;
j.writer
.write_all(&row)
.map_err(|e| anyhow!("journal write: {e}"))?;
// (4) userspace -> kernel. WITHOUT THIS, (5) SYNCS NOTHING.
j.writer.flush().map_err(|e| anyhow!("journal flush: {e}"))?;
// (5) kernel -> platter.
j.writer
.get_ref()
.sync_all()
.map_err(|e| anyhow!("journal fsync: {e}"))?;
Ok((offset, len))
}
fn name(&self) -> &'static str {
"SafeWriter"
}
fn durability(&self) -> &'static str {
"full — blob fsynced, then a journal row fsynced; crash after return keeps both"
}
fn blobs_shared(&self) -> Option<Arc<File>> {
Some(Arc::clone(&self.blobs))
}
}
// ── generation 0: the seal that makes an archive exist at all ────────────────
/// What one [`seal_generation_zero`] put on disk. Every count is taken off the
/// journal and the sealed file, never echoed back from an argument.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SealReport {
/// The archive that now exists.
pub archive: PathBuf,
/// Verbatim packs that got an index row — one row each, `chunk_seq = 0`.
pub packs_sealed: u64,
/// Acked packs the journal has since tombstoned. They keep their bytes in
/// the blob region and get **no** row, which is what makes them dead payload
/// for the compaction that follows.
pub packs_retired: u64,
/// Acked packs whose extent runs past the bytes that were copied — a pack
/// acked *after* the copy began. They get no row either, because a row for
/// one would address bytes this archive does not carry. Normally `0`; it is
/// reported rather than swallowed so a seal that raced a push says so.
pub packs_after_copy: u64,
/// Bytes of blob region carried into the archive — the whole blob file,
/// verbatim, so every journal extent still addresses the bytes it named.
pub blob_bytes: u64,
/// Size of the sealed archive, blob region and metadata tail together.
pub sealed_total_bytes: u64,
}
/// **Write generation 0.**
///
/// Both [`Gc`](crate::gc::Gc) implementations compact an archive that already
/// exists — [`NewGeneration`](crate::gc::NewGeneration) writes `repository.g1.znippy`
/// beside `repository.znippy`. Nothing created `repository.znippy`, so a `gc()`
/// on a store that had never been sealed died on `stat: No such file or
/// directory`. This is the function that had been missing.
///
/// # An archive is a blob region plus a metadata tail
///
/// The blob region here is the store's `objects.pack` — the verbatim pushed
/// packs, at the offsets the journal named — and it is copied **whole and
/// unchanged**. That is not laziness: a journal extent is `(offset, len)` into
/// that file, and the same extents are what the store's `objects` table holds
/// for every object. Rewriting the region to squeeze the dead packs out would
/// move every offset after the first hole and invalidate every one of those
/// rows. Reclaiming that payload is the compaction's job, and it can do it
/// precisely because a retired pack gets no row here.
///
/// The tail is written past the copied blob region by
/// [`ArrowIpcSink`](znippy_common::ArrowIpcSink) — the same writer
/// `HotArchive::seal` uses, with the same reserved-section hook, so there is one
/// metadata writer in this constellation and not two (LAW 5).
///
/// # One row per acked pack
///
/// | column | value |
/// |---|---|
/// | `relative_path` | `objects.pack.<ordinal>` — gunnar's `synthetic_path` convention, and the ordinal is the pack's position among the journal's [`JournalRow::Pack`] rows |
/// | `blob_offset` / `blob_size` | the journal extent, unchanged |
/// | `uncompressed_size` | the same length: a verbatim pack is stored raw |
/// | `compressed` | `false` — the bytes on disk are the pack's own |
/// | `chunk_seq` / `fdata_offset` | `0` — one chunk per pack |
/// | `checksum` | blake3 **over the stored bytes**, which for `compressed: false` are also the original bytes — the domain `write_blobs` uses and the domain `extract_file_verified` checks against |
///
/// A tombstoned pack is skipped, and so is one whose extent runs past the bytes
/// that were copied — that second case is a pack acked *after* the copy began,
/// and a row for it would point into a hole. Both are **counted** on the
/// [`SealReport`] rather than swallowed.
///
/// # Ordering
///
/// The blob region is copied first and the journal is read second, because the
/// blob file is append-only: anything the journal names within the copied length
/// is certainly present, while the reverse order could name an extent the copy
/// had not reached. The tail is written into a staging sibling and the archive's
/// own name appears only at the final `rename(2)`, so an interruption at any
/// byte leaves no half-sealed archive under the name a reader opens.
pub fn seal_generation_zero(
blobs: &Path,
journal: Option<&Path>,
archive: &Path,
reserved: Vec<znippy_common::ReservedSection>,
) -> Result<SealReport> {
use znippy_common::index::{ChunkLoc, data_subindex_schema};
use znippy_common::{ArchiveMetaSink, ArrowIpcSink, GroupKey, base_batch_from_rows};
let staged = staging_sibling(archive);
let sealed = (|| -> Result<SealReport> {
// (1) the blob region, verbatim. Every journal extent still addresses
// the bytes it named because not one of them moved.
let blob_bytes = std::fs::copy(blobs, &staged).map_err(|e| {
anyhow!(
"copying the blob region {} into {}: {e}",
blobs.display(),
staged.display()
)
})?;
// (2) the journal, second — see the ordering note above.
let rows = match journal {
Some(p) if p.exists() => read_journal(p)?,
_ => Vec::new(),
};
let packs = acked_packs(&rows);
let retired = retired_offsets(&rows);
let file = OpenOptions::new()
.read(true)
.write(true)
.open(&staged)
.map_err(|e| anyhow!("reopening {} to seal into: {e}", staged.display()))?;
let mut paths: Vec<String> = Vec::with_capacity(packs.len());
let mut locs: Vec<ChunkLoc> = Vec::with_capacity(packs.len());
let mut packs_retired = 0u64;
let mut packs_after_copy = 0u64;
for (ordinal, &(offset, len)) in packs.iter().enumerate() {
if retired.contains(&offset) {
packs_retired += 1;
continue;
}
if offset.saturating_add(len) > blob_bytes {
// Acked after the copy began. A row for it would address bytes
// this archive does not carry.
packs_after_copy += 1;
continue;
}
paths.push(format!("objects.pack.{ordinal}"));
locs.push(ChunkLoc {
chunk_seq: 0,
fdata_offset: 0,
blob_offset: offset,
blob_size: len,
uncompressed_size: len,
compressed: false,
checksum: blake3_extent(&file, offset, len)?,
});
}
let packs_sealed = paths.len() as u64;
// (3) the metadata tail, past the blob region.
let file = Arc::new(file);
let mut sink = ArrowIpcSink::new(Arc::clone(&file), blob_bytes);
if !reserved.is_empty() {
sink = sink.with_reserved_builder(Box::new(move |_| Ok(reserved)));
}
if !paths.is_empty() {
let batch = base_batch_from_rows(&paths, &locs)?;
sink.push_subindex(
data_subindex_schema().as_ref(),
&[batch],
GroupKey {
pkg_type: 0,
repo: String::new(),
module_name: String::new(),
},
)?;
}
// `finish` fsyncs before it returns the total.
let sealed_total_bytes = Box::new(sink).finish()?;
Ok(SealReport {
archive: archive.to_path_buf(),
packs_sealed,
packs_retired,
packs_after_copy,
blob_bytes,
sealed_total_bytes,
})
})();
let sealed = match sealed {
Ok(s) => s,
Err(e) => {
let _ = std::fs::remove_file(&staged);
return Err(e);
}
};
// (4) the archive's own name, atomically, and only now.
std::fs::rename(&staged, archive).map_err(|e| {
let _ = std::fs::remove_file(&staged);
anyhow!(
"renaming {} into place as {}: {e}",
staged.display(),
archive.display()
)
})?;
sync_parent_dir(archive);
Ok(sealed)
}
/// blake3 over `(offset, len)` of `file`, streamed rather than buffered whole:
/// a consolidated pack is tens of megabytes and there is no reason for the seal
/// to hold one in memory to hash it.
fn blake3_extent(file: &File, offset: u64, len: u64) -> Result<[u8; 32]> {
const WINDOW: usize = 1 << 20;
let mut hasher = znippy_common::blake3::Hasher::new();
let mut buf = vec![0u8; WINDOW.min(len.max(1) as usize)];
let mut done = 0u64;
while done < len {
let want = ((len - done) as usize).min(buf.len());
file.read_exact_at(&mut buf[..want], offset + done)
.map_err(|e| anyhow!("reading the pack at ({offset}, {len}) to checksum it: {e}"))?;
hasher.update(&buf[..want]);
done += want as u64;
}
Ok(*hasher.finalize().as_bytes())
}
/// A sibling name nothing else can be holding: pid and nanos, the same shape
/// `gc.rs` stages its compaction under.
fn staging_sibling(archive: &Path) -> PathBuf {
let unique = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let mut p = archive.as_os_str().to_owned();
p.push(format!(".seal-{}-{unique}", std::process::id()));
PathBuf::from(p)
}
/// fsync the directory so the rename is durable and not merely visible.
/// Best-effort, like `gc::sync_dir`: a filesystem that will not open a directory
/// is not a reason to fail a seal that otherwise succeeded.
fn sync_parent_dir(path: &Path) {
if let Some(parent) = path.parent()
&& let Ok(f) = File::open(parent)
{
let _ = f.sync_all();
}
}