znippy-plugin-git 0.1.3

//! Guards for the push path: [`ArchiveWrite`]'s three arms and the shared
//! per-account indexer.
//!
//! **Every assertion here is on applied output** — bytes read back off the disk
//! through a *fresh* file handle, journal rows decoded out of the Arrow IPC
//! stream, index rows decoded out of the built `RecordBatch`. Not one of them
//! asserts that a call returned `Ok`, because `Ok` is exactly what a writer that
//! silently dropped the payload also returns.
//!
//! **LAW 2 — every guard below was seen RED.** Each test's doc comment records
//! the exact edit that broke it and what the runner then printed. Runs on
//! 2026-08-07, oden, `--no-default-features`, `/proc/loadavg` 0.26–2.1 across
//! the session.

use std::fs::File;
use std::io::Read;
use std::os::unix::fs::FileExt;
use std::path::Path;
use std::sync::Arc;

use znippy_common::arrow::array::{StringArray, UInt32Array, UInt64Array};
use znippy_plugin_git::archive_write::{
    ArchiveWrite, FastWriter, Faults, SafeWriter, read_journal,
};
use znippy_plugin_git::indexer::{AccountIndexer, IndexJob, Lookup, PushPath};
use znippy_plugin_git::uring_write::UringWriter;

// ── fixtures ────────────────────────────────────────────────────────────────

/// Bytes shaped like a v2 packfile: the 12-byte header a real pack opens with
/// (`PACK`, version, object count) followed by an incompressible body.
///
/// Incompressible on purpose. These bytes are stored **verbatim**, and a body of
/// zeroes would let a writer that quietly deflated the payload still pass a
/// length check.
fn pack_shaped(objects: u32, body: usize) -> Vec<u8> {
    let mut v = Vec::with_capacity(12 + body);
    v.extend_from_slice(b"PACK");
    v.extend_from_slice(&2u32.to_be_bytes());
    v.extend_from_slice(&objects.to_be_bytes());
    // xorshift64*, so the body is deterministic across runs and has no
    // compressible structure.
    let mut s: u64 = 0x9E37_79B9_7F4A_7C15 ^ (objects as u64) << 32 ^ body as u64;
    while v.len() < 12 + body {
        s ^= s >> 12;
        s ^= s << 25;
        s ^= s >> 27;
        v.extend_from_slice(&s.wrapping_mul(0x2545_F491_4F6C_DD1D).to_le_bytes());
    }
    v.truncate(12 + body);
    v
}

/// Read `len` bytes at `offset` through a **fresh** handle. Never the writer's
/// own `File` — that would let a writer that never issued the syscall still pass
/// by handing back its own buffer.
fn read_back(path: &Path, offset: u64, len: u64) -> Vec<u8> {
    let f = File::open(path).expect("reopen archive");
    let mut buf = vec![0u8; len as usize];
    f.read_exact_at(&mut buf, offset).expect("pread extent");
    buf
}

fn sha1_hex(bytes: &[u8]) -> String {
    use sha1::{Digest, Sha1};
    let mut h = Sha1::new();
    h.update(bytes);
    hex::encode(h.finalize())
}

fn loadavg() -> String {
    let mut s = String::new();
    File::open("/proc/loadavg")
        .and_then(|mut f| f.read_to_string(&mut s))
        .map(|_| ())
        .unwrap_or_default();
    s.split_whitespace().take(3).collect::<Vec<_>>().join(" ")
}

// ── impl 1 ──────────────────────────────────────────────────────────────────

/// `FastWriter` puts the caller's bytes, unaltered, at the extent it reported.
///
/// Asserts the bytes read back through a second handle, not the return value:
/// the extent has to *address* the payload, and the payload has to be
/// byte-identical (verbatim — no deflate, no framing, no length prefix).
///
/// Seen RED by `self.blobs.write_all_at(bytes, offset)?;` →
/// `self.blobs.write_all_at(&bytes[..bytes.len() / 2], offset)?;` in
/// `FastWriter::append`: observed
/// `assertion `left == right` failed: pack A verbatim on disk` at
/// `tests/archive_write.rs:107` — the extent still said 4108 B, `append` still
/// returned `Ok`, and the trailing 2054 B read back as a hole. Restored.
#[test]
fn fast_writer_stores_the_pack_verbatim_at_the_returned_extent() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    let w = FastWriter::create(&archive).unwrap();

    let a = pack_shaped(3, 4096);
    let b = pack_shaped(17, 200);
    let (ao, al) = w.append(&a).unwrap();
    let (bo, bl) = w.append(&b).unwrap();

    assert_eq!(al, a.len() as u64, "extent length is the input length");
    assert_eq!(bl, b.len() as u64);
    assert_eq!(bo, ao + al, "second append starts where the first ended");
    assert_eq!(read_back(&archive, ao, al), a, "pack A verbatim on disk");
    assert_eq!(read_back(&archive, bo, bl), b, "pack B verbatim on disk");
}

