znippy-plugin-git 0.1.1

//! 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;
}

// ── 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 f = File::open(path).map_err(|e| anyhow!("journal open {}: {e}", path.display()))?;
    if f.metadata()?.len() == 0 {
        return Ok(Vec::new());
    }
    let reader = match arrow::ipc::reader::StreamReader::try_new(std::io::BufReader::new(f), 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
            .column(0)
            .as_any()
            .downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow!("journal column 0 is not u64"))?;
        let sizes = batch
            .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.
pub(crate) fn open_blobs(path: &Path) -> Result<(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 = f.metadata()?.len();
    Ok((f, end))
}

// ── 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: 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"
    }
}

// ── 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: 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 already = f.metadata()?.len();
        // 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 already == 0 {
            j.writer
                .write_all(&encode_journal_schema()?)
                .map_err(|e| anyhow!("journal schema: {e}"))?;
            // The schema message itself is durable before any row is claimed.
            j.writer.flush().map_err(|e| anyhow!("journal flush: {e}"))?;
            j.writer
                .get_ref()
                .sync_all()
                .map_err(|e| anyhow!("journal fsync: {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"
    }
}

// ── 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();
    }
}