// ── impl 2 ──────────────────────────────────────────────────────────────────

/// `SafeWriter`'s journal row is **on disk** and **addresses the stored bytes**.
///
/// Both halves matter. The row is decoded out of the on-disk Arrow IPC stream
/// through a fresh handle, and the extent it carries is then used to `pread` the
/// payload back and compare it to what was pushed. A journal that recorded the
/// wrong offset would pass a row-count assertion and fail this one.
///
/// Seen RED by `journal_batch(offset, len)` → `journal_batch(offset + 1, len)`
/// in `SafeWriter::append`: observed
/// `assertion `left == right` failed: journal rows are the extents append
/// returned` at `tests/archive_write.rs:139` — every row was present and the
/// row *count* was still right; only the offsets were off by one. Restored.
#[test]
fn safe_writer_journal_row_is_on_disk_and_addresses_the_stored_bytes() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    let w = SafeWriter::create(&archive).unwrap();
    let jpath = w.journal_file();

    let packs: Vec<Vec<u8>> = vec![pack_shaped(1, 188), pack_shaped(9, 65536)];
    let mut extents = Vec::new();
    for p in &packs {
        extents.push(w.append(p).unwrap());
    }

    let rows = read_journal(&jpath).expect("decode journal");
    assert_eq!(rows.len(), packs.len(), "one journal row per append");
    assert_eq!(rows, extents, "journal rows are the extents append returned");
    for (p, (off, len)) in packs.iter().zip(&rows) {
        assert_eq!(
            &read_back(&archive, *off, *len),
            p,
            "the journal's extent addresses the payload"
        );
    }
}

/// The **BufWriter trap**: `flush()` before `sync_all()` is load-bearing.
///
/// `SafeWriter` writes its journal through a `BufWriter`, so the row lives in a
/// userspace `Vec` until `flush()`. `sync_all()` on its own
/// succeeds, fsyncs a real fd, and syncs **nothing that was just written**. This
/// guard reads the journal back through a separate handle — which can only see
/// bytes the kernel has — and demands the row be there.
///
/// Seen RED by deleting `j.writer.flush()...?;` (step 4) from
/// `SafeWriter::append`, leaving step 5's `sync_all()` in place: observed
/// `assertion `left == right` failed: the row must be past userspace before the
/// fsync / left: 0 / right: 1` at `tests/archive_write.rs:174`. `append`
/// returned `Ok`, the `fsync` succeeded, and the row was not on disk. Restored.
#[test]
fn journal_flush_before_sync_is_load_bearing() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    let w = SafeWriter::create(&archive).unwrap();
    let jpath = w.journal_file();

    let p = pack_shaped(2, 512);
    let extent = w.append(&p).unwrap();

    // A fresh handle sees only what reached the kernel.
    let rows = read_journal(&jpath).expect("decode journal");
    assert_eq!(
        rows.len(),
        1,
        "the row must be past userspace before the fsync"
    );
    assert_eq!(rows[0], extent);
}

/// **A reopen appends to the journal; it does not truncate it.**
///
/// The journal is the durable half of §13.12's `indexed` bit — a pack is
/// unabsorbed *iff* its extent is here and its rows are not in the index — so a
/// writer opened over an existing archive has to be able to read what the last
/// process acked. Three writers in succession, each closed before the next
/// opens, and every extent from all three must still decode out of the one file.
///
/// Asserted on applied output: the rows read back through a fresh handle *are*
/// the extents the three `append` calls returned, in order, and each one still
/// addresses its own payload in the blob file.
///
/// Seen RED by restoring `File::create(&path)` in `SafeWriter::create_inner`
/// (which is what it did before this guard existed): observed
/// `assertion `left == right` failed: a reopen dropped the earlier rows / left:
/// [(65560, 524)] / right: [(0, 4108), (4108, 61452), (65560, 524)]` — the bytes
/// of the first two packs were still on disk and nothing on disk pointed at them
/// any more. Restored.
///
/// Seen RED a second time by emitting the schema message on every open rather
/// than only for an empty file (`if already == 0` → unconditional): observed
/// `assertion `left == right` failed: a reopen dropped the earlier rows / left:
/// [(0, 4108)] / right: [(0, 4108), (4108, 61452), (65560, 524)]` — arrow's
/// `StreamReader` stops at the second schema message, so every row written after
/// the first reopen was silently invisible while the file kept growing.
#[test]
fn a_reopened_safe_writer_appends_to_the_journal_it_finds() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    let packs: Vec<Vec<u8>> = vec![
        pack_shaped(1, 4096),
        pack_shaped(2, 61440),
        pack_shaped(3, 512),
    ];

    let mut extents = Vec::new();
    let mut jpath = std::path::PathBuf::new();
    for p in &packs {
        // A new writer per pack: this is a process restart in every way that
        // matters to the file, which is the only thing the journal knows about.
        let w = SafeWriter::create(&archive).unwrap();
        jpath = w.journal_file();
        extents.push(w.append(p).unwrap());
        drop(w);
    }

    let rows = read_journal(&jpath).expect("decode journal");
    assert_eq!(rows, extents, "a reopen dropped the earlier rows");
    for (p, (off, len)) in packs.iter().zip(&rows) {
        assert_eq!(
            &read_back(&archive, *off, *len),
            p,
            "the journal's extent no longer addresses the payload it was written for"
        );
    }
}

/// **A store that has acked nothing is rebuilt, not fsynced.**
///
/// `SafeWriter::create` no longer barriers the schema message it writes into a
/// fresh journal — on gunnar's create path that was 8.9 ms of a 68.5 ms
/// push-create spent making durable a file that references nothing and that a
/// transactional create discards on failure anyway. The property the barrier
/// used to hold is held here instead, by checking on the way in: a journal whose
/// header is not this build's schema message, in a file no longer than one, was
/// interrupted before any row could exist behind it, and is written again.
///
/// Every torn length is tried, because "it works for the length I happened to
/// pick" is not the claim. The assertion is on applied output: a pack acked
/// through the repaired writer is addressable through a journal decoded from a
/// fresh handle.
///
/// Seen RED by dropping the repair (leaving only `if already == 0`): observed
/// `decode journal: Err(... Unable to get root as message ...)` at the first
/// torn length — the exact state the removed fsync used to prevent, left
/// unhandled.
#[test]
fn an_interrupted_schema_message_is_rewritten_rather_than_inherited() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    // The schema message this build writes, at its real length.
    let jpath = SafeWriter::journal_path(&archive);
    drop(SafeWriter::create(&archive).unwrap());
    let schema_len = std::fs::metadata(&jpath).unwrap().len();
    assert!(schema_len > 0, "a fresh journal must hold the schema message");

    for cut in 0..schema_len {
        let f = File::options().write(true).open(&jpath).unwrap();
        f.set_len(cut).unwrap();
        drop(f);

        // Both durable arms, alternating, because LAW 5 claims they are
        // indistinguishable on disk and one repair that only the safe arm knew
        // about is exactly how that would stop being true.
        let pack = pack_shaped(1, 4096);
        let (off, len) = if cut % 2 == 0 {
            let w = SafeWriter::create(&archive).unwrap();
            let e = w.append(&pack).unwrap();
            drop(w);
            e
        } else {
            match UringWriter::create(&archive) {
                Ok(w) => {
                    let e = w.append(&pack).unwrap();
                    drop(w);
                    e
                }
                // No ring on this kernel: the safe arm still has to hold.
                Err(_) => {
                    let w = SafeWriter::create(&archive).unwrap();
                    let e = w.append(&pack).unwrap();
                    drop(w);
                    e
                }
            }
        };

        let rows = read_journal(&jpath).expect("decode journal");
        assert_eq!(
            rows.last().copied(),
            Some((off, len)),
            "cut at {cut}: the ack after the repair is not in the journal"
        );
        assert_eq!(
            &read_back(&archive, off, len),
            &pack,
            "cut at {cut}: the journal's extent does not address the payload"
        );
        std::fs::remove_file(&jpath).ok();
        std::fs::remove_file(&archive).ok();
        drop(SafeWriter::create(&archive).unwrap());
    }
}

/// **The crash-ordering contract.** An interruption between the two fsyncs
/// leaves **orphan bytes**, never a dangling reference.
///
/// Injected at exactly the point a machine would die: after the blob `fsync`,
/// before the journal row. The on-disk state a real power loss leaves there is
/// byte-for-byte what this produces — `append` returns `Err` only because a Rust
/// function must return something.
///
/// Two assertions, and the second is the one that matters:
/// * the pack bytes ARE on disk at the extent that was reserved — orphan payload
///   the seal will drop;
/// * the journal names **no** extent — nothing points into a hole.
///
/// Seen RED by moving the journal block (steps 3–5) *above* the
/// `die_between_fsyncs` bail in `SafeWriter::append` — i.e. writing the row
/// first, the ordering this contract forbids: observed
/// `the journal must not reference bytes whose fsync never happened / journal
/// rows: [(0, 4108)]` at `tests/archive_write.rs:228`. A durable, live
/// reference to an extent whose bytes were never fsynced: the dangling
/// reference this ordering exists to make impossible. Restored.
#[test]
fn crash_between_fsyncs_leaves_orphan_bytes_not_a_dangling_reference() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    let w = SafeWriter::create_with_faults(&archive, Faults {
        die_between_fsyncs: true,
        ..Default::default()
    })
    .unwrap();
    let jpath = w.journal_file();

    let p = pack_shaped(5, 4096);
    let err = w.append(&p).unwrap_err();
    assert!(
        err.to_string().contains("injected crash"),
        "the fault fired: {err}"
    );
    drop(w);

    // Orphan bytes: the payload is on the platter at the reserved extent…
    assert_eq!(
        read_back(&archive, 0, p.len() as u64),
        p,
        "the fsynced blob bytes survived the crash"
    );
    // …and nothing references them.
    let rows = read_journal(&jpath).expect("decode journal");
    assert!(
        rows.is_empty(),
        "the journal must not reference bytes whose fsync never happened / journal rows: {rows:?}"
    );
}

// ── impl 3 ──────────────────────────────────────────────────────────────────

/// The io_uring chain lands the pack **and** its journal row.
///
/// One `io_uring_enter` carrying `Write → Fsync → WriteFixed → Fsync`, linked
/// with `IOSQE_IO_LINK`. If this kernel cannot run it the test **fails loudly**
/// rather than skipping — a silent skip is a guard that can never be red.
///
/// The journal is asserted through the same `read_journal` the `SafeWriter`
/// guard uses, which is the point: the io_uring arm writes the *same* Arrow IPC
/// journal format, built by hand into a registered buffer instead of through
/// arrow's `BufWriter`.
///
/// Seen RED by `.offset(journal_at)` → `.offset(journal_at + 8)` on the
/// `WriteFixed` SQE in `UringWriter::append` — same op, same byte count, wrong
/// place: observed `assertion `left == right` failed: one journal row per
/// append / left: 0 / right: 2` at `tests/archive_write.rs:278`. Both appends
/// returned `Ok`, the writer's own short-write check passed (224 of 224 B), the
/// pack bytes were on disk, and the journal decoded to nothing. Restored.
///
/// (A cruder break — deleting the `WriteFixed` SQE entirely — was tried first
/// and is *not* what is recorded here: `UringWriter`'s own short-journal-write
/// check caught it before the test could, so it proved the writer, not the
/// guard.)
#[test]
fn uring_writer_chain_lands_bytes_and_journal_row() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    let w = match UringWriter::create(&archive) {
        Ok(w) => w,
        Err(e) => panic!(
            "io_uring arm unavailable on this kernel — not skipping, this is a real failure: {e}"
        ),
    };
    let jpath = w.journal_file().to_path_buf();

    let packs = vec![pack_shaped(4, 200), pack_shaped(64, 1 << 20)];
    let mut extents = Vec::new();
    for p in &packs {
        extents.push(w.append(p).unwrap());
    }

    for (p, (off, len)) in packs.iter().zip(&extents) {
        assert_eq!(
            &read_back(&archive, *off, *len),
            p,
            "io_uring wrote the pack verbatim"
        );
    }
    let rows = read_journal(&jpath).expect("decode journal");
    assert_eq!(rows.len(), packs.len(), "one journal row per append");
    assert_eq!(rows, extents);
}

// ── the three arms agree on what they store ─────────────────────────────────

/// All three arms store **byte-identical** payload. They differ in durability,
/// never in content.
///
/// Seen RED by making `FastWriter::append` write `&bytes[1..]`: observed
/// `pread extent: Error { kind: UnexpectedEof, message: "failed to fill whole
/// buffer" }` at `tests/archive_write.rs:60` — the extent claimed 4108 B and
/// the file held 4107. Restored.
#[test]
fn all_three_arms_store_byte_identical_payload() {
    let dir = tempfile::tempdir().unwrap();
    let p = pack_shaped(11, 4096);

    let mut stored: Vec<(&'static str, Vec<u8>)> = Vec::new();
    for (name, make) in [
        ("FastWriter", 0u8),
        ("SafeWriter", 1),
        ("UringWriter", 2),
    ] {
        let archive = dir.path().join(format!("{name}.znippy"));
        let w: Box<dyn ArchiveWrite> = match make {
            0 => Box::new(FastWriter::create(&archive).unwrap()),
            1 => Box::new(SafeWriter::create(&archive).unwrap()),
            _ => Box::new(UringWriter::create(&archive).unwrap()),
        };
        assert_eq!(w.name(), name);
        let (off, len) = w.append(&p).unwrap();
        assert_eq!(len, p.len() as u64, "{name} stored a different length");
        stored.push((name, read_back(&archive, off, len)));
    }
    for (name, bytes) in &stored {
        assert_eq!(
            bytes, &p,
            "{name} stored something other than the pushed pack"
        );
    }
}

// ── the shared indexer ──────────────────────────────────────────────────────

/// The index tables are built **after** the bytes are down, and they carry rows
/// derived from the bytes — not echoes of the job that was submitted.
///
/// `pack_id`, `blob_offset` and `blob_size` came in on the channel; `object_count`,
/// `pack_version` and `pack_sha1` did **not** — they can only be produced by
/// reading the extent back out of the archive, which is the whole point of the
/// deferred index. So the assertion is on those three.
///
/// Seen RED by `h.update(&buf);` → `h.update(&buf[..1.min(buf.len())]);` in
/// `indexer::build_row`: observed `assertion `left == right` failed: pack_sha1
/// is a hash of the stored bytes / left: "511993d3c99719e38a6779073019dacd7178ddb9"
/// right: "8dc6cbb5a1129b90d8f44d307c492746aa1fcdb8"` at
/// `tests/archive_write.rs:375`. Five rows still appeared, with the right ids,
/// offsets, lengths and object counts. Restored.
#[test]
fn index_tables_are_built_after_the_bytes_and_carry_derived_rows() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    let w = SafeWriter::create(&archive).unwrap();
    let jpath = w.journal_file();
    let path = PushPath::new(Box::new(w), &archive, Some(jpath)).unwrap();

    let packs: Vec<Vec<u8>> = (0..5).map(|i| pack_shaped(100 + i, 1024 * (i as usize + 1))).collect();
    let mut ids = Vec::new();
    for p in &packs {
        ids.push(path.push_pack("acct-alpha", p).unwrap());
    }

    let idx = path.indexer("acct-alpha");
    idx.wait_caught_up();
    assert_eq!(idx.rows(), packs.len(), "one index row per pack");

    let tables = idx.tables();
    assert!(!tables.is_empty(), "index tables exist");
    let mut seen = 0usize;
    for t in &tables {
        let id = t.column(0).as_any().downcast_ref::<UInt64Array>().unwrap();
        let off = t.column(1).as_any().downcast_ref::<UInt64Array>().unwrap();
        let len = t.column(2).as_any().downcast_ref::<UInt64Array>().unwrap();
        let ver = t.column(3).as_any().downcast_ref::<UInt32Array>().unwrap();
        let cnt = t.column(4).as_any().downcast_ref::<UInt32Array>().unwrap();
        let sha = t.column(5).as_any().downcast_ref::<StringArray>().unwrap();
        for r in 0..t.num_rows() {
            let i = id.value(r) as usize;
            let (pid, (o, l)) = ids[i];
            assert_eq!(id.value(r), pid);
            assert_eq!(off.value(r), o);
            assert_eq!(len.value(r), l);
            // Derived — could only come from reading the stored bytes back.
            assert_eq!(ver.value(r), 2, "pack_version parsed from the header");
            assert_eq!(
                cnt.value(r),
                100 + i as u32,
                "object_count parsed from the header"
            );
            assert_eq!(
                sha.value(r),
                sha1_hex(&packs[i]),
                "pack_sha1 is a hash of the stored bytes"
            );
            seen += 1;
        }
    }
    assert_eq!(seen, packs.len());
    assert_eq!(loadavg().split(' ').count(), 3, "loadavg recorded: {}", loadavg());
}

/// A read that arrives **before** the index is ready is slower, never wrong.
///
/// Deterministic by construction: the extent is appended but no [`IndexJob`] is
/// submitted, so the indexed bit is provably unset. In that state
/// [`AccountIndexer::lookup`] must hand back the journal — the durable record
/// written on the ack path — and the bytes at the journal's extent must be the
/// pushed pack. Only after the job is submitted and drained may it answer
/// [`Lookup::Indexed`].
///
/// Seen RED by replacing the journal fallback in `AccountIndexer::lookup` with
/// a bare `Lookup::Unknowable`: observed
/// `pre-index read must fall back to scanning, got Unknowable` at
/// `tests/archive_write.rs:431`. Restored.
///
/// Recorded because it is the honest version of the story: the *first* break
/// tried here was deleting the `indexed.contains(&pack_id)` check that used to
/// sit in front of the row map, and the test **stayed green** — the two
/// structures could never disagree, so the check was decoration. It was removed
/// rather than guarded: the indexed bit is now membership in the published-row
/// map itself (LAW 5, fix by construction). What is left to break is the
/// fallback, and that does go red.
#[test]
fn read_before_the_index_falls_back_to_scanning_and_is_not_wrong() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    let w = SafeWriter::create(&archive).unwrap();
    let jpath = w.journal_file();

    let p = pack_shaped(7, 3000);
    let (off, len) = w.append(&p).unwrap();

    let idx = AccountIndexer::start(
        "acct-beta",
        Arc::new(File::open(&archive).unwrap()),
        Some(jpath),
    );

    // Nothing submitted: the indexed bit is unset, by construction.
    assert!(!idx.is_indexed(0));
    let before = idx.lookup(0);
    match &before {
        Lookup::ScanJournal(extents) => {
            assert!(
                extents.contains(&(off, len)),
                "the scan fallback must expose the extent: {extents:?}"
            );
            assert_eq!(
                read_back(&archive, off, len),
                p,
                "scanning the fallback extent yields the real pack"
            );
        }
        other => panic!("pre-index read must fall back to scanning, got {other:?}"),
    }
    assert!(
        matches!(before, Lookup::ScanJournal(_)),
        "pre-index read must not claim an index hit"
    );

    idx.submit(IndexJob {
        pack_id: 0,
        offset: off,
        len,
    })
    .unwrap();
    idx.wait_caught_up();

    match idx.lookup(0) {
        Lookup::Indexed(row) => {
            assert_eq!((row.offset, row.len), (off, len));
            assert_eq!(row.object_count, 7);
            assert_eq!(row.sha1, sha1_hex(&p));
        }
        other => panic!("after draining, the index must answer: {other:?}"),
    }
}

/// Idle indexers **sleep**. They do not poll and they do not spin.
///
/// One indexer per account only works if an idle account is free. Measured as
/// process CPU out of `/proc/self/stat` across 400 ms of wall time with 64
/// account indexers parked — a spin loop over `try_recv` would burn 64 cores'
/// worth of it.
///
/// Seen RED by replacing `let Ok(first) = rx.recv() else { return };` in
/// `index_worker` with a `loop { match rx.try_recv() { Ok(j) => break j, Empty
/// => continue, Disconnected => return } }` spin: observed
/// `64 idle indexers burned 12920 ms of CPU in 400 ms of wall time (loadavg
/// 6.39 1.91 1.05) — they are spinning, not sleeping` at
/// `tests/archive_write.rs:498`, against a 200 ms budget. 32× over, and the
/// box's 1-minute load went from ~0.3 to 6.39 while doing no work at all.
/// Restored to the blocking `recv()`.
#[test]
fn idle_indexers_sleep_they_do_not_spin() {
    fn cpu_ms() -> u64 {
        let mut s = String::new();
        File::open("/proc/self/stat")
            .unwrap()
            .read_to_string(&mut s)
            .unwrap();
        // utime and stime are fields 14 and 15, after the (possibly
        // paren-wrapped) comm field.
        let tail = &s[s.rfind(')').unwrap() + 1..];
        let f: Vec<&str> = tail.split_whitespace().collect();
        let ticks: u64 = f[11].parse::<u64>().unwrap() + f[12].parse::<u64>().unwrap();
        let hz = unsafe { libc::sysconf(libc::_SC_CLK_TCK) } as u64;
        ticks * 1000 / hz.max(1)
    }

    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    std::fs::write(&archive, b"").unwrap();
    let shared = Arc::new(File::open(&archive).unwrap());

    let idlers: Vec<AccountIndexer> = (0..64)
        .map(|i| AccountIndexer::start(&format!("acct-{i}"), shared.clone(), None))
        .collect();

    let before = cpu_ms();
    std::thread::sleep(std::time::Duration::from_millis(400));
    let burned = cpu_ms() - before;

    assert!(
        burned < 200,
        "64 idle indexers burned {burned} ms of CPU in 400 ms of wall time \
         (loadavg {}) — they are spinning, not sleeping",
        loadavg()
    );
    drop(idlers);
}

/// Accounts get **separate** indexers: one account's queue is not another's.
///
/// Asserts the isolation as applied state — each account's index holds exactly
/// its own packs and none of the other's, and the two `Arc`s are different
/// objects.
///
/// Seen RED by making `IndexerPool::indexer` key every account to the same
/// entry (`m.entry(account.to_string())` → `m.entry("shared".to_string())`):
/// observed `two accounts must not share one indexer` at
/// `tests/archive_write.rs:536` — the `Arc::ptr_eq` fired before the row counts
/// were even reached. Restored.
#[test]
fn accounts_do_not_share_an_indexer() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");
    let w = SafeWriter::create(&archive).unwrap();
    let jpath = w.journal_file();
    let path = PushPath::new(Box::new(w), &archive, Some(jpath)).unwrap();

    let mut one = Vec::new();
    let mut two = Vec::new();
    for i in 0..2 {
        one.push(path.push_pack("acct-one", &pack_shaped(1 + i, 300)).unwrap());
    }
    for i in 0..3 {
        two.push(path.push_pack("acct-two", &pack_shaped(50 + i, 700)).unwrap());
    }

    let a = path.indexer("acct-one");
    let b = path.indexer("acct-two");
    assert!(
        !Arc::ptr_eq(&a, &b),
        "two accounts must not share one indexer"
    );
    a.wait_caught_up();
    b.wait_caught_up();

    assert_eq!(a.rows(), 2, "acct-one indexed only its own packs");
    assert_eq!(b.rows(), 3, "acct-two indexed only its own packs");
    for (id, _) in &one {
        assert!(a.is_indexed(*id), "acct-one's bit set for {id}");
        assert!(!b.is_indexed(*id), "acct-two must not claim {id}");
    }
    for (id, _) in &two {
        assert!(b.is_indexed(*id));
        assert!(!a.is_indexed(*id));
    }
    assert_eq!(path.pool().accounts(), 2);
}

// ── the append-side gc ──────────────────────────────────────────────────────

/// **Orphan bytes are given back at the next open, and the live pack is not.**
///
/// The crash-ordering contract above leaves bytes on the platter that no extent
/// addresses (`crash_between_fsyncs_leaves_orphan_bytes_not_a_dangling_reference`),
/// and the append cursor is taken from the blob file's LENGTH — so before this
/// they were stepped over by the next append and unreachable for ever.
/// `SafeWriter::create_inner`'s own comment names the shape: "a reopen that
/// dropped them would leave their bytes unreferenced for ever".
///
/// Nothing referenced them, so nothing is lost by taking them back: a push whose
/// journal row never landed was never acked and the client sends it again.
///
/// Asserted on **applied output** — the file's length on disk and the live
/// pack's bytes read back through a fresh handle — never on a return value.
///
/// RED-FIRST: with `reclaim_orphan_tail` returning `len` unconditionally the
/// file stays at both packs' length and the assertion reads
/// "orphan bytes were not reclaimed".
#[test]
fn orphan_bytes_past_the_last_journal_row_are_reclaimed_at_the_next_open() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("repo.znippy");

    // A good pack, acked: bytes AND journal row.
    let good = pack_shaped(5, 4096);
    let (off, len) = {
        let w = SafeWriter::create(&archive).unwrap();
        let e = w.append(&good).unwrap();
        e
    };
    let after_good = std::fs::metadata(&archive).unwrap().len();
    assert_eq!(after_good, off + len, "the good pack is the whole file");

    // A second pack that dies between the two fsyncs: bytes on disk, no row.
    let orphan = pack_shaped(9, 8192);
    {
        let w = SafeWriter::create_with_faults(&archive, Faults {
            die_between_fsyncs: true,
            ..Default::default()
        })
        .unwrap();
        let err = w.append(&orphan).unwrap_err();
        assert!(err.to_string().contains("injected crash"), "the fault fired: {err}");
    }
    let with_orphan = std::fs::metadata(&archive).unwrap().len();
    assert!(
        with_orphan > after_good,
        "the orphan bytes are not on disk, so this proves nothing: {with_orphan} vs {after_good}"
    );

    // The reopen is the gc.
    drop(SafeWriter::create(&archive).unwrap());

    assert_eq!(
        std::fs::metadata(&archive).unwrap().len(),
        after_good,
        "orphan bytes were not reclaimed"
    );
    // And the pack that WAS acked is untouched, byte for byte.
    assert_eq!(
        read_back(&archive, off, len),
        good,
        "the reclaim cut into a live extent"
    );
    let rows = read_journal(&SafeWriter::journal_path(&archive)).expect("decode journal");
    assert_eq!(rows, vec![(off, len)], "the journal lost or gained a row: {rows:?}");
}

/// **An archive with no journal is left alone.** [`FastWriter`] keeps none, so
/// an absent journal is a normal arm and not evidence that anything is orphaned.
/// Truncating on it would delete every pack a `FastWriter` ever stored.
///
/// RED-FIRST: with the `!journal.exists()` guard removed, `read_journal` fails
/// on the missing path and `FastWriter::create` returns `Err` — the arm stops
/// opening at all.
#[test]
fn an_archive_with_no_journal_is_never_truncated() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("fast.znippy");
    let p = pack_shaped(4, 2048);
    let (off, len) = {
        let w = FastWriter::create(&archive).unwrap();
        w.append(&p).unwrap()
    };
    assert!(!SafeWriter::journal_path(&archive).exists(), "FastWriter wrote a journal");

    drop(FastWriter::create(&archive).unwrap());

    assert_eq!(
        std::fs::metadata(&archive).unwrap().len(),
        off + len,
        "a journal-less archive was truncated"
    );
    assert_eq!(read_back(&archive, off, len), p);
}

/// **A journal that reads as empty is not evidence either.** `read_journal`
/// answers `Ok(vec![])` both for a genuinely empty log and for one whose SCHEMA
/// message is torn, and those are indistinguishable from here. Truncating on
/// that reading would turn one bad message into the loss of every pack.
///
/// RED-FIRST: with the `None` arm truncating to 0 instead of returning `len`,
/// the whole archive is destroyed and the assertion reads "an unreadable
/// journal cost the archive its packs".
#[test]
fn an_unreadable_journal_leaves_the_blobs_alone() {
    let dir = tempfile::tempdir().unwrap();
    let archive = dir.path().join("torn.znippy");
    let p = pack_shaped(6, 4096);
    let (off, len) = {
        let w = SafeWriter::create(&archive).unwrap();
        w.append(&p).unwrap()
    };
    // Tear the journal down to nothing — the shape a crash in the schema
    // message leaves, and the shape an empty log has.
    std::fs::write(SafeWriter::journal_path(&archive), b"").unwrap();

    drop(SafeWriter::create(&archive).unwrap());

    assert_eq!(
        std::fs::metadata(&archive).unwrap().len(),
        off + len,
        "an unreadable journal cost the archive its packs"
    );
    assert_eq!(read_back(&archive, off, len), p, "the payload did not survive");
}