znippy_plugin_git/

store.rs

//! The storage functions. Nothing else is permitted here.
//!
//! **Twelve, then thirteen.** The count moved on 2026-08-10 when
//! [`GitOps::put_refs_cas`] landed — atomic batch compare-and-swap, the
//! `git push --atomic` primitive, which was previously *unrepresentable*:
//! `put_refs` is a batch without CAS, `update_ref` is CAS without a batch, and
//! `--atomic` is defined as both at once. It is a storage concern and it belongs
//! here. What does **not** belong here is the *reading* contract — decoded
//! reads, `HEAD`, negotiation, pack emission — which is [`crate::serve`].
//!
//! **ZNIPPY-GIT APACHE ARROW IPC IS LAW.** Storage is Arrow IPC in a znippy
//! archive. Not a packfile directory, not loose objects, not `gix-odb`, not a
//! filesystem layout borrowed from somebody else's model. gix is a codec for
//! framing bytes onto the wire and nothing more: it never owns a handle, never
//! decides where a byte lives, and never appears in a signature below.
//!
//! ---
//!
//! The twelve are the methods of [`GitOps`], implemented once, for
//! [`GitStore`] — the handle and the trait live in [`crate::git_ops`], and
//! everything each method reaches for was already built and already measured:
//!
//! | this file calls | which is | measured in |
//! |---|---|---|
//! | `PushPath` → `SafeWriter` | blob fsync, then the journal row | `archive_write` |
//! | `ObjectReadStack` | stree → Arrow → redb tail | `read_stack` |
//! | `RefLog` | one Arrow IPC frame per push | `refs` / `pushlog` |
//! | `pack_walk` + `resolve` | the split and the oids | `pack_walk` / `resolve` |
//! | `NewGeneration` → `compact_archive` | base znippy's compaction | `gc` |
//!
//! No method below invents storage, an index, a durability contract or a
//! concurrency mechanism. If one looks like it does, that is the bug.

use std::path::Path;

use anyhow::{anyhow, bail, Context, Result};
use znippy_common::ReservedSection;

use crate::gc::GcReport;
use crate::git_ops::{lookup_path, GitOps, GitStore, LookupPath, RefRow, Stored, TxId};
use crate::index_layout::ObjectIndex;
use crate::pack_walk::walk;
use crate::refs::RefUpdate;
use git_storage_trait::{Observed, RefCas, RefRejection, RefTarget};

/// An object id, borrowed.
pub type Oid<'a> = &'a [u8];

/// A byte range inside the archive: `(offset, len)`.
pub type Extent = (u64, u64);

impl<S: ObjectIndex + 'static> GitStore<S> {
    /// **The one body behind [`GitOps::put_pack`] and
    /// [`GitOps::put_pack_observed`].**
    ///
    /// Private and inherent rather than a third trait method: the difference
    /// between the two is one `Option`, and two copies of the ack path — the
    /// closure check, the verbatim append, the journal ordering, the queued
    /// index job — is exactly the twinning that ends with one of them quietly
    /// missing a step. The steps are numbered on [`GitOps::put_pack`]'s doc,
    /// which is where a reader looking for *what a push does* will arrive.
    fn put_pack_watching(
        &self,
        bytes: &[u8],
        progress: Option<&dyn git_storage_trait::PackProgress>,
    ) -> Result<TxId> {
        if bytes.is_empty() {
            bail!("an empty push is not a pack");
        }
        // (1) + (2) — one walk, and the check falls out of it. The listener,
        // when there is one, rides that same walk: `walk` is `walk_observed`
        // with a no-op, so these two arms are one loop and not two.
        let walked = match progress {
            Some(sink) => crate::pack_walk::walk_observed(
                bytes,
                self.hash_kind().oid_len(),
                &mut |done, total| sink.indexed(done, total),
            ),
            None => walk(bytes, self.hash_kind().oid_len()),
        }
        .context("the pushed pack could not be split")?;
        self.external_bases_exist(bytes, &walked)?;

        // (3) + (4) — verbatim, blob fsync, journal row, journal fsync.
        // (5) — and the pack-level index job onto the account's channel.
        let (pack_id, extent) = self
            .push_path()
            .push_pack(self.account(), bytes)
            .context("storing the pack verbatim")?;

        // The object-level index work: 24 bytes of extent, queued, drained by
        // `absorb_pending` off this path. Until it runs the pack is un-indexed
        // and every read falls back rather than answering absent (§13.12).
        self.queue(pack_id, extent)?;

        Ok(TxId {
            pack_id: Some(pack_id),
            extent: Some(extent),
            push_seq: None,
        })
    }
}

impl<S: ObjectIndex + 'static> GitOps for GitStore<S> {
    // ── STORE ───────────────────────────────────────────────────────────────

    /// One push, and the order inside it is the contract.
    ///
    /// 1. the pack's bytes, durable — [`put_pack`](GitOps::put_pack)
    /// 2. **then** the refs that point into them — [`put_refs`](GitOps::put_refs)
    ///
    /// Not interchangeable, and it is the same argument `SafeWriter` makes one
    /// level down about the blob and its journal row: a crash between the two
    /// leaves objects nobody points at, which a GC reclaims. The reverse order
    /// leaves a **ref pointing at objects that are not there**, which is a
    /// corrupt repository that no later pass can repair.
    fn put(&self, pack: &[u8], refs: &[RefUpdate]) -> Result<TxId> {
        let mut tx = if pack.is_empty() {
            TxId::default()
        } else {
            self.put_pack(pack)?
        };
        if !refs.is_empty() {
            tx.push_seq = self.put_refs(refs)?.push_seq;
        }
        Ok(tx)
    }

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// **The ack path.** Walk, check, store verbatim, ack, then queue the index.
    ///
    /// 1. **Walk every entry** ([`crate::pack_walk`]). Not optional and not an
    ///    extra pass: a pack entry has no length field, so finding the entry
    ///    boundaries *is* how the pack gets split at all.
    /// 2. **The closure check, free from that walk** (§13.7). Every `OFS_DELTA`
    ///    base must land on an entry boundary in this pack — answered from the
    ///    boundary set the walk just produced, consulting no index. Only a
    ///    `REF_DELTA` base that is *outside* the pack is looked up, and that one
    ///    reads `objects.oid` and nothing else, exactly as §13's table says
    ///    receive-pack does. A pack that fails is refused **before** a byte is
    ///    stored.
    /// 3. **The bytes, verbatim.** `SafeWriter::append` writes the caller's
    ///    buffer at its own address — no re-compression, no re-encoding, no
    ///    re-framing, not one copy in userspace. The client's own deflate and
    ///    every delta chain survive, which is what makes a later clone a
    ///    byte-range copy instead of a re-pack (§14).
    /// 4. **Durable before returning**: the blob is fsynced, *then* the journal
    ///    row that references it, then that is fsynced. znippy's `hot.rs`
    ///    ordering, so a crash between the two leaves orphan bytes nobody points
    ///    at rather than a dangling reference.
    /// 5. **Then** the index job goes on the channel and this returns. Nothing
    ///    that could be done later is done here: no oid is computed, no table is
    ///    built, no chain is resolved.
    fn put_pack(&self, bytes: &[u8]) -> Result<TxId> {
        self.put_pack_watching(bytes, None)
    }

    /// [`put_pack`](GitOps::put_pack), publishing the entry count as step (1)
    /// produces it.
    ///
    /// **The count comes out of the walk that was already happening.** Step (1)
    /// above is not optional and not an extra pass — it is how the pack gets
    /// split — so the number of entries taken in is a fact this method already
    /// has, once per object, and until now it simply had nobody to tell. That
    /// is the whole of this change: no second decoder, no second inflate, and
    /// the `put_pack` above is this method with no listener, so there is still
    /// one implementation of the walk-check-store sequence (LAW 5).
    ///
    /// # What the numbers mean, and where they stop
    ///
    /// `done` reaches `total` when the **walk** finishes, which is before the
    /// bytes are durable and long before the indexer has absorbed them. That is
    /// the honest boundary: this store's ack path is deliberately "walk, check,
    /// append, queue" (§13.6-8), so there is no later per-object event to count.
    /// A caller wanting "how much is on disk" is asking for
    /// [`GitOps::put_pack`]'s return, which is what it has always been.
    fn put_pack_observed(
        &self,
        bytes: &[u8],
        progress: &dyn git_storage_trait::PackProgress,
    ) -> Result<TxId> {
        self.put_pack_watching(bytes, Some(progress))
    }

    /// This store's answer is `true`, and the conformance suite holds it to it.
    fn publishes_pack_progress(&self) -> bool {
        true
    }

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// One ref transaction: one Arrow IPC frame, fsynced.
    ///
    /// The frame boundary *is* the transaction (see [`crate::pushlog`]) — three
    /// branches in one push are three rows in one batch, and they either all land
    /// or none do. There is no lock file and no second journal.
    ///
    /// **Every target is checked against the index first.** A ref that points at
    /// an object the repository does not have is a corrupt repository, and it is
    /// refused here rather than written and discovered later. A deletion has no
    /// target and is not checked.
    fn put_refs(&self, updates: &[RefUpdate]) -> Result<TxId> {
        if updates.is_empty() {
            bail!("an empty ref update is not a transaction");
        }
        for u in updates {
            for (what, oid_hex) in [("target", &u.target), ("peeled", &u.peeled)] {
                let Some(hex_oid) = oid_hex else { continue };
                let raw = hex::decode(hex_oid)
                    .map_err(|e| anyhow!("{}'s {what} `{hex_oid}` is not hex: {e}", u.name))?;
                if !self.has(&raw)? {
                    bail!(
                        "{} would point at {hex_oid}, which this repository does not have — the \
                         ref update is refused rather than left dangling",
                        u.name
                    );
                }
            }
        }
        let push_seq = self.ref_log().push(updates)?;
        Ok(TxId {
            pack_id: None,
            extent: None,
            push_seq: Some(push_seq),
        })
    }

    // ── READ ────────────────────────────────────────────────────────────────

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// The object's **stored** bytes: one index lookup and one `pread` of the
    /// extent, with no decode of any kind in between.
    ///
    /// What comes back is the pack entry exactly as the client sent it, which for
    /// a delta is a delta. [`Stored::obj_type`] says which, so the bytes cannot
    /// be mistaken for the object's content — §14 makes the verbatim bytes the
    /// truth and the resolved object a derived cache, and that cache does not
    /// exist yet, so handing back delta bytes labelled "the object" would be the
    /// one thing this crate refuses everywhere: a wrong answer where an honest
    /// one was available.
    fn get(&self, oid: Oid<'_>) -> Result<Option<Stored>> {
        let Some(row) = self.lookup_one(oid)? else {
            return Ok(None);
        };
        let bytes = self.read_extent(row.offset, row.len)?;
        Ok(Some(Stored {
            obj_type: row.obj_type,
            uncompressed_size: row.uncompressed_size,
            extent: (row.offset, row.len),
            bytes,
        }))
    }

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// The negotiation call: `have` sends up to a thousand of these and **most of
    /// them miss**. It reads `objects.oid` and no other column.
    ///
    /// It takes the **serial** path, not `lookup_batch` with one element: that is
    /// 1.7× slower (818 ns against 482), and [`lookup_path`] is where that
    /// decision is written down.
    ///
    /// # Why this returns `Result<bool>` and not `bool`
    ///
    /// The signature changed deliberately. A store with a pack whose bytes are
    /// durable but whose objects are not indexed yet cannot answer "no" — the
    /// object may be in that pack. So it absorbs the pack first and, if that
    /// fails, **says so**. A `bool` could only have lied, and a wrong "absent"
    /// during negotiation makes a client send nothing and lose data.
    fn has(&self, oid: Oid<'_>) -> Result<bool> {
        Ok(self.lookup_one(oid)?.is_some())
    }

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// The object's **post-resolution** size — what it inflates to once its delta
    /// chain is applied. That is the fact git's own `.idx` and `.rev` together
    /// cannot answer, and it is why the quota gate is index-only here.
    fn size(&self, oid: Oid<'_>) -> Result<Option<u64>> {
        Ok(self.lookup_one(oid)?.map(|r| r.uncompressed_size))
    }

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// **The wire path**, and batch by construction: for each oid the two facts a
    /// clone needs — `offset` and `len` — and no others, so a byte-range copy out
    /// of the verbatim pack can start immediately.
    ///
    /// `out[i]` answers `oids[i]`. One oid takes the serial path
    /// ([`lookup_path`]); the batch path saturates at
    /// [`BATCH_SATURATES_AT`](crate::git_ops::BATCH_SATURATES_AT) oids, so a
    /// larger batch is passed through whole rather than split — splitting would
    /// cost a pass and buy nothing.
    fn extents(&self, oids: &[Oid<'_>]) -> Result<Vec<Option<Extent>>> {
        match lookup_path(oids.len()) {
            LookupPath::Serial => Ok(vec![self.lookup_one(oids[0])?.map(|r| (r.offset, r.len))]),
            LookupPath::Batch => Ok(self
                .lookup_batch_never_wait(oids)?
                .into_iter()
                .map(|r| r.map(|r| (r.offset, r.len)))
                .collect()),
        }
    }

    // ── REFS ────────────────────────────────────────────────────────────────

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// The whole ref namespace, tags peeled — the ref advertisement, and
    /// `ls-refs` after a prefix filter. Name-sorted, because the log folds into a
    /// `BTreeMap` and `ls-refs` wants a prefix scan.
    ///
    /// The cost is a scan of a structure sized by **pushes**, not by repository
    /// size.
    ///
    /// # `HEAD` is not in here, and that is the contract
    ///
    /// **Changed 2026-08-10, and it is a fix rather than a preference.** The
    /// other backend's `iter()` *"walks `refs/` (loose and packed) and
    /// deliberately excludes the pseudo-refs such as `HEAD`, which is exactly the
    /// contract every other backend honours"*. This one did not, so the two arms
    /// disagreed about whether `HEAD` is a row — and the conformance suite could
    /// not see it, because it never exercised `HEAD` at all.
    ///
    /// The reason `HEAD` cannot be a row is a type constraint, not taste: a name
    /// type that admits `HEAD` also admits `MERGE_HEAD` and `FETCH_HEAD`, so a
    /// row stream carrying pseudo-refs means either widening the name type or
    /// filtering at every consumer. It gets an accessor pair instead —
    /// [`GitServe::head`](crate::serve::GitServe::head) and
    /// [`set_head`](crate::serve::GitServe::set_head) — and this is the one
    /// filter, in the one place.
    ///
    /// Nothing else changes: the ref log still *stores* `HEAD` (a push writes it
    /// like any other row), `live_set` still reaches it because it reads
    /// `ref_state` directly, and the filter is one `!=`.
    fn refs(&self) -> Result<Vec<RefRow>> {
        let mut out = Vec::new();
        for (name, state) in self
            .ref_state()?
            .into_iter()
            .filter(|(name, _)| name != crate::serve::HEAD)
        {
            let decode = |h: &Option<String>| -> Result<Option<Vec<u8>>> {
                match h {
                    Some(h) => {
                        Ok(Some(hex::decode(h).map_err(|e| {
                            anyhow!("ref {name}: `{h}` is not a hex oid: {e}")
                        })?))
                    }
                    None => Ok(None),
                }
            };
            out.push(RefRow {
                oid: decode(&state.target)?,
                peeled: decode(&state.peeled)?,
                symref_target: state.symref_target.clone(),
                name,
            });
        }
        Ok(out)
    }

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// **Compare-and-swap**, in the same log a push writes its refs to — one
    /// mechanism, not a second one for single updates.
    ///
    /// `old` is what the caller believes the ref is: `None` means *it must not
    /// exist* (a create), `Some(oid)` means *it must be exactly this*. `new` of
    /// `None` deletes. A mismatch names both values and **writes nothing**.
    ///
    /// The read-compare-append is serialised on the store's ref gate. Without it
    /// two CAS calls could both read the old value and both append, and the
    /// second would silently overwrite an update it had compared against
    /// successfully — the classic lost update, and the only thing a CAS is for.
    fn update_ref(&self, name: &str, old: Option<Oid<'_>>, new: Option<Oid<'_>>) -> Result<TxId> {
        let _gate = self
            .ref_gate()
            .lock()
            .map_err(|_| anyhow!("the ref gate is poisoned"))?;

        // One keyed read under the gate, not a clone of the whole namespace:
        // this CAS compares and swaps exactly `name`.
        let current = self.ref_get(name)?;
        let actual: Option<Vec<u8>> = match current.as_ref().and_then(|s| s.target.as_deref()) {
            Some(h) => Some(
                hex::decode(h)
                    .map_err(|e| anyhow!("ref {name} holds `{h}`, not a hex oid: {e}"))?,
            ),
            None => None,
        };
        if actual.as_deref() != old {
            // ── TYPED, like `put_refs_cas` three hundred lines down ─────────
            //
            // This raised a bare `bail!` while its own sibling raised
            // `RefRejection::Cas`, so a caller had a string and no way to tell
            // a lost race from a broken disk. `gunnar-wire`'s receive-pack
            // says so at the call site: "A LOST COMPARE-AND-SWAP AND A BROKEN
            // DISK ARE THE SAME STRING HERE … Nothing may recover the
            // distinction by matching on this text; the fix is a typed
            // rejection on the contract."
            //
            // The contract already had one. It was this method that did not
            // use it. The `Display` text is unchanged — `RefRejection::Cas`
            // renders the same sentence — so nothing that reads the message
            // moves, and a caller that downcasts now gets `name`, `expected`
            // and `actual` as values instead of parsing prose.
            return Err(anyhow::Error::new(RefRejection::Cas {
                name: name.to_string(),
                expected: match old {
                    None => Observed::Nothing,
                    Some(o) => Observed::oid(o),
                },
                actual: match actual.as_deref() {
                    None => Observed::Nothing,
                    Some(a) => Observed::oid(a),
                },
            }));
        }

        let update = match new {
            Some(n) => RefUpdate::set(name, hex::encode(n)),
            None => RefUpdate::delete(name),
        };
        // Through `put_refs`, so the existence check and the log format are the
        // same code a push uses (LAW 5) — a CAS cannot create a dangling ref that
        // a push would have been refused for.
        if new.is_some() {
            self.put_refs(&[update])
        } else {
            let push_seq = self.ref_log().push(&[update])?;
            Ok(TxId {
                pack_id: None,
                extent: None,
                push_seq: Some(push_seq),
            })
        }
    }

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// **`git push --atomic`: every edit or none**, in one Arrow IPC frame.
    ///
    /// The two halves come from different places and neither is reimplemented:
    ///
    /// * **the batch atomicity is the frame.** [`crate::pushlog`] makes the frame
    ///   boundary the transaction — three branches in one push are three rows in
    ///   one batch and they either all land or none do. There is no lock file and
    ///   no second journal, so there is no partial-apply state to recover from;
    /// * **the compare half is checked here, against one snapshot**, before a
    ///   single [`RefUpdate`] is built.
    ///
    /// It goes out through [`put_refs`](GitOps::put_refs) rather than straight to
    /// the log, so the dangling-target refusal and the frame format are the same
    /// code a push uses (LAW 5): an atomic batch cannot create a ref pointing at
    /// an object the repository does not have, which a second writer here would
    /// eventually have allowed.
    ///
    /// # `S-023`, and why the check is a whole pass of its own
    ///
    /// Every expectation is evaluated **before any of them is applied**, and the
    /// loop deliberately does not fuse with the one that builds the updates. The
    /// defect it exists for is gix's: a backend that short-circuits an edge whose
    /// new value already equals the current one never evaluates the expectation
    /// the caller wrote, so an `old: None` — *must not exist* — becomes a silent
    /// success and *"exactly one creator wins"* stops being true. This log has no
    /// such short-circuit, but the ordering is what makes that irrelevant rather
    /// than lucky, and a future optimisation that adds one cannot break it from
    /// here.
    ///
    /// # What this arm cannot raise, stated rather than hidden
    ///
    /// **[`RefRejection::Locked`] never comes out of this backend.** The
    /// read-compare-append is serialised on the store's ref gate, which a second
    /// writer *blocks* on rather than failing against — so contention here is a
    /// wait, never a rejection. The variant is in the contract because the gix
    /// arm, which takes real per-ref lock files, raises it. A poisoned gate stays
    /// an ordinary error: it is a fault, not the transient thing a caller retries.
    fn put_refs_cas(&self, edits: &[RefCas<'_>]) -> Result<TxId> {
        // Not an error. A deletions-free push that had nothing to apply calls
        // this, and `put_refs` refuses an empty batch — rightly, since an empty
        // ref update is not a transaction. An empty *atomic* batch is a no-op
        // that succeeded, and saying so here is what keeps the special case out
        // of every call site.
        if edits.is_empty() {
            return Ok(TxId::default());
        }

        let _gate = self
            .ref_gate()
            .lock()
            .map_err(|_| anyhow!("the ref gate is poisoned"))?;

        // ONE read of the namespace for the whole check, so every expectation is
        // compared against one instant rather than against a namespace that may
        // move between them.
        let current = self.ref_state()?;
        let observed = |name: &str| -> Observed {
            match current.get(name) {
                None => Observed::Nothing,
                Some(s) => match (&s.symref_target, &s.target) {
                    (Some(points_to), _) => {
                        Observed::Value(RefTarget::Symbolic(points_to.clone()))
                    }
                    (None, Some(h)) => match hex::decode(h) {
                        Ok(raw) => Observed::Value(RefTarget::Object(raw)),
                        // Not dropped into `Nothing`: a ref that exists and
                        // cannot be read is not a ref that was absent, and
                        // reporting it as absent tells the pushing client its
                        // create is free when it is not.
                        Err(e) => Observed::Unreadable(format!("`{h}` is not a hex oid: {e}")),
                    },
                    (None, None) => Observed::Unreadable(
                        "the ref log holds a row naming neither an object nor another ref".into(),
                    ),
                },
            }
        };

        // S-023: the whole check, before anything is applied.
        for e in edits {
            let expected = match e.old {
                None => Observed::Nothing,
                Some(o) => Observed::oid(o),
            };
            let actual = observed(&e.name);
            if actual != expected {
                return Err(anyhow::Error::new(RefRejection::Cas {
                    name: e.name.clone(),
                    expected,
                    actual,
                }));
            }
        }

        let updates: Vec<RefUpdate> = edits
            .iter()
            .map(|e| match e.new {
                // The peel rides through `put_refs` into the same ref row, so a
                // tag ingested this way advertises with zero object reads
                // (`RefCas::peeled`, gunnar S-080). It is carried only on a set;
                // a delete has no row to peel.
                Some(n) => {
                    let update = RefUpdate::set(e.name.clone(), hex::encode(n));
                    match &e.peeled {
                        Some(p) => update.with_peeled(hex::encode(p)),
                        None => update,
                    }
                }
                None => RefUpdate::delete(e.name.clone()),
            })
            .collect();
        self.put_refs(&updates)
    }

    // ── GRAPH ───────────────────────────────────────────────────────────────

    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// **Selection: `want` minus `have`, as an `andnot` of roaring bitmaps.**
    ///
    /// Every returned oid is an object — commit, tree *and* blob — because the
    /// bitmaps are built by [`crate::reach::build_reach`], which walks the trees.
    /// A `want` that names a commit contributes that commit's whole closure; a
    /// `want` that names a tag or a blob contributes itself.
    ///
    /// A `have` this repository does not know contributes nothing: the safe
    /// direction is to send more, never less.
    ///
    /// The bitmaps are over the **store's own ordinal space**, rebuilt by the same
    /// fold that rebuilds them, never over the Arrow projection's ordinals — an
    /// [`crate::index_layout::IndexRow::ordinal`] is a row address within one
    /// projection generation, so a bitmap over those would address different
    /// objects after any rebuild, silently.
    ///
    /// # The per-oid `Vec` is paid HERE and nowhere else on the serving path
    ///
    /// The answer is computed flat (see
    /// [`GitStore::reachable_raw`](crate::git_ops::GitStore::reachable_raw));
    /// this splits it back out because the eleven say `Vec<Vec<u8>>` and this
    /// method's callers are maintenance ones — a GC live set, a conformance
    /// harness — not the serving path. `crate::serve`'s `select` and
    /// `emit_pack` take the flat form directly.
    fn reachable(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> Result<Vec<Vec<u8>>> {
        Ok(self
            .reachable_raw(want, have)?
            .iter()
            .map(<[u8]>::to_vec)
            .collect())
    }

    // ── MAINT ───────────────────────────────────────────────────────────────

    /// Garbage collection, in the order §13.20 fixes:
    ///
    /// 1. **compute reachability** — every object reachable from every ref, over
    ///    the same bitmaps [`reachable`](GitOps::reachable) uses;
    /// 2. **drop the dead rows from the index** — the one operation that is not
    ///    append-only, and the projection is rebuilt inside it;
    /// 3. **then** base znippy's compaction, through [`crate::gc::NewGeneration`]
    ///    (link, compact, verify, rename, unlink last).
    ///
    /// That order is why **base znippy needs no new method**: by the time
    /// `compact_archive` runs, "live" already means what git means.
    ///
    /// A repository with no ref pointing at anything is refused rather than
    /// emptied.
    ///
    /// **Appended, not reworded: step 1b, the journal.** Between the live set and
    /// the drop there is now one more durable act. §13.12's `indexed` bit is
    /// derived on open as *extent in the journal, rows not in the index*, and
    /// dropping **every** row of a pack produces exactly that state — so before
    /// this existed the next open re-queued the pack and every object this GC had
    /// just decided was dead came back.
    /// [`GitStore::retire_dead_packs`](crate::git_ops::GitStore::retire_dead_packs)
    /// appends a tombstone naming each all-dead pack, and it runs **before**
    /// `drop_dead_rows` on purpose: killed before it, the rows are still there
    /// and the GC simply did not happen; killed after it, the pack can never be
    /// re-queued whether the rows went or not. A partly dead pack keeps rows and
    /// is never tombstoned, so nothing about it changes.
    ///
    /// Step 2 also refolds what the two tables feed — the commit graph, the tree
    /// payloads, the ordinal space, the bitmaps — because a derivation that still
    /// names a dropped oid is wrong rather than stale. That is inside
    /// `drop_dead_rows`, so any caller of it gets it.
    fn gc(&self) -> Result<GcReport> {
        self.absorb_pending()?;
        // **What the GC may touch is fixed BEFORE reachability is computed.**
        // `live_set` walks from the refs as they are now; a push that lands
        // after this point appends rows and a pack that no ref names yet, and
        // to `live` those objects are simply unknown. The old code dropped
        // what was not live — so it dropped them. Two snapshots taken first:
        // the rows that exist (`seen`) and the journal's end (`acked_end`).
        // Dead is `seen − live`; a pack is retirable only if it was acked
        // before `acked_end`. A row or pack that arrives during the GC is in
        // neither and survives, whatever the refs say about it yet.
        let seen = self.index().oids_in_order()?;
        let acked_end = self.acked_end()?;
        let live = self.live_set()?;
        let dead = crate::git_ops::dead_oids(seen, &live);
        let retired = self.retire_dead_packs(&live, acked_end)?;
        let before = self.index().len() as u64;
        let dropped = self.drop_dead_rows(&dead)?;
        let after = self.index().len() as u64;
        if before.saturating_sub(after) != dropped || after > before {
            bail!(
                "the index dropped {dropped} rows but went from {before} to {after} — refusing to \
                 compact an archive whose index does not agree with itself"
            );
        }
        let mut report = self.gc_arm().run(self.archive_path()).with_context(|| {
            format!(
                "compacting {} after dropping {dropped} dead rows",
                self.archive_path().display()
            )
        })?;
        report.retired_packs = retired.len() as u64;
        Ok(report)
    }
}

impl<S: ObjectIndex + 'static> GitStore<S> {
    /// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
    ///
    /// Fold the live logs into the reserved Arrow sections an archive carries:
    /// `__gunnar_refs__`, `__gunnar_graph__`, `__gunnar_reach__`.
    ///
    /// **The twelfth method, and the one that is NOT on [`GitOps`]:** it returns
    /// `znippy_common::ReservedSection` (Arrow `RecordBatch` payloads), which a
    /// gix backend has no analog for — so it is inherent on the concrete store,
    /// which is how gunnar already calls it (never through the trait).
    ///
    /// It absorbs first, because a section built while a pack is still
    /// un-indexed would be **silently incomplete**, and a silently incomplete
    /// index is worse than none (the same argument [`crate::lib`] makes about not
    /// wiring `GitIndexBuilder` into the CLI). Every generation number in the
    /// graph section is recomputed by that fold.
    ///
    /// **…and then it writes the archive.** Until 2026-08-08 it only *returned*
    /// the sections and nothing on any path created
    /// [`archive_path`](GitStore::archive_path) at all — so `gc()`'s last step
    /// died on `stat repository.znippy: No such file or directory`, having
    /// already done its first four. Both [`Gc`](crate::gc::Gc) implementations
    /// compact an archive that exists; **this** is what makes one exist.
    /// [`seal_generation_zero`](crate::archive_write::seal_generation_zero) has
    /// the layout argument.
    ///
    /// The sections are still returned, and they are the same values that were
    /// sealed rather than a second derivation of them — `ReservedSection` is
    /// `Clone` for exactly that reason.
    pub fn seal(&self) -> Result<Vec<ReservedSection>> {
        self.absorb_pending()?;
        self.index().rebuild()?;
        let sections = self.reserved_sections()?;
        self.seal_archive(sections.clone()).with_context(|| {
            format!("sealing generation 0 at {}", self.archive_path().display())
        })?;
        Ok(sections)
    }
}

/// CLAUDE FUCKED UP AND MUST NOT FUCK UP AGAIN. ZNIPP-GIT APACHE ARROR IPC IS LAW
///
/// Compact `src` into a **named** destination, leaving `src` untouched.
///
/// The file-level primitive behind [`GitOps::gc`]'s third step, for a caller that
/// wants to choose the name rather than take
/// [`crate::gc::next_generation`]'s. Three existing calls and no fourth
/// mechanism:
///
/// 1. `hard_link(src, dst)` — a second name for the same inode. **Not one byte is
///    copied**, and `src` keeps pointing at the original for the whole run.
/// 2. `compact_archive(dst)` — base znippy's own compaction, verbatim, against
///    the new name.
/// 3. read every entry back through the ordinary reader — the same verification
///    [`crate::gc::NewGeneration`] gates on, so a `dst` that does not read back
///    is removed and the error says so.
///
/// `dst` must not exist. Silently compacting over a file is how a generation gets
/// lost.
pub fn compact(src: &Path, dst: &Path) -> Result<()> {
    if dst.exists() {
        bail!(
            "{} already exists — refusing to compact over it",
            dst.display()
        );
    }
    std::fs::hard_link(src, dst).with_context(|| {
        format!(
            "hard-linking {} to {} — compaction runs against a second name for the same inode, \
             which needs both on one filesystem",
            src.display(),
            dst.display()
        )
    })?;
    if let Err(e) = znippy_common::compact_archive(dst) {
        let _ = std::fs::remove_file(dst);
        return Err(e.context(format!(
            "compacting {} into {}",
            src.display(),
            dst.display()
        )));
    }
    if let Err(e) = crate::gc::read_back_every_entry(dst) {
        let _ = std::fs::remove_file(dst);
        return Err(e.context(format!(
            "{} did not read back after compaction — it was removed and {} is untouched",
            dst.display(),
            src.display()
        )));
    }
    Ok(())
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::git_ops::BATCH_SATURATES_AT;
    use crate::index_layout::ObjType;
    use crate::object::{canonical, GitHashKind, GitObjectKind};
    use crate::resolve::{resolve, NoBases};
    use std::path::PathBuf;
    use std::time::Instant;

    /// A store directory for one test.
    ///
    /// It returns a `PathBuf` and not a `tempfile::TempDir`, so **nothing
    /// deletes it when the test ends** — 82 call sites take it as a path and
    /// several deliberately outlive a store to reopen it, which is what the
    /// borrowed form would forbid. The directories are therefore swept on the
    /// way in instead of dropped on the way out; see [`sweep_dead_runs`].
    pub(crate) fn tmpdir(tag: &str) -> PathBuf {
        sweep_dead_runs();
        let d = std::env::temp_dir().join(format!(
            "znippy-git-store-{tag}-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_nanos())
                .unwrap_or(0)
        ));
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    /// **Delete the store directories left behind by test runs that are over.**
    ///
    /// MEASURED on oden 2026-08-11: **1 088 of these directories, 96 GB**, and
    /// `std::env::temp_dir()` on that box is `/tmp`, which is a `tmpfs` — so
    /// that was 96 GB of *RAM* held by test runs that had exited days earlier,
    /// on the shared machine every performance figure in this crate is taken
    /// on. Each one is a whole store, and one with an exploded table is ~180 MB.
    ///
    /// Two conditions, both required, because the only way this can do harm is
    /// by deleting a directory a running test still wants:
    ///
    /// 1. **The pid in the name is not a live process.** The name has carried
    ///    the pid since it was written; nothing consulted it until now.
    /// 2. **The directory has not been touched for an hour.** Redundant against
    ///    a correct pid check, and there precisely because pids are reused: a
    ///    fresh run that inherited a dead run's pid has a fresh mtime.
    ///
    /// Once per process, not once per call: 82 call sites would otherwise scan
    /// the whole of `/tmp` 82 times.
    fn sweep_dead_runs() {
        static ONCE: std::sync::Once = std::sync::Once::new();
        ONCE.call_once(|| {
            let Ok(entries) = std::fs::read_dir(std::env::temp_dir()) else {
                return;
            };
            for e in entries.flatten() {
                let name = e.file_name();
                let Some(rest) = name
                    .to_str()
                    .and_then(|n| n.strip_prefix("znippy-git-store-"))
                else {
                    continue;
                };
                // `<tag>-<pid>-<nanos>`, and a tag may contain `-`, so the pid
                // is the second field from the right.
                let Some(pid) = rest.rsplit('-').nth(1).and_then(|p| p.parse::<u32>().ok()) else {
                    continue;
                };
                if pid == std::process::id() || Path::new(&format!("/proc/{pid}")).exists() {
                    continue;
                }
                let stale = e
                    .metadata()
                    .and_then(|m| m.modified())
                    .ok()
                    .and_then(|t| t.elapsed().ok())
                    .is_some_and(|age| age > std::time::Duration::from_secs(3600));
                if stale {
                    let _ = std::fs::remove_dir_all(e.path());
                }
            }
        });
    }

    /// **The corpus pack, with its objects already resolved so a test knows what
    /// oids to ask for.**
    ///
    /// # 🔴 What this was, and why it is not that any more
    ///
    /// Until 2026-08-25 this scanned `~/git` and returned **the
    /// first `.pack` under 8 MiB, in `read_dir` order, that resolved standalone
    /// to more than 100 objects**. Thirty-five tests take their corpus from
    /// here, so the subject of all thirty-five was whichever repository the
    /// filesystem happened to name first on one developer's box — and this box
    /// had seven agents cloning and removing repositories beside it.
    ///
    /// It had drifted. `cargo test -p znippy-plugin-git --lib` on `3358bc3`,
    /// clean tree: **197 passed, 8 failed**, every failure a property of the
    /// found pack (it was not reachability-closed, it carried `REF_DELTA`s, and
    /// it was 933 KB where a bound wanted over 1 MB). And with the corpus root
    /// simply absent — a CI box, or anyone else's machine — all thirty-five died
    /// in the fixture: *"panicked at src/store.rs:763: Os { code: 2, kind:
    /// NotFound }"*.
    ///
    /// It is now [`crate::pack_fixture::corpus`]: a linear history generated in
    /// Rust from constants that module states, touching no filesystem. See that
    /// module for the eight failures in full and for why generating beat
    /// committing a `.pack`.
    ///
    /// **Built once per test binary.** The corpus is deterministic and
    /// immutable, and thirty-five call sites were each re-reading and
    /// re-resolving a pack.
    pub(crate) fn corpus_pack() -> (Vec<u8>, Vec<crate::resolve::Resolved>) {
        static ONCE: std::sync::OnceLock<(Vec<u8>, Vec<crate::resolve::Resolved>)> =
            std::sync::OnceLock::new();
        ONCE.get_or_init(|| {
            let pack = crate::pack_fixture::corpus().pack.clone();
            let rows = resolve(&pack, GitHashKind::Sha1, 0, &NoBases)
                .expect("the generated corpus resolves standalone");
            (pack, rows)
        })
        .clone()
    }

    /// A tiny pack built here, so a test can push something that is not tied to
    /// this machine. One object, whole, of whatever type is asked for.
    ///
    /// One writer for the framing: `one_blob_pack` and `one_tag_pack` differ by
    /// the three bits git spends on an object type and by nothing else, and two
    /// copies of a pack header is exactly the kind of twin LAW 5 is about.
    pub(crate) fn one_object_pack(kind: GitObjectKind, body: &[u8]) -> (Vec<u8>, Vec<u8>) {
        use std::io::Write;
        let type_bits: u8 = match kind {
            GitObjectKind::Commit => 1,
            GitObjectKind::Tree => 2,
            GitObjectKind::Blob => 3,
            GitObjectKind::Tag => 4,
        };
        let mut pack = b"PACK".to_vec();
        pack.extend_from_slice(&2u32.to_be_bytes());
        pack.extend_from_slice(&1u32.to_be_bytes());
        let mut size = body.len() as u64;
        let mut header = vec![(type_bits << 4) | (size as u8 & 0x0f)];
        size >>= 4;
        while size > 0 {
            let last = header.len() - 1;
            header[last] |= 0x80;
            header.push((size & 0x7f) as u8);
            size >>= 7;
        }
        pack.extend_from_slice(&header);
        let mut e = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
        e.write_all(body).unwrap();
        pack.extend_from_slice(&e.finish().unwrap());
        pack.extend_from_slice(&[0u8; 20]);
        let oid = GitHashKind::Sha1.oid_of(&canonical(kind, body));
        (pack, oid)
    }

    /// One blob, whole.
    pub(crate) fn one_blob_pack(body: &[u8]) -> (Vec<u8>, Vec<u8>) {
        one_object_pack(GitObjectKind::Blob, body)
    }

    /// **One ANNOTATED TAG, pointing at `target_hex`.**
    ///
    /// The object format is the point of the fixture: a tag's payload opens
    /// `object <hex>\n`, and that line is the only thing that connects a tag to
    /// the history it names. A reachability walk that does not read it answers
    /// with the tag and nothing under it.
    pub(crate) fn one_tag_pack(target_hex: &str, name: &str) -> (Vec<u8>, Vec<u8>) {
        let body = format!(
            "object {target_hex}\ntype commit\ntag {name}\ntagger t <t@t> 0 +0000\n\nannotated\n"
        );
        one_object_pack(GitObjectKind::Tag, body.as_bytes())
    }

    /// **The bytes come back byte for byte.** A real repository's pack is pushed,
    /// absorbed, and then every object is fetched by oid and compared against the
    /// exact slice of the pushed bytes it came from.
    ///
    /// This is the verbatim contract asserted as applied output — on the file on
    /// disk, not on what `put` returned. A store that re-compressed, re-framed or
    /// re-encoded anything cannot pass it.
    ///
    /// Seen RED by making `put_pack` store `&bytes[..bytes.len() - 1]`: "the
    /// extent covers the whole pack — left: 5653301, right: 5653302". One byte
    /// short of verbatim and the extent no longer covers the push.
    #[test]
    fn every_stored_object_reads_back_as_the_exact_bytes_that_were_pushed() {
        let dir = tmpdir("verbatim");
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack, rows) = corpus_pack();

        // The drain is parked in front of its absorb for as long as this lives,
        // so "queued, not run" below is a **state** and not a timing window. It
        // used to be neither: `absorb_pending() == 1` raced the drain for the
        // gate and lost whenever the drain got there first, which is a wrong
        // expectation rather than a wrong store — both callers end in the same
        // rows and the count only says which one paid.
        let gate = store.hold_absorb_gate();
        let tx = store.put(&pack, &[]).unwrap();
        let (offset, len) = tx.extent.expect("a pack push records its extent");
        assert_eq!(len, pack.len() as u64, "the extent covers the whole pack");

        // The file on disk holds the pushed bytes, unchanged.
        let on_disk = std::fs::read(store.blobs_path()).unwrap();
        assert_eq!(
            &on_disk[offset as usize..(offset + len) as usize],
            &pack[..],
            "the archive does not hold the pack verbatim"
        );

        assert_eq!(
            store.unindexed_packs(),
            1,
            "the index job is queued, not run"
        );
        assert_eq!(
            store.object_count(),
            0,
            "a row landed while the gate was held"
        );
        drop(gate);
        store.wait_indexed();
        assert_eq!(
            store.absorb_pending().unwrap(),
            0,
            "the drain left index work"
        );
        assert_eq!(store.object_count(), rows.len(), "every object is indexed");

        for (i, r) in rows.iter().enumerate() {
            let got = store
                .get(&r.oid)
                .unwrap()
                .unwrap_or_else(|| panic!("object {i} {} is missing", hex::encode(&r.oid)));
            let from = (offset + r.offset) as usize;
            assert_eq!(
                got.bytes,
                &pack[from..from + r.len as usize],
                "object {i} {}: the stored bytes are not the pushed bytes",
                hex::encode(&r.oid)
            );
            assert_eq!(got.extent, (offset + r.offset, r.len));
            assert_eq!(got.uncompressed_size, r.uncompressed_size);
            assert_eq!(got.obj_type, r.stored_type);
            assert_eq!(store.size(&r.oid).unwrap(), Some(r.uncompressed_size));
            assert!(store.has(&r.oid).unwrap());
        }

        // A delta really is labelled a delta, so nobody can mistake its bytes for
        // content. (If this pack had none, the assertion above would be vacuous.)
        let deltas = rows
            .iter()
            .filter(|r| matches!(r.stored_type, ObjType::OfsDelta | ObjType::RefDelta))
            .count();
        assert!(deltas > 0, "the fixture pack carries no deltas");
        let d = rows
            .iter()
            .find(|r| matches!(r.stored_type, ObjType::OfsDelta | ObjType::RefDelta))
            .unwrap();
        let got = store.get(&d.oid).unwrap().unwrap();
        assert!(matches!(
            got.obj_type,
            ObjType::OfsDelta | ObjType::RefDelta
        ));
        assert_ne!(
            got.bytes.len() as u64,
            got.uncompressed_size,
            "a delta's stored bytes are not its resolved size"
        );

        // An oid nothing pushed is absent, and that is an answer, not a guess.
        assert!(!store.has(&[0xab; 20]).unwrap());
        assert!(store.get(&[0xab; 20]).unwrap().is_none());
        assert_eq!(store.size(&[0xab; 20]).unwrap(), None);
    }

    /// **Durability, in the order that matters.** `put_pack` must not return
    /// before the bytes are on the device *and* a journal row on the device
    /// points at them — and if the machine dies between the two, what is left
    /// must be orphan bytes, never a row pointing into a hole.
    ///
    /// Asserted by reading the two files back from disk after a fault injected
    /// into the **real** `SafeWriter::append` (LAW 5: the fault goes into the one
    /// writer, rather than a hand-rolled twin of it being tested).
    ///
    /// Seen RED by moving the blob `fsync` in `SafeWriter::append` to *after* the
    /// journal row and its `fsync`, with the injected crash between them — i.e.
    /// the reference is made durable before the bytes it references: "the journal
    /// claims 1 extent(s) after a crash before the blob was durable — that is a
    /// dangling reference, which is the failure this ordering exists to prevent".
    /// Restored.
    ///
    /// **And the honest limit of this guard, found by trying a weaker mutation
    /// first.** Moving the blob `fsync` to the end *without* moving the fault did
    /// **not** turn it red: the fault still fires before the journal row, so the
    /// on-disk state is still orphan bytes, and an `fsync` that did not happen is
    /// unobservable from inside the process — the bytes read back out of the page
    /// cache either way. What this guard proves is therefore the **ordering of the
    /// reference against the bytes**, which is the part a crash can expose. It
    /// cannot prove an `fsync` reached the platter; nothing short of cutting power
    /// can.
    #[test]
    fn a_pack_is_durable_before_put_returns_and_the_fsyncs_are_ordered() {
        use crate::archive_write::{read_journal, ArchiveWrite, Faults, SafeWriter};

        let dir = tmpdir("durable");
        let (pack, _) = one_blob_pack(b"durability is an ordering property");

        // 1. The ordinary path: after `put_pack` returns, both files hold it.
        let store = GitStore::open(&dir, "alice").unwrap();
        let tx = store.put_pack(&pack).unwrap();
        let (offset, len) = tx.extent.unwrap();
        let journal = SafeWriter::journal_path(store.blobs_path());
        assert_eq!(
            read_journal(&journal).unwrap(),
            vec![(offset, len)],
            "the journal row that claims the extent is not there when put_pack returned"
        );
        assert_eq!(
            std::fs::metadata(store.blobs_path()).unwrap().len(),
            offset + len,
            "the blob file does not end where the journal says the pack does"
        );

        // 2. A crash between the two fsyncs leaves ORPHAN BYTES, not a dangling
        //    reference. Same `append`, one fault flipped.
        let dir2 = tmpdir("crash");
        let blobs = dir2.join("objects.pack");
        let w = SafeWriter::create_with_faults(
            &blobs,
            Faults {
                die_between_fsyncs: true,
                ..Default::default()
            },
        )
        .unwrap();
        assert!(
            w.append(&pack).is_err(),
            "the injected crash must not return Ok"
        );
        drop(w);
        let orphan = std::fs::read(&blobs).unwrap();
        assert_eq!(orphan, pack, "the blob bytes were fsynced before the crash");
        assert!(
            read_journal(&SafeWriter::journal_path(&blobs))
                .unwrap()
                .is_empty(),
            "the journal claims {} extent(s) after a crash before the blob was durable — that is a \
             dangling reference, which is the failure this ordering exists to prevent",
            read_journal(&SafeWriter::journal_path(&blobs)).unwrap().len()
        );

        // 3. And a store reopened over the orphan file answers for nothing,
        //    rather than for bytes nobody claimed.
        let reopened = GitStore::open(&dir2, "alice").unwrap();
        assert_eq!(reopened.object_count(), 0);
    }

    /// **The closure check refuses a pack before storing it**, and it does so
    /// without consulting the index for anything except a genuinely external
    /// base.
    ///
    /// Asserted on applied output: the blob file does not grow, and no journal
    /// row appears.
    ///
    /// Seen RED by moving `external_bases_exist` to *after* `push_pack`: "the
    /// refused pack was stored anyway: 0 bytes became 65".
    #[test]
    fn a_pack_that_fails_the_closure_check_is_refused_before_a_byte_is_stored() {
        let dir = tmpdir("closure");
        let store = GitStore::open(&dir, "alice").unwrap();

        // A thin pack: one ref-delta against an oid this repository has never
        // seen.
        let mut thin = b"PACK".to_vec();
        thin.extend_from_slice(&2u32.to_be_bytes());
        thin.extend_from_slice(&1u32.to_be_bytes());
        thin.push(0x74); // type 7, size 4
        thin.extend_from_slice(&[0x11; 20]);
        {
            use std::io::Write;
            let mut e = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
            e.write_all(&[0x04, 0x04, 0x90, 0x04]).unwrap();
            thin.extend_from_slice(&e.finish().unwrap());
        }
        thin.extend_from_slice(&[0u8; 20]);

        let before = std::fs::metadata(store.blobs_path()).unwrap().len();
        let err = store
            .put_pack(&thin)
            .expect_err("a dangling base is refused");
        let msg = err.to_string();
        assert!(
            msg.contains(&hex::encode([0x11; 20])),
            "names the base: {msg}"
        );
        let after = std::fs::metadata(store.blobs_path()).unwrap().len();
        assert_eq!(
            before, after,
            "the refused pack was stored anyway: {before} bytes became {after}"
        );
        assert_eq!(store.unindexed_packs(), 0, "and no index job was queued");

        // A corrupt pack — a delta base that lands nowhere — is refused too, and
        // as corruption rather than as thinness.
        let (mut corrupt, _) = one_blob_pack(b"x");
        corrupt[11] = 2; // claim two objects, provide one
        assert!(store.put_pack(&corrupt).is_err());
        assert_eq!(
            std::fs::metadata(store.blobs_path()).unwrap().len(),
            before,
            "a corrupt pack was stored"
        );
    }

    /// **CAS: a mismatch moves nothing.** The ref namespace is read back from the
    /// log after every attempt, so what is asserted is the ref's applied state
    /// and not the call's return value.
    ///
    /// Seen RED by comparing `actual.is_some() != old.is_some()` — existence
    /// only, the CAS bug that looks right: "stale old value: TxId { pack_id: None,
    /// extent: None, push_seq: Some(1) }", i.e. the swap from a wrong old value
    /// was accepted and written.
    #[test]
    fn a_compare_and_swap_that_loses_the_race_moves_no_ref() {
        let dir = tmpdir("cas");
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack_a, oid_a) = one_blob_pack(b"first");
        let (pack_b, oid_b) = one_blob_pack(b"second");
        store.put_pack(&pack_a).unwrap();
        store.put_pack(&pack_b).unwrap();
        store.absorb_pending().unwrap();

        let name = "refs/heads/main";
        let target = |s: &GitStore| -> Option<Vec<u8>> {
            s.refs()
                .unwrap()
                .into_iter()
                .find(|r| r.name == name)
                .and_then(|r| r.oid)
        };

        // ★ A LOST CAS IS TYPED, not a string a caller has to parse.
        //
        // `update_ref` raised a bare `bail!` while its own sibling
        // `put_refs_cas` raised `RefRejection::Cas`, so `gunnar-wire`'s
        // receive-pack could not tell a lost race from a broken disk and said
        // so in a comment at the call site. The contract had the type all
        // along; this method did not use it.
        //
        // Seen RED by restoring the `bail!`: "a lost CAS must carry
        // RefRejection::Cas, not a string: compare-and-swap on refs/heads/main
        // failed: it is absent, the caller expected 0101…" — the message was
        // right and the TYPE was missing, which is exactly the failure a
        // string-shaped error hides.
        let err = store
            .update_ref(name, Some(&oid_a), Some(&oid_b))
            .expect_err("a create that claims a previous value must fail");
        match err.downcast_ref::<git_storage_trait::RefRejection>() {
            Some(git_storage_trait::RefRejection::Cas {
                name: n,
                expected,
                actual,
            }) => {
                assert_eq!(n, name, "the rejection names the wrong ref");
                assert_eq!(
                    *actual,
                    git_storage_trait::Observed::Nothing,
                    "the ref did not exist, so `actual` must say so"
                );
                assert_ne!(
                    *expected,
                    git_storage_trait::Observed::Nothing,
                    "the caller DID claim a previous value; `expected` must carry it"
                );
            }
            other => panic!(
                "a lost CAS must carry RefRejection::Cas, not a string: {err:#} (downcast: {})",
                if other.is_some() { "wrong variant" } else { "none" }
            ),
        }
        assert_eq!(target(&store), None, "the failed create wrote something");

        let tx = store.update_ref(name, None, Some(&oid_a)).unwrap();
        assert!(tx.push_seq.is_some());
        assert_eq!(target(&store).as_deref(), Some(&oid_a[..]));

        // A swap from the WRONG old value: refused, and nothing moves.
        let err = store
            .update_ref(name, Some(&oid_b), Some(&oid_b))
            .expect_err("stale old value");
        assert!(err.to_string().contains("compare-and-swap"), "{err}");
        assert_eq!(
            target(&store).as_deref(),
            Some(&oid_a[..]),
            "refs/heads/main moved on a failed CAS"
        );

        // The right old value: it moves.
        store.update_ref(name, Some(&oid_a), Some(&oid_b)).unwrap();
        assert_eq!(target(&store).as_deref(), Some(&oid_b[..]));

        // A ref may not point at an object we do not have.
        let err = store
            .update_ref(name, Some(&oid_b), Some(&[0xcd; 20]))
            .expect_err("dangling target");
        assert!(err.to_string().contains("does not have"), "{err}");
        assert_eq!(target(&store).as_deref(), Some(&oid_b[..]));

        // Delete: gone from the namespace, and the log is the history.
        store.update_ref(name, Some(&oid_b), None).unwrap();
        assert_eq!(target(&store), None);
        assert!(store.refs().unwrap().iter().all(|r| r.name != name));
    }

    /// **`put_refs_cas` persists a `RefCas`'s peeled target into the ref row.**
    ///
    /// This is the write half of gunnar S-080: the advertisement re-peels every
    /// annotated tag on every clone/fetch/`ls-remote` for want of a stored
    /// peel, and the store is where the peel has to be kept. A tag ref applied
    /// through the atomic CAS path must land its peeled column, exactly as one
    /// applied through [`GitOps::put_refs`] does — otherwise a tag pushed with
    /// `git push --atomic` would advertise unpeeled while the same tag pushed
    /// without `--atomic` advertised peeled.
    ///
    /// Seen RED by dropping `e.peeled` in `put_refs_cas`'s `RefUpdate::set`
    /// arm — the ref lands, `oid` is right, and `peeled` comes back `None`.
    #[test]
    fn put_refs_cas_persists_a_tags_peeled_target() {
        let dir = tmpdir("cas-peel");
        let store = GitStore::open(&dir, "alice").unwrap();
        // Two real objects: one stands in for the tag object the ref names, the
        // other for the commit it peels to. The store only requires the ref's
        // own target to be present; the peel rides through as data.
        let (pack_tag, oid_tag) = one_blob_pack(b"tag-object");
        let (pack_commit, oid_commit) = one_blob_pack(b"peeled-commit");
        store.put_pack(&pack_tag).unwrap();
        store.put_pack(&pack_commit).unwrap();
        store.absorb_pending().unwrap();

        store
            .put_refs_cas(&[RefCas {
                name: "refs/tags/v9".into(),
                old: None,
                new: Some(&oid_tag),
                peeled: Some(oid_commit.clone()),
            }])
            .unwrap();

        let row = store
            .refs()
            .unwrap()
            .into_iter()
            .find(|r| r.name == "refs/tags/v9")
            .expect("the tag ref was applied");
        assert_eq!(row.oid.as_deref(), Some(&oid_tag[..]), "the ref target");
        assert_eq!(
            row.peeled.as_deref(),
            Some(&oid_commit[..]),
            "the CAS path dropped the peeled target the RefCas carried"
        );

        // A plain (non-tag) create through the same path carries no peel and so
        // stores none — the branch-head case that must stay `None`.
        store
            .put_refs_cas(&[RefCas {
                name: "refs/heads/plain".into(),
                old: None,
                new: Some(&oid_commit),
                peeled: None,
            }])
            .unwrap();
        let branch = store
            .refs()
            .unwrap()
            .into_iter()
            .find(|r| r.name == "refs/heads/plain")
            .expect("the branch was applied");
        assert_eq!(
            branch.peeled, None,
            "a non-tag create must not invent a peeled target"
        );
    }

    /// **The bounded walk answers EXACTLY what a full bitmap table answers** —
    /// under-send and over-send both named, over every commit in the fixture.
    ///
    /// # What is actually at stake here
    ///
    /// Until 2026-08-14 the live table bitmapped **every** commit
    /// (`ReachPolicy { max_commits: usize::MAX }`), not as a choice but because
    /// `reachable_oids` had no fallback: a `want` with no bitmap contributed
    /// itself and nothing else. MEASURED on oden — 530 218 of a 4-object fetch's
    /// 575 349 allocations were that build, thrown away again by the next push's
    /// `refold`. [`crate::reach::accumulate`] is the walk that makes a sampled
    /// table legal, and `LIVE_REACH_COMMITS` is now 512.
    ///
    /// **A walk that stops too early UNDER-SENDS, and a clone that under-sends
    /// is silent data loss**: the pack indexes, `fsck`s and applies, the client
    /// exits zero, and the repository is missing objects it will not discover
    /// for days. `select` refusing was a correct-but-expensive answer to exactly
    /// this hazard, and this test is what replaces it.
    ///
    /// # Three arms over one store, and why three
    ///
    /// | cap | table | what runs |
    /// |---|---|---|
    /// | `usize::MAX` | every commit | no walk at all — **the reference** |
    /// | `0` | empty | the walk does *everything* |
    /// | `2` | two commits | the mixed case, where phase 1's stopping rule actually fires |
    ///
    /// Two arms would not be enough. `0` never exercises the stop-at-a-bitmap
    /// rule, which is the one thing that could OR in a *closed* set at the wrong
    /// moment, and `usize::MAX` never exercises the walk. The middle arm is the
    /// only one where both halves meet.
    ///
    /// # Both directions are named, and that is the anti-hollow half
    ///
    /// The comparison is a **symmetric difference**, not a length check and not
    /// a subset check. An over-send passes every downstream test we own — the
    /// pack indexes, `--check-self-contained-and-connected` accepts it, `fsck`
    /// is clean — and today's gix defect was exactly that shape. So `missing`
    /// and `extra` are computed and printed separately: a walk that stops short
    /// names the objects it lost, and a walk that ORs too much names the objects
    /// it invented.
    ///
    /// # Seen RED, 2026-08-14, three ways — and the directions are instructive
    ///
    /// * **stop one commit short** (`continue` before pushing parents in phase
    ///   1) — *"cap 0 vs the full table disagree for want 75c9e5c3…: MISSING 0
    ///   object(s) …; EXTRA 2 object(s) the full table did not select, first
    ///   Some(\"75c9e5c3d1d08dd92cc913c70ad10288a19eea4e\")"*.
    /// * **lose objects in the tree walk** (skip every seventh ordinal in
    ///   `accumulate_tree`) — *"cap 0 …: MISSING 2 object(s) the full table
    ///   selected, first Some(\"28b3c3658111299869d1f63f49a8cfb2635c0cfe\");
    ///   EXTRA 0"*.
    /// * **OR in an unrelated commit's bitmap** — *"cap 2 …: MISSING 13
    ///   object(s) the full table selected, first
    ///   Some(\"078f18e68253d6fc56e8cba230ef9f32dec2e2ab\"); EXTRA 0"*.
    ///
    /// 🔴 **Note which direction each one came out.** Truncating the walk
    /// produced an *over*-send and over-collecting produced an *under*-send —
    /// the opposite of the intuition in both cases — because the identical walk
    /// serves the `have` side, and an error there enters the answer through
    /// `union − had` with its sign flipped. That is precisely why the comparison
    /// is a symmetric difference and reports **both** counts: a guard that
    /// checked only for missing objects would have passed the first break, and a
    /// guard that checked only lengths would have named nothing.
    ///
    /// The premise is guarded too: with the arms wired to the same policy this
    /// test passes trivially, so the table size at each cap is asserted before
    /// any answer is compared.
    #[test]
    fn the_bounded_walk_answers_exactly_what_a_full_bitmap_table_answers() {
        use crate::reach::ReachPolicy;

        let dir = tmpdir("reach-walk");
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack, _) = corpus_pack();
        store.put(&pack, &[]).unwrap();
        store.absorb_pending().unwrap();

        let commits: Vec<String> = store.graph_snapshot().iter().map(|c| c.oid.clone()).collect();
        assert!(
            commits.len() > 2,
            "premise: the fixture needs more commits than the smallest cap under test, or every \
             arm bitmaps everything and the walk never runs — found {}",
            commits.len()
        );

        let full = ReachPolicy {
            max_commits: usize::MAX,
        };
        let none = ReachPolicy { max_commits: 0 };
        let some = ReachPolicy { max_commits: 2 };

        // The premise, measured off the tables themselves rather than assumed.
        // A cap that did not actually shrink the table would make every
        // comparison below a comparison of one code path with itself.
        let n_full = store.reach_bitmaps_with(full, false).unwrap().len();
        let n_none = store.reach_bitmaps_with(none, false).unwrap().len();
        let n_some = store.reach_bitmaps_with(some, false).unwrap().len();
        assert_eq!(
            n_full,
            commits.len(),
            "premise: the uncapped table must bitmap every commit"
        );
        assert_eq!(
            n_none, 0,
            "premise: the zero cap must produce NO bitmaps, or the walk arm is not a walk arm"
        );
        assert!(
            n_some > 0 && n_some < commits.len(),
            "premise: the middle cap must bitmap SOME commits and not all ({n_some} of {}), or \
             the stop-at-a-bitmap rule is never reached",
            commits.len()
        );

        // A stride through the graph as `want`, each against its successor as
        // `have` — so the walk is asked about tips, roots and the middle.
        //
        // A stride and not every commit: `reachable_oids_with` rebuilds the
        // table on every call here (`cache: false`, and it must be — see that
        // method), so the exhaustive form is quadratic and ran for 334 s against
        // this fixture. Twelve probes spread across the history cost seconds and
        // cover the same three positions; the arms that matter are the three
        // caps, not the commit count.
        let stride = commits.len().div_ceil(12).max(1);
        let probes: Vec<usize> = (0..commits.len())
            .step_by(stride)
            .chain([commits.len() - 1])
            .collect();
        let mut walked_any = false;
        for i in probes {
            let w = &commits[i];
            let w_raw = hex::decode(w).unwrap();
            let h_raw = (i + 1 < commits.len()).then(|| hex::decode(&commits[i + 1]).unwrap());
            let haves: Vec<&[u8]> = h_raw.iter().map(|v| v.as_slice()).collect();

            let reference: std::collections::BTreeSet<String> = store
                .reachable_oids_with(&[&w_raw], &haves, full, false)
                .unwrap()
                .into_iter()
                .collect();

            for (label, policy) in [("cap 0", none), ("cap 2", some)] {
                let got: std::collections::BTreeSet<String> = store
                    .reachable_oids_with(&[&w_raw], &haves, policy, false)
                    .unwrap()
                    .into_iter()
                    .collect();
                let missing: Vec<&String> = reference.difference(&got).collect();
                let extra: Vec<&String> = got.difference(&reference).collect();
                assert!(
                    missing.is_empty() && extra.is_empty(),
                    "{label} vs the full table disagree for want {w}: MISSING {} object(s) the \
                     full table selected, first {:?}; EXTRA {} object(s) the full table did not \
                     select, first {:?}",
                    missing.len(),
                    missing.first(),
                    extra.len(),
                    extra.first()
                );
            }
            walked_any = true;
        }
        assert!(walked_any, "no commit was compared");

        // And the answers are not vacuously equal because they are all empty.
        let tip = hex::decode(&commits[0]).unwrap();
        let n = store
            .reachable_oids_with(&[&tip], &[], none, false)
            .unwrap()
            .len();
        assert!(
            n > 1,
            "premise: a commit must reach more than itself for the comparison above to have any \
             content — got {n}"
        );
    }

    /// **`want` minus `have` is an `andnot`, over objects and not just commits.**
    ///
    /// Built on a real repository's pack: the second commit's closure minus the
    /// first's must be exactly the objects the second commit introduced, and
    /// every returned oid must be one the store really holds.
    ///
    /// Seen RED by returning `union` instead of `union - had`: "removing the
    /// parent's closure removed nothing: 8 vs 8".
    ///
    /// MEASURED on the fixture: the child commit's closure is 8 objects, the
    /// parent's is 5, and the difference is 3 — the objects that commit
    /// introduced.
    #[test]
    fn reachable_is_want_minus_have_over_objects() {
        let dir = tmpdir("reach");
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack, _) = corpus_pack();
        store.put(&pack, &[]).unwrap();
        store.absorb_pending().unwrap();
        assert!(store.commit_count() > 1, "the fixture has no history");

        // Two commits in parent→child order: the graph is folded that way.
        let (child, parent) = {
            let mut pair = None;
            for c in store.graph_snapshot() {
                if let Some(p) = c.parents.first() {
                    pair = Some((c.oid.clone(), p.clone()));
                    break;
                }
            }
            pair.expect("a commit with a parent")
        };
        let child_raw = hex::decode(&child).unwrap();
        let parent_raw = hex::decode(&parent).unwrap();

        let all = store.reachable(&[&child_raw], &[]).unwrap();
        let delta = store.reachable(&[&child_raw], &[&parent_raw]).unwrap();
        let had = store.reachable(&[&parent_raw], &[]).unwrap();

        assert!(
            !all.is_empty(),
            "a commit reaches at least itself and its tree"
        );
        assert!(
            delta.len() < all.len(),
            "removing the parent's closure removed nothing: {} vs {}",
            delta.len(),
            all.len()
        );
        assert_eq!(
            all.len(),
            delta.len() + had.iter().filter(|h| all.contains(h)).count(),
            "the three sets do not add up — want minus have is not an andnot"
        );
        for oid in delta.iter().chain(all.iter()) {
            assert!(
                store.has(oid).unwrap(),
                "reachable named {} which the store does not hold",
                hex::encode(oid)
            );
        }
        // The commit itself is in its own closure; its parent is not in the delta.
        assert!(all.contains(&child_raw));
        assert!(!delta.contains(&parent_raw));
        eprintln!(
            "closure {} objects, minus the parent's {} leaves {}",
            all.len(),
            had.len(),
            delta.len()
        );
    }

    /// **The batch of one really is slower**, measured here rather than asserted
    /// from the plan — and the serial path is what `extents` takes for it.
    ///
    /// The ratio is machine- and load-dependent, so the guard on it is loose (it
    /// requires the batch path to be no *faster*, which is the direction the
    /// dispatch depends on) while the number is printed for the record. The
    /// dispatch itself is asserted exactly.
    ///
    /// MEASURED, release, oden, loadavg 4.94, 2687 objects: `lookup` 628 ns,
    /// `lookup_batch` of one 639 ns (**1.02x worse**, not the 1.7x the bare Arrow
    /// arms show), `lookup_batch` of 100 **161 ns/oid** — 3.9x better than the
    /// serial path. See [`lookup_path`] for why both numbers are recorded.
    ///
    /// Seen RED by making `lookup_path` return `Batch` for every n:
    /// "assertion `left == right` failed — left: Batch, right: Serial".
    ///
    /// The honest limit: a *timing* assertion cannot catch `extents` ignoring the
    /// dispatch, because the two paths differ by 2% here. That is exactly why the
    /// dispatch is a named function with an exact assertion on it rather than a
    /// comment above an `if`.
    #[test]
    fn the_batch_of_one_really_is_slower_than_the_serial_path() {
        assert_eq!(lookup_path(1), LookupPath::Serial);
        assert_eq!(lookup_path(2), LookupPath::Batch);
        assert_eq!(lookup_path(0), LookupPath::Batch);
        assert_eq!(BATCH_SATURATES_AT, 100);

        let dir = tmpdir("batch");
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack, rows) = corpus_pack();
        store.put(&pack, &[]).unwrap();
        store.absorb_pending().unwrap();
        let oids: Vec<&[u8]> = rows.iter().map(|r| r.oid.as_slice()).collect();

        // Both paths answer identically for one oid — the dispatch is an
        // optimisation, never a difference in the answer.
        for oid in oids.iter().take(64) {
            assert_eq!(
                store.extents(&[*oid]).unwrap(),
                store.index().extents_batch(&[*oid]),
                "the serial and batch paths disagree for {}",
                hex::encode(oid)
            );
        }

        // 🔴 **Interleaved rounds, and the FASTEST of each is the reading.**
        //
        // This took one timed pass of each, one after the other, and compared
        // them. On a box running anything else that is a coin toss with extra
        // steps: the two passes are 20 000 lookups apart in time, so a
        // scheduler steal or a migration that lands on the first one alone
        // shifts the ratio by more than the 10 % the assertion allows. Seen RED
        // that way 2026-08-25 inside a full `cargo test` run, and green three
        // times in a row on the same binary in isolation — the failure was the
        // other 200 tests, not the dispatch.
        //
        // A stolen timeslice can only make a measurement SLOWER, never faster,
        // so the minimum over rounds is the least-perturbed sample of each and
        // the noise floor cancels out of the ratio. Interleaved rather than
        // batched per side, so a slow patch of wall clock hits both.
        let n = 20_000usize;
        const ROUNDS: usize = 5;
        let mut serial = f64::MAX;
        let mut batched = f64::MAX;
        for _ in 0..ROUNDS {
            let t = Instant::now();
            for i in 0..n {
                let _ = store.index().lookup(oids[i % oids.len()]);
            }
            serial = serial.min(t.elapsed().as_nanos() as f64 / n as f64);

            let t = Instant::now();
            for i in 0..n {
                let _ = store.index().lookup_batch(&[oids[i % oids.len()]]);
            }
            batched = batched.min(t.elapsed().as_nanos() as f64 / n as f64);
        }
        // What the batch path is actually for: 100 oids in one call, which is
        // where it saturates.
        let hundred: Vec<&[u8]> = oids.iter().take(BATCH_SATURATES_AT).copied().collect();
        let per_oid_at_100 = {
            let rounds = n / BATCH_SATURATES_AT;
            let t = Instant::now();
            for _ in 0..rounds {
                let _ = store.index().lookup_batch(&hundred);
            }
            t.elapsed().as_nanos() as f64 / (rounds * hundred.len()) as f64
        };
        eprintln!(
            "load {}; {} objects: lookup {serial:.0} ns, lookup_batch[1] {batched:.0} ns \
             ({:.2}x, best of {ROUNDS} interleaved rounds), lookup_batch[{}] \
             {per_oid_at_100:.0} ns/oid",
            std::fs::read_to_string("/proc/loadavg")
                .unwrap_or_default()
                .trim(),
            store.index().len(),
            batched / serial,
            hundred.len(),
        );
        // 2026-08-10: the fused batch walk (no sort phase) collapsed the old
        // 1.02x batch-of-one penalty to a measured TIE — 604-620 ns both sides,
        // winner decided by noise, seen flipping run-to-run on an idle box. The
        // `Serial` dispatch for n=1 stays correct (equal time, two fewer Vec
        // allocations), so the guard keeps only the direction that would make
        // it WRONG: a batch of one materially faster than serial. 10% is
        // outside the tie's observed 0.3% jitter and inside any real win.
        //
        // RELEASE ONLY (2026-10-16). Every number above that this claim rests on
        // is a release build (~600 ns a lookup). Under `cargo test`'s debug
        // profile the same lookup is ~20 000 ns of unoptimised call layers, and
        // which side those layers weigh on is a property of the debug codegen,
        // not of the dispatch: measured on dwarf at load 27, debug, 0.90x — red
        // by 0.5 %, while the release build of the same code is the tie above.
        // A ratio of unoptimised code says nothing about the dispatch, so in a
        // debug build the reading is printed (above) and not judged; the exact
        // `lookup_path` asserts at the top of this test still run in both.
        if cfg!(debug_assertions) {
            eprintln!(
                "NOT JUDGED in a debug build: the batch-of-one ratio is a release claim \
                 (run with --release to judge it)"
            );
            return;
        }
        assert!(
            batched >= serial * 0.9,
            "a batch of one measured MATERIALLY faster ({batched:.0} ns vs {serial:.0} ns) — \
             if that is repeatable the dispatch in `lookup_path` is wrong and must be \
             changed, not kept"
        );
    }

    /// **`seal` folds the live logs into reserved Arrow sections**, and the
    /// generation numbers in the graph section are recomputed by that fold.
    ///
    /// Asserted by decoding the sections back out of their Arrow IPC bytes.
    ///
    /// Seen RED by dropping the `assign_generations` call from `refold`:
    /// "e58a5032ac3b36f15f08c759121efc8f08680e2b has generation 0".
    #[test]
    fn seal_emits_the_reserved_sections_and_recomputes_every_generation() {
        use znippy_common::{GUNNAR_GRAPH_MODULE, GUNNAR_REACH_MODULE, GUNNAR_REFS_MODULE};

        let dir = tmpdir("seal");
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack, _) = corpus_pack();
        store.put(&pack, &[]).unwrap();
        store.absorb_pending().unwrap();
        let tip = store
            .graph_snapshot()
            .iter()
            .max_by_key(|c| c.generation)
            .expect("a graph")
            .clone();
        store
            .update_ref(
                "refs/heads/main",
                None,
                Some(&hex::decode(&tip.oid).unwrap()),
            )
            .unwrap();

        let sections = store.seal().unwrap();
        let names: Vec<&str> = sections.iter().map(|s| s.module_name.as_str()).collect();
        for want in [GUNNAR_REFS_MODULE, GUNNAR_GRAPH_MODULE, GUNNAR_REACH_MODULE] {
            assert!(names.contains(&want), "{want} is not sealed: {names:?}");
        }

        // Generations: a root is 1, a child is 1 + max(parents), and the graph
        // this store folded must say so.
        let graph = store.graph_snapshot();
        let by_oid: std::collections::HashMap<&str, &crate::graph::CommitNode> =
            graph.iter().map(|c| (c.oid.as_str(), c)).collect();
        let mut with_parents = 0usize;
        for c in &graph {
            assert!(c.generation >= 1, "{} has generation 0", c.oid);
            let known: Vec<&crate::graph::CommitNode> = c
                .parents
                .iter()
                .filter_map(|p| by_oid.get(p.as_str()).copied())
                .collect();
            if known.is_empty() {
                assert_eq!(c.generation, 1, "{} is a root here", c.oid);
            } else {
                with_parents += 1;
                let expect = 1 + known.iter().map(|p| p.generation).max().unwrap();
                assert_eq!(
                    c.generation, expect,
                    "{} must have generation {expect}",
                    c.oid
                );
                assert!(
                    c.generation > 1,
                    "a commit with a parent must have generation > 1"
                );
            }
        }
        assert!(with_parents > 0, "no commit in the fixture has a parent");
        eprintln!(
            "{} commits sealed, {with_parents} with parents",
            graph.len()
        );
    }

    /// **`seal` writes generation 0, and `gc` then has something to compact.**
    ///
    /// This is the hole the four working steps of `gc()` used to fall into:
    /// absorb, live set, retire, drop all ran, and then
    /// `NewGeneration::run(archive_path())` died on
    /// `stat …/repository.znippy: No such file or directory` because **nothing
    /// ever created generation 0**. Every other test in this file that reached
    /// step 5 hand-built the archive with `znippy_common::create_archive`, which
    /// is why the gap survived: the fixture was doing the store's job.
    ///
    /// Asserted on **applied output**, through znippy's own reader and nothing
    /// else:
    ///
    /// * the live pack comes back out of the archive **byte for byte**, via
    ///   `extract_file_verified`, which reconstructs the entry and blake3-checks
    ///   it against the index — so a wrong checksum, a wrong extent or a wrong
    ///   `compressed` flag all fail here rather than being written and believed;
    /// * the tombstoned pack has **no** entry, because a row for it is exactly
    ///   what would stop the compaction reclaiming its bytes;
    /// * the three reserved sections are addressable **out of the file**, not
    ///   merely present in the returned vector;
    /// * `gc()` then runs end to end and `NewGeneration` produces
    ///   `repository.g1.znippy`, verified, with the old generation gone.
    ///
    /// # Seen RED, and one of the reds was in this guard
    ///
    /// Every mutation below was applied to the real code, run, and reverted.
    ///
    /// 1. **The hole itself.** With `seal` returning the sections and writing
    ///    nothing — the code as it stood — this fails at
    ///    `seal did not create …/repository.znippy`, and `gc()` fails with the
    ///    original `compacting …/repository.znippy after dropping 2683 dead
    ///    rows: stat …/repository.znippy: No such file or directory`.
    /// 2. **A dropped row** — `.take(packs.len() - 1)` on the row loop:
    ///    `objects.pack.2 is not in the sealed archive: ["objects.pack.0"]`.
    ///    **This is the mutation that first found a hollow guard, and the guard
    ///    was mine.** Against the two-pack fixture this test started with, the
    ///    row dropped was the tombstoned one, which is skipped anyway — the
    ///    mutation stayed GREEN. The fixture now pushes three packs and
    ///    tombstones the middle one, so there are two live rows and dropping
    ///    either is visible.
    /// 3. **No tombstone honoured** — the `retired.contains` skip disabled:
    ///    `the tombstoned pack got an index row anyway … ["objects.pack.0",
    ///    "objects.pack.2", "objects.pack.1"]`. A row for a retired pack is what
    ///    would keep its payload alive through every future compaction.
    /// 4. **A corrupted extent** — `blob_offset: offset + 1`. It writes, lists
    ///    and opens perfectly; it dies at the blake3 gate inside
    ///    `extract_file_verified`: `checksum mismatch for objects.pack.0 at
    ///    fdata_offset 0`.
    /// 5. **The checksum domain** — hashing the *path* instead of the bytes,
    ///    which is precisely the mistake that produces an archive that writes
    ///    cleanly and verifies wrong. Same gate, same message. Together 4 and 5
    ///    are what establish that the checksum is blake3 over the bytes at
    ///    `blob_offset`, and that this test can tell.
    /// 6. **The `compressed` flag** — `true` on bytes that were stored raw:
    ///    `OpenZL getDecompressedSize: ZL_getDecompressedSize failed`. The flag
    ///    is load-bearing, not decorative.
    /// 7. **A lying report** — `packs_retired: 0` in the `SealReport` literal:
    ///    `the report miscounts the tombstoned packs`. The report is read back
    ///    by a second seal, which also pins that a seal is a snapshot: taking it
    ///    twice yields the same entries, not a second generation.
    /// 8. **No reserved builder attached** — the sections are still *returned*,
    ///    so a guard that only inspected the return value would pass:
    ///    `__gunnar_refs__ is not in the sealed archive's manifest`. That is why
    ///    the sections are read back out of the file by module name.
    ///
    #[test]
    fn seal_writes_generation_zero_that_gc_can_compact() {
        use znippy_common::{
            read_reserved_section_bytes, ZnippyArchive, ZnippyReader, GUNNAR_GRAPH_MODULE,
            GUNNAR_REACH_MODULE, GUNNAR_REFS_MODULE,
        };

        let dir = tmpdir("seal-g0");
        let store = GitStore::open(&dir, "alice").unwrap();
        // THREE packs, and the middle one is tombstoned before the seal. Two
        // live rows rather than one is not decoration: with a single live entry,
        // "dropped a row" and "wrote no rows at all" are the same failure, and a
        // `.take(packs.len() - 1)` mutation of the row loop stayed GREEN against
        // an earlier two-pack version of this fixture. Three packs with a gap in
        // the middle also pin the ordinal: it is the position among the
        // journal's `Pack` rows, so the survivors are 0 and **2**, not 0 and 1.
        let (pack, _) = corpus_pack();
        let (doomed, doomed_oid) = one_blob_pack(b"this pack is retired before the seal");
        let (kept, kept_oid) = one_blob_pack(b"this pack is unreferenced but not tombstoned");
        store.put(&pack, &[]).unwrap();
        let doomed_tx = store.put_pack(&doomed).unwrap();
        store.put_pack(&kept).unwrap();
        store.absorb_pending().unwrap();
        for (what, oid) in [("doomed", &doomed_oid), ("kept", &kept_oid)] {
            assert!(
                store.has(oid).unwrap(),
                "the {what} blob was never stored, so this fixture proves nothing"
            );
        }

        // A ref, so the live set is not empty and `gc` is not refused.
        let root = store
            .graph_snapshot()
            .into_iter()
            .find(|c| c.generation == 1)
            .expect("a root commit");
        let root_raw = hex::decode(&root.oid).unwrap();
        store
            .update_ref("refs/heads/root", None, Some(&root_raw))
            .unwrap();

        // Exactly one tombstone, written the way `gc` writes them. Named
        // explicitly rather than via `retire_dead_packs`, which would tombstone
        // BOTH one-blob packs and leave nothing to prove the ordinal with.
        crate::archive_write::retire_packs(
            &crate::archive_write::SafeWriter::journal_path(store.blobs_path()),
            &[doomed_tx.extent.unwrap().0],
        )
        .unwrap();

        assert!(
            !store.archive_path().exists(),
            "something created the archive before the seal ran"
        );
        let sections = store.seal().unwrap();

        // ── the archive exists and reads back through znippy's own reader ────

        assert!(
            store.archive_path().exists(),
            "seal did not create {}",
            store.archive_path().display()
        );
        let ar = ZnippyArchive::open(store.archive_path()).unwrap();
        let listed = ar.list_files().unwrap();
        assert!(
            listed.contains(&"objects.pack.0".to_string()),
            "objects.pack.0 is not in the sealed archive: {listed:?}"
        );
        assert!(
            listed.contains(&"objects.pack.2".to_string()),
            "objects.pack.2 is not in the sealed archive: {listed:?}"
        );
        assert!(
            !listed.contains(&"objects.pack.1".to_string()),
            "the tombstoned pack got an index row anyway, so the compaction can never \
             reclaim its bytes: {listed:?}"
        );
        assert_eq!(
            listed.len(),
            2,
            "the seal wrote entries for packs the journal never acked as live: {listed:?}"
        );

        // The bytes, reconstructed and blake3-checked by the reader itself.
        assert_eq!(
            ar.extract_file_verified("objects.pack.0").unwrap(),
            pack,
            "objects.pack.0 does not read back as the pack that was pushed"
        );
        assert_eq!(
            ar.extract_file_verified("objects.pack.2").unwrap(),
            kept,
            "objects.pack.2 does not read back as the pack that was pushed"
        );

        // ── the reserved sections, addressed out of the file ─────────────────
        let names: Vec<&str> = sections.iter().map(|s| s.module_name.as_str()).collect();
        for want in [GUNNAR_REFS_MODULE, GUNNAR_GRAPH_MODULE, GUNNAR_REACH_MODULE] {
            assert!(names.contains(&want), "{want} was not returned: {names:?}");
            let bytes = read_reserved_section_bytes(store.archive_path(), want)
                .unwrap()
                .unwrap_or_else(|| panic!("{want} is not in the sealed archive's manifest"));
            assert!(!bytes.is_empty(), "{want} is sealed as zero bytes");
        }

        // ── the report, and that a second seal is not a different archive ───
        //
        // A count nobody asserts on is a count that can be wrong for ever, so
        // the report is read here — and reading it means sealing again, which
        // pins the other half: a seal is a snapshot and re-taking it produces
        // the same entries, not a second generation.
        let again = store
            .seal_archive(store.reserved_sections().unwrap())
            .unwrap();
        assert_eq!(
            again.packs_sealed, 2,
            "the report miscounts the sealed packs"
        );
        assert_eq!(
            again.packs_retired, 1,
            "the report miscounts the tombstoned packs"
        );
        assert_eq!(
            again.packs_after_copy, 0,
            "nothing was pushed during this seal, so nothing can have raced it"
        );
        let mut once = listed.clone();
        once.sort();
        let mut twice = ZnippyArchive::open(store.archive_path())
            .unwrap()
            .list_files()
            .unwrap();
        twice.sort();
        // Sorted on both sides: `list_files` does not promise an order, and it
        // was observed returning the same two entries the other way round.
        assert_eq!(
            twice, once,
            "sealing twice produced a different set of entries"
        );

        // ── and now the fifth step of `gc` has something to compact ──────────
        let report = store.gc().unwrap();
        assert_eq!(report.strategy, "NewGeneration");
        assert!(report.verified, "the new generation was not read back");
        assert_eq!(
            report.archive,
            store.archive_path().with_file_name("repository.g1.znippy"),
            "NewGeneration did not produce generation 1"
        );
        assert!(report.archive.exists(), "the new generation is not on disk");
        assert!(
            !store.archive_path().exists(),
            "the old generation was not retired"
        );
        assert!(
            report.bytes_after < report.bytes_before,
            "the compaction reclaimed nothing: {} → {} bytes — the retired pack's payload \
             should have gone",
            report.bytes_before,
            report.bytes_after
        );

        // The live pack survives the compaction; the retired one's row never
        // existed, so its bytes are what got reclaimed.
        let g1 = ZnippyArchive::open(&report.archive).unwrap();
        assert_eq!(
            g1.extract_file_verified("objects.pack.0").unwrap(),
            pack,
            "generation 1 does not carry the live pack"
        );
        eprintln!(
            "seal: {} bytes, gc: {} → {} bytes, {}",
            std::fs::metadata(&report.archive)
                .map(|m| m.len())
                .unwrap_or(0),
            report.bytes_before,
            report.bytes_after,
            report.archive.display()
        );
    }

    /// **GC in the order §13.20 fixes**: reachability, then the dead index rows,
    /// then base znippy's compaction.
    ///
    /// The archive is a real znippy archive built the way `gc`'s own fixture
    /// builds one, so step 3 really runs `compact_archive` and really produces a
    /// new generation. Steps 1 and 2 are asserted as applied output: the
    /// unreachable oid stops resolving and the reachable one still does.
    ///
    /// Seen RED by having `gc` drop rows *before* computing the live set — the
    /// order §13.20 fixes, inverted: "gc dropped the live object too". Seen RED a
    /// second time by making `drop_dead_rows` return `Ok(0)` without removing
    /// anything: "the dead object still resolves after gc".
    ///
    /// MEASURED on the corpus: a ref on the root commit leaves **130** live
    /// objects of 1352 — that commit, its tree and the 128 blobs the tree names
    /// — and the other 1222 rows go. (It read "5 of 2687" while the corpus was
    /// whatever pack this box's `read_dir` named first. Both numbers are
    /// properties of the fixture's root commit, which is exactly why one of
    /// them used to change when an unrelated repository was cloned.)
    #[test]
    fn gc_computes_reachability_drops_the_dead_rows_then_compacts() {
        let dir = tmpdir("gc");
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack, _) = corpus_pack();
        store.put(&pack, &[]).unwrap();
        store.absorb_pending().unwrap();

        // A ref on one commit, and an object that commit cannot reach.
        let graph = store.graph_snapshot();
        let root = graph
            .iter()
            .find(|c| c.generation == 1)
            .expect("a root commit")
            .clone();
        let root_raw = hex::decode(&root.oid).unwrap();
        let live_from_root: std::collections::HashSet<Vec<u8>> = store
            .reachable(&[&root_raw], &[])
            .unwrap()
            .into_iter()
            .collect();
        let all: Vec<Vec<u8>> = store.index().oids_in_order().unwrap();
        let dead = all
            .iter()
            .find(|o| !live_from_root.contains(*o))
            .expect("the root does not reach everything in a real repository")
            .clone();

        store
            .update_ref("refs/heads/root", None, Some(&root_raw))
            .unwrap();

        // A real znippy archive for step 3 to compact.
        let files = vec![
            ("pack-0.pack".to_string(), pack.clone()),
            ("pack-1.pack".to_string(), pack.clone()),
        ];
        znippy_common::create_archive(store.archive_path(), &files, 3).unwrap();

        let before = store.index().len();
        let report = store.gc().unwrap();

        assert!(
            store.has(&root_raw).unwrap(),
            "gc dropped the live object too"
        );
        assert!(
            !store.has(&dead).unwrap(),
            "the dead object still resolves after gc"
        );
        assert!(
            store.index().len() < before,
            "gc dropped nothing: {} rows before and after",
            before
        );
        assert_eq!(
            store.index().len(),
            live_from_root.len(),
            "exactly the live set survives"
        );
        assert_eq!(report.strategy, "NewGeneration");
        assert!(report.verified, "the new generation was not read back");
        assert!(report.archive.exists(), "the new generation is not on disk");
        assert!(
            !store.archive_path().exists(),
            "the old generation was not retired"
        );
        eprintln!(
            "gc: {before} rows → {} live, {} → {} bytes, {}",
            store.index().len(),
            report.bytes_before,
            report.bytes_after,
            report.archive.display()
        );

        // And a repository with no refs at all is refused rather than emptied.
        let dir2 = tmpdir("gc-norefs");
        let empty = GitStore::open(&dir2, "alice").unwrap();
        empty.put(&pack, &[]).unwrap();
        empty.absorb_pending().unwrap();
        let n = empty.index().len();
        assert!(empty.gc().is_err(), "a GC that would delete everything ran");
        assert_eq!(empty.index().len(), n, "the refused GC dropped rows anyway");
    }

    // ── the GC's journal half: a pack that is *wholly* dead ──────────────────

    /// What [`one_live_pack_and_one_doomed_pack`] hands back: an open store, and
    /// the three oids the guards ask about.
    struct DoomedFixture {
        dir: PathBuf,
        store: GitStore,
        /// The only object of the pack that dies whole.
        doomed: Vec<u8>,
        /// The commit a ref points at.
        root: Vec<u8>,
        /// That commit's tree — a live object that is not the ref itself.
        tree: Vec<u8>,
    }

    /// One repository with two packs: a real one that a ref reaches into, and a
    /// one-blob pack nothing will ever point at.
    ///
    /// The second pack is what makes the difference visible: after a GC it has
    /// **no rows at all**, which is the state a pack that was in flight when the
    /// machine died also leaves behind.
    ///
    /// The store is returned **open**, so a guard runs its first GC in the same
    /// process lifetime that pushed. The reopen is the thing under test and it
    /// belongs in the guard, not in the fixture.
    fn one_live_pack_and_one_doomed_pack(tag: &str) -> DoomedFixture {
        let dir = tmpdir(tag);
        let (pack, _) = corpus_pack();
        let (doomed, doomed_oid) = one_blob_pack(b"no ref will ever reach this blob");
        let store = GitStore::open(&dir, "alice").unwrap();
        store.put(&pack, &[]).unwrap();
        store.put_pack(&doomed).unwrap();
        store.absorb_pending().unwrap();
        assert!(
            store.has(&doomed_oid).unwrap(),
            "the doomed blob was never stored, so this fixture proves nothing"
        );

        let root = store
            .graph_snapshot()
            .into_iter()
            .find(|c| c.generation == 1)
            .expect("a root commit");
        let root_raw = hex::decode(&root.oid).unwrap();
        store
            .update_ref("refs/heads/root", None, Some(&root_raw))
            .unwrap();
        let tree = hex::decode(root.tree.as_ref().expect("the root commit names a tree")).unwrap();
        assert!(store.has(&tree).unwrap(), "the root's tree is not indexed");

        // A real znippy archive, so the compaction step has something to compact.
        let files = vec![
            ("pack-0.pack".to_string(), pack.clone()),
            ("pack-1.pack".to_string(), pack.clone()),
        ];
        znippy_common::create_archive(store.archive_path(), &files, 3).unwrap();
        DoomedFixture {
            dir,
            store,
            doomed: doomed_oid,
            root: root_raw,
            tree,
        }
    }
    /// **A row that arrives after the GC took its snapshot survives the drop.**
    /// `live_set` walks from the refs as they are; a push landing during the GC
    /// appends rows no ref names yet, and to `live` they are unknown. The old
    /// `retain(|o| live.contains(o))` dropped them — a GC deleting a push in
    /// flight. Now the GC drops `seen − live`, and a row in neither set is kept.
    ///
    /// Staged without a concurrent push: one synthetic unreferenced row is
    /// appended BEFORE the snapshot (seen, not live → dead), one AFTER it (late
    /// → must survive), and the drop is asked to do its worst.
    ///
    /// # 🔴 Seen red
    ///
    /// With the snapshot taken after the late append — which is exactly what
    /// "dead = not in live, over the current index" amounted to — the late row
    /// lands in `dead`: "the late row is in neither set and is not dead" fires,
    /// and had it not, `store.has(&late)` would read false one line further
    /// down. Restored.
    #[test]
    fn a_row_appended_after_the_gc_scanned_survives_the_drop() {
        let dir = tmpdir("gc-late-row");
        let (pack, _rows) = corpus_pack();
        let store = GitStore::open(&dir, "alice").unwrap();
        store.put_pack(&pack).unwrap();
        store.wait_indexed();
        // A ref, or `live_set` refuses to compute a reachability that would
        // empty the repository — the right refusal, and not the one under test.
        let root = store
            .graph_snapshot()
            .into_iter()
            .find(|c| c.generation == 1)
            .expect("a root commit");
        let root_raw = hex::decode(&root.oid).unwrap();
        store
            .update_ref("refs/heads/root", None, Some(&root_raw))
            .unwrap();
        let fake = |b: u8| crate::index_layout::IndexEntry {
            oid: vec![b; 20],
            offset: 0,
            len: 0,
            obj_type: crate::index_layout::ObjType::Blob,
            uncompressed_size: 0,
            delta_base: 0,
        };
        let early = fake(0xe1);
        let late = fake(0x1a);
        store.index().append(std::slice::from_ref(&early)).unwrap();

        // The GC's two snapshots, then reachability — in that order.
        let seen = store.index().oids_in_order().unwrap();
        let live = store.live_set().unwrap();
        // A push lands while the GC thinks.
        store.index().append(std::slice::from_ref(&late)).unwrap();

        let dead = crate::git_ops::dead_oids(seen, &live);
        assert!(dead.contains(&early.oid), "the early unreferenced row is dead");
        assert!(!dead.contains(&late.oid), "the late row is in neither set and is not dead");
        let dropped = store.drop_dead_rows(&dead).unwrap();
        assert!(dropped >= 1, "the drop must have dropped the early row at least");
        assert!(!store.has(&early.oid).unwrap(), "the dead row must be gone");
        assert!(
            store.has(&late.oid).unwrap(),
            "a row appended after the GC's snapshot was dropped"
        );
    }


    /// The journal's rows, split into packs and tombstones, read off the disk.
    fn journal_state(dir: &Path) -> (Vec<(u64, u64)>, Vec<u64>) {
        use crate::archive_write::{acked_packs, read_journal, retired_offsets, SafeWriter};
        let rows = read_journal(&SafeWriter::journal_path(&dir.join("objects.pack"))).unwrap();
        let mut retired: Vec<u64> = retired_offsets(&rows).into_iter().collect();
        retired.sort_unstable();
        (acked_packs(&rows), retired)
    }

    /// **THE BUG: a pack whose every object a GC found dead came back on the
    /// next open.**
    ///
    /// The `indexed` bit is derived as *extent in the journal, rows not in the
    /// index*, and a GC that drops every row of a pack produces exactly that
    /// state — so the reopen re-queued the pack and re-absorbed the objects the
    /// GC had just decided were dead. A partly dead pack keeps rows and was never
    /// affected, which is why this needs a pack that dies **whole**.
    ///
    /// Asserted on applied output on both sides of a process-lifetime boundary:
    /// the tombstone rows on disk, `has()` on the dead oid, and the object count
    /// after the reopen's drain has been waited for — a resurrection puts the row
    /// back, so counting is not a proxy for it, it *is* it.
    ///
    /// Seen RED by making `Absorber::adopt_journal` ignore the tombstones
    /// (`else if retired.contains(&extent.0)` → `else if false`), which is
    /// exactly the reader as it stood before this fix: "a dead object came back
    /// the instant the store was reopened". Restored.
    ///
    /// Seen RED a second time by not retiring anything in `GitOps::gc` (`let
    /// retired: Vec<u64> = Vec::new();` in place of the `retire_dead_packs`
    /// call — the writer as it stood before this fix): "gc retired 0 pack(s), not
    /// the one whose objects all died — left: 0, right: 1". Restored.
    ///
    /// Seen RED a third time by computing deadness the way a retirement written
    /// *after* the drop would have to (`occupied[i]` → `!occupied[i]` in
    /// `retire_dead_packs`): the same "gc retired 0 pack(s)" — after the drop an
    /// all-dead pack is indistinguishable from an unabsorbed one, which is the
    /// same confusion this whole change is about, one level up. Restored.
    ///
    /// And the derived half, seen RED by removing the `refold()` from
    /// `drop_dead_rows`: "the graph still holds 551 commits after gc dropped
    /// every dead one — left: 551, right: 1", with `reachable()` still selecting
    /// objects the index can no longer serve. Restored.
    #[test]
    fn a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened() {
        let DoomedFixture {
            dir,
            store,
            doomed: doomed_oid,
            root: root_raw,
            tree,
        } = one_live_pack_and_one_doomed_pack("gc-wholly-dead");

        let (rows_after_gc, retired_by_gc) = {
            let report = store.gc().unwrap();
            assert_eq!(
                report.retired_packs, 1,
                "gc retired {} pack(s), not the one whose objects all died",
                report.retired_packs
            );
            assert!(
                !store.has(&doomed_oid).unwrap(),
                "the dead object still resolves in the store that dropped it"
            );
            assert!(store.has(&root_raw).unwrap(), "gc dropped the live commit");

            // **Everything the indexer derived is dropped with the rows, in the
            // same process** — not merely rebuilt by the next open. The graph
            // holds one commit, and a selection over the bitmaps and the ordinal
            // space cannot name a dead object.
            assert_eq!(
                store.commit_count(),
                1,
                "the graph still holds {} commits after gc dropped every dead one",
                store.commit_count()
            );
            let selected = store.reachable(&[&root_raw], &[]).unwrap();
            assert!(
                !selected.contains(&doomed_oid),
                "a selection after gc still names a dead object — the bitmaps or the ordinal \
                 space were not refolded"
            );
            assert_eq!(
                selected.len(),
                store.index().len(),
                "a selection after gc names {} objects and the index holds {}",
                selected.len(),
                store.index().len()
            );
            (store.index().len(), report.retired_packs)
        };
        drop(store);

        // The tombstone is on disk, and it names the doomed pack's offset — not
        // the live one's.
        let (packs, retired) = journal_state(&dir);
        assert_eq!(packs.len(), 2, "both packs are still acked: {packs:?}");
        assert_eq!(
            retired,
            vec![packs[1].0],
            "the journal retired {retired:?}, and the doomed pack starts at {}",
            packs[1].0
        );
        assert_eq!(retired.len() as u64, retired_by_gc);

        // ── the reopen: the whole point ──────────────────────────────────────
        let store = GitStore::open(&dir, "alice").unwrap();
        assert!(
            !store.has(&doomed_oid).unwrap(),
            "a dead object came back the instant the store was reopened"
        );
        store.wait_indexed();
        assert!(
            !store.has(&doomed_oid).unwrap(),
            "the reopen re-queued the wholly dead pack and its objects came back"
        );
        assert_eq!(
            store.index().len(),
            rows_after_gc,
            "the dead objects came back on the reopen: the store that dropped them holds \
             {rows_after_gc} rows, the reopened one holds {}",
            store.index().len()
        );
        assert_eq!(
            store.unindexed_packs(),
            0,
            "the retired pack is queued as work"
        );
        assert_eq!(
            store.absorb_pending().unwrap(),
            0,
            "a read falling back would re-absorb the retired pack"
        );
        // And the live side of the same repository is untouched by all of it.
        assert!(store.has(&root_raw).unwrap(), "the live commit is gone");
        assert!(store.has(&tree).unwrap(), "the live tree is gone");
        assert_eq!(
            store.commit_count(),
            1,
            "the graph after gc + reopen holds {} commits, not the one live one",
            store.commit_count()
        );
    }

    /// **A partly dead pack is not retired, and comes through a GC and a reopen
    /// with its live objects.**
    ///
    /// This is the case that already worked and must keep working: the pack still
    /// has rows, so the crash-recovery diff calls it absorbed and nothing about
    /// it changes. Asserted on applied output: **no** tombstone appears in the
    /// journal, the live objects read back after the reopen, and the dead ones
    /// stay dead.
    ///
    /// Seen RED by retiring on `has_live[i]` instead of `!has_live[i]` in
    /// `retire_dead_packs` — i.e. tombstoning the packs that are *alive*: "gc
    /// retired the partly dead pack — its live objects are one reopen from a
    /// fallback that will never come — left: 1, right: 0", and the journal
    /// assertion below holds the same finding in on-disk bytes. The live objects
    /// still read back under that mutation, because a pack with rows is adopted
    /// as absorbed whatever the journal says about it — which is exactly why this
    /// guard asserts on the journal and not only on the reads. Restored.
    #[test]
    fn a_partly_dead_pack_is_never_retired_and_survives_a_reopen() {
        let dir = tmpdir("gc-partly-dead");
        let (pack, _) = corpus_pack();
        let store = GitStore::open(&dir, "alice").unwrap();
        store.put(&pack, &[]).unwrap();
        store.absorb_pending().unwrap();

        let root = store
            .graph_snapshot()
            .into_iter()
            .find(|c| c.generation == 1)
            .expect("a root commit");
        let root_raw = hex::decode(&root.oid).unwrap();
        let live: std::collections::HashSet<Vec<u8>> = store
            .reachable(&[&root_raw], &[])
            .unwrap()
            .into_iter()
            .collect();
        let all: Vec<Vec<u8>> = store.index().oids_in_order().unwrap();
        let dead = all
            .iter()
            .find(|o| !live.contains(*o))
            .expect("a real pack holds more than one commit's closure")
            .clone();
        assert!(
            live.len() < all.len(),
            "the fixture pack is not partly dead: {} live of {}",
            live.len(),
            all.len()
        );
        store
            .update_ref("refs/heads/root", None, Some(&root_raw))
            .unwrap();
        let files = vec![("pack-0.pack".to_string(), pack.clone())];
        znippy_common::create_archive(store.archive_path(), &files, 3).unwrap();

        let report = store.gc().unwrap();
        assert_eq!(
            report.retired_packs, 0,
            "gc retired the partly dead pack — its live objects are one reopen from a fallback \
             that will never come"
        );
        let rows_after_gc = store.index().len();
        drop(store);

        let (packs, retired) = journal_state(&dir);
        assert_eq!(packs.len(), 1);
        assert!(
            retired.is_empty(),
            "gc tombstoned a partly dead pack: {retired:?}"
        );

        let store = GitStore::open(&dir, "alice").unwrap();
        store.wait_indexed();
        assert_eq!(
            store.index().len(),
            rows_after_gc,
            "the reopen changed the row count of a partly dead pack's repository"
        );
        for oid in &live {
            assert!(
                store.has(oid).unwrap(),
                "the live object {} did not survive gc + reopen",
                hex::encode(oid)
            );
        }
        assert!(
            !store.has(&dead).unwrap(),
            "a dead object came back on the reopen of a partly dead pack"
        );
        // The pack was adopted as absorbed rather than re-queued: no work is
        // owed, and a read falling back would re-absorb it.
        assert_eq!(
            store.unindexed_packs(),
            0,
            "the surviving pack was re-queued"
        );
        assert_eq!(store.absorb_pending().unwrap(), 0);
    }

    /// **An interruption at each step of the retirement leaves a readable
    /// archive and loses nothing.**
    ///
    /// The two steps are run as `GitOps::gc` runs them and the process boundary
    /// is a real one — the store is dropped and reopened between them, which is
    /// what a `kill -9` there would leave, because everything each step writes is
    /// fsynced before it returns.
    ///
    /// | killed after | on disk | what must be true |
    /// |---|---|---|
    /// | the tombstone | rows still there | **nothing is lost** — every object still reads, dead ones included |
    /// | the drop | rows gone, no compaction | the dead objects are gone and **stay** gone |
    /// | the compaction | a new generation | `gc.rs`'s own `an_interruption_at_every_step_leaves_a_readable_archive` |
    ///
    /// The first row is why the tombstone goes **first**: a GC killed there is a
    /// GC that did not happen, and the repository is exactly as it was. The
    /// reverse order has a window where the rows are gone and the journal still
    /// claims an unabsorbed pack, which is the resurrection this whole change is
    /// about.
    ///
    /// The archive itself is read back through the ordinary reader at every
    /// checkpoint, so "readable" is the whole file decoding, not a stat.
    ///
    /// Seen RED by computing deadness from `!occupied[i]` in `retire_dead_packs`
    /// — the shape a retirement written after the drop needs, which cannot see a
    /// pack that still has its rows: "the doomed pack was not retired: [] —
    /// left: 0, right: 1", at the first checkpoint, with every row still in the
    /// index. Restored.
    #[test]
    fn an_interruption_at_each_step_of_the_retirement_loses_nothing() {
        let DoomedFixture {
            dir,
            store,
            doomed: doomed_oid,
            root: root_raw,
            tree,
        } = one_live_pack_and_one_doomed_pack("gc-interrupt");
        let archive = dir.join("repository.znippy");
        let readable = |where_: &str| {
            crate::gc::read_back_every_entry(&archive)
                .unwrap_or_else(|e| panic!("the archive does not read back {where_}: {e:?}"));
        };
        readable("before anything ran");

        // ── killed right after the tombstone, before any row was dropped ─────
        //
        // The live set is computed once and carried, which is what `gc()` does
        // with it too: it is an input to both steps, and computing it twice would
        // make this guard depend on a reachability fold rather than on the two
        // mutations it is here to interrupt.
        let (live, dead, rows_before) = {
            let seen = store.index().oids_in_order().unwrap();
            let live = store.live_set().unwrap();
            let dead = crate::git_ops::dead_oids(seen, &live);
            let retired = store.retire_dead_packs(&live, u64::MAX).unwrap();
            assert_eq!(
                retired.len(),
                1,
                "the doomed pack was not retired: {retired:?}"
            );
            let n = store.index().len();
            assert!(
                store.has(&doomed_oid).unwrap(),
                "retiring the journal row dropped an index row — the two steps are not separable"
            );
            (live, dead, n)
        };
        drop(store);
        let (packs, on_disk) = journal_state(&dir);
        assert_eq!(
            on_disk,
            vec![packs[1].0],
            "the tombstone is not on disk before a single row was dropped: {on_disk:?}"
        );
        readable("after the tombstone");

        {
            // The reopen a crash there produces: everything is still here.
            let store = GitStore::open(&dir, "alice").unwrap();
            store.wait_indexed();
            assert_eq!(
                store.index().len(),
                rows_before,
                "a crash between the tombstone and the drop lost rows"
            );
            assert!(
                store.has(&doomed_oid).unwrap(),
                "a crash between the tombstone and the drop lost the objects of the pack it \
                 retired — the GC had not decided anything yet"
            );
            assert!(store.has(&root_raw).unwrap() && store.has(&tree).unwrap());
            // And resuming is idempotent: the pack is already retired.
            assert!(
                store.retire_dead_packs(&live, u64::MAX).unwrap().is_empty(),
                "the resumed GC retired the same pack twice"
            );
        }

        // ── killed after the drop, before the compaction ─────────────────────
        let rows_after_drop = {
            let store = GitStore::open(&dir, "alice").unwrap();
            let dropped = store.drop_dead_rows(&dead).unwrap();
            assert!(dropped > 0, "the drop dropped nothing");
            assert!(!store.has(&doomed_oid).unwrap());
            store.index().len()
        };
        readable("after the drop, before the compaction");
        let (_, still_retired) = journal_state(&dir);
        assert_eq!(still_retired.len(), 1, "the drop lost the tombstone");

        {
            let store = GitStore::open(&dir, "alice").unwrap();
            store.wait_indexed();
            assert!(
                !store.has(&doomed_oid).unwrap(),
                "a crash between the drop and the compaction resurrected the dead objects"
            );
            assert_eq!(store.index().len(), rows_after_drop);
            assert!(store.has(&root_raw).unwrap() && store.has(&tree).unwrap());

            // ── and the compaction still completes on the resumed store ──────
            let report = store.gc().unwrap();
            assert_eq!(
                report.retired_packs, 0,
                "the resumed gc retired an already-retired pack"
            );
            assert!(
                report.verified,
                "the resumed gc did not verify its generation"
            );
            assert!(report.archive.exists());
            crate::gc::read_back_every_entry(&report.archive)
                .expect("the new generation does not read back");
            assert!(!store.has(&doomed_oid).unwrap());
            assert!(store.has(&root_raw).unwrap());
        }
    }

    /// `compact` leaves its source untouched and refuses to overwrite.
    ///
    /// Seen RED by removing the `dst.exists()` refusal: "the refusal must be the
    /// explicit one and must name the file: hard-linking …/repo.znippy to
    /// …/repo.g1.znippy — compaction runs against a second name for the same
    /// inode …". The first version of this guard only asserted *that* it failed,
    /// and that version stayed green under the same mutation, because `hard_link`
    /// refuses EEXIST as well. It now asserts which mechanism refused.
    #[test]
    fn compact_names_its_destination_and_never_overwrites() {
        let dir = tmpdir("compact");
        let src = dir.join("repo.znippy");
        let dst = dir.join("repo.g1.znippy");
        let files = vec![
            ("pack-0.pack".to_string(), vec![7u8; 400_000]),
            ("pack-1.pack".to_string(), vec![9u8; 400_000]),
        ];
        znippy_common::create_archive(&src, &files, 3).unwrap();
        let before = std::fs::read(&src).unwrap();

        compact(&src, &dst).unwrap();
        assert!(dst.exists(), "compact produced no destination");
        assert_eq!(
            std::fs::read(&src).unwrap(),
            before,
            "compact modified its source"
        );

        let err = compact(&src, &dst).expect_err("dst exists");
        // Asserted on WHICH mechanism refused, not merely that something did:
        // `hard_link` would fail with EEXIST too, so a guard that only checked
        // for an error could not tell the explicit refusal from the syscall's —
        // and the difference is whether a filesystem operation is attempted
        // against a file that must not be touched at all.
        assert!(
            err.to_string().contains("refusing to compact over it")
                && err.to_string().contains(&dst.display().to_string()),
            "the refusal must be the explicit one and must name the file: {err}"
        );
        assert_eq!(
            std::fs::read(&src).unwrap(),
            before,
            "the refused second compaction touched the source"
        );
    }
}

#[cfg(test)]
mod emit_set_tests {
    use crate::index_layout::ObjectIndex as _;
    use crate::pack_walk::topological_order;
    use crate::store::tests::{corpus_pack, tmpdir};
    use crate::{GitOps, GitStore};

    /// **A SUBSET emits a pack holding EXACTLY the subset, and stock git accepts
    /// it.**
    ///
    /// Emitting a whole pack proves the encoder; it cannot prove anything about
    /// a request whose delta bases fall outside it, because every base is
    /// present by construction. A subset is that case.
    ///
    /// # What this asserted until 2026-08-11, and why it inverted
    ///
    /// It asserted `entries.len() >= half.len()` under the heading *"closure
    /// pulled in nothing"* — the set was expected to come back **larger**,
    /// because `emit_set` added the delta bases. That is the defect: a base
    /// pulled into a narrowed request is an object the client did not ask for,
    /// and a tree pulled in that way owes children the pack does not carry. See
    /// [`GitStore::emit_set`] for the production failure and the counts.
    ///
    /// So the assertion is now an equality, and it is the strong direction: the
    /// emitted set is the requested set, entry for entry. The delta whose base
    /// was left out is emitted **whole**, which is what `pack-objects` does and
    /// what keeps the pack connected.
    #[test]
    fn a_subset_emits_exactly_the_subset_and_stock_git_accepts_it() {
        if std::process::Command::new("git")
            .arg("--version")
            .output()
            .is_err()
        {
            eprintln!("skipping: no git on PATH");
            return;
        }
        let dir = tmpdir("emit-subset");
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack, _) = corpus_pack();
        store.put_pack(&pack).unwrap();
        store.absorb_pending().unwrap();

        let all = store.index().oids_in_order().unwrap();
        assert!(all.len() > 8, "corpus too small to take a subset of");
        // Every other object: enough deltas land with their base excluded that
        // the boundary rule has real work to do.
        let half: Vec<&[u8]> = all.iter().step_by(2).map(Vec::as_slice).collect();
        let rows = store.index().lookup_batch(&half);
        let inside: std::collections::HashSet<u64> =
            rows.iter().flatten().map(|r| r.offset).collect();
        let boundary = rows
            .iter()
            .flatten()
            .filter(|r| r.delta_base != 0 && !inside.contains(&r.delta_base))
            .count();
        assert!(
            boundary > 0,
            "this subset excludes no delta base, so it cannot tell an exact emission from a \
             closed one and the test proves nothing"
        );

        let entries = store.emit_set(&half, true, None).unwrap();
        assert_eq!(
            entries.len(),
            half.len(),
            "the emitted set must be the requested set — {} asked, {} emitted",
            half.len(),
            entries.len()
        );
        assert_eq!(
            entries.iter().filter(|e| e.recompressed).count(),
            boundary,
            "exactly the entries whose base was excluded must be rebuilt whole"
        );

        let (ordered, missing) = topological_order(entries);
        assert!(
            missing.is_empty(),
            "no entry may still name a base that is not here: {missing:?}"
        );

        // Walk what we emitted, before handing it to git. If the header count
        // and the body disagree, that is the shape that crashes an indexer.
        {
            let mut b = Vec::new();
            store.emit_ordered(&ordered, &mut b).unwrap();
            let declared = u32::from_be_bytes([b[8], b[9], b[10], b[11]]);
            match crate::pack_walk::walk(&b, store.hash_kind().oid_len()) {
                Ok(w) => assert_eq!(
                    w.entries.len(),
                    declared as usize,
                    "the pack header declares {declared} entries and the body walks to {}",
                    w.entries.len()
                ),
                Err(e) => panic!("our own walk cannot read what we emitted: {e:#}"),
            }
        }

        let mut bytes = Vec::new();
        let report = store.emit_ordered(&ordered, &mut bytes).unwrap();
        assert_eq!(
            report.copied + report.recompressed,
            report.written,
            "every entry is either copied or rebuilt, and the receipt must add up"
        );
        assert_eq!(
            report.recompressed as usize, boundary,
            "only the excluded-base entries may be rebuilt"
        );

        // **No `--strict` here, and that is deliberate.** An arbitrary every-other
        // subset is not reachability-closed — a tree in it will name a blob that
        // is not — so `--strict`'s connectivity walk would be judging the
        // *caller's* selection, not this emitter. What is being asserted is
        // SELF-CONTAINMENT: that every delta in the pack resolves inside it,
        // which plain `index-pack` answers with `pack has N unresolved deltas`.
        // Connectivity has its own test, on a request that is closed:
        // `a_narrowed_clone_is_served_a_connected_pack_not_its_delta_bases`.
        let out_dir = tmpdir("emit-subset-idx");
        crate::git_oracle::assert_git_accepts(
            &out_dir,
            "dst.git",
            &bytes,
            crate::git_oracle::Strictness::SelfContained,
        );
    }

    /// 🔴 **An oid this repository does not hold is REFUSED by name — it is not
    /// quietly left out of the pack.**
    ///
    /// `emit_set` used to `continue` past a `lookup_batch` miss. That is the
    /// silent-under-send shape: the request asks for N objects, the pack carries
    /// N-1, `PackStats` reports success and the *client* is the first to notice.
    /// It also made the two engines behind one contract disagree — gunnar's
    /// in-memory arm has always refused with *"emit_pack was asked for X, which
    /// this store does not hold"* — so a benchmark across the two was comparing
    /// two different contracts.
    ///
    /// # Why the assertions are shaped this way
    ///
    /// **RED before the change** on the first one: `emit_set` returned `Ok` with
    /// two entries for three asked oids, so `expect_err` panicked.
    ///
    /// The error must **name the oid**, because an operator reading
    /// `git.upload_pack.failed` needs the object, not the fact that something
    /// was missing; asserted on the rendered message.
    ///
    /// And the refusal must be *conditional*, or a function that errored
    /// unconditionally would pass the first two: the same store, the same two
    /// real oids and no absent one, must still emit.
    #[test]
    fn an_oid_the_store_does_not_hold_is_refused_by_name_not_skipped() {
        let dir = tmpdir("emit-missing");
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack, _) = corpus_pack();
        store.put_pack(&pack).unwrap();
        store.absorb_pending().unwrap();

        let all = store.index().oids_in_order().unwrap();
        assert!(all.len() > 2, "corpus too small");
        // A well-formed oid of the right width that this store cannot hold: it
        // is not in the index, and asking for it is what a caller with a stale
        // set does.
        let absent = vec![0xABu8; store.hash_kind().oid_len()];
        assert!(
            store.index().lookup(&absent).is_none(),
            "the fixture oid must really be absent or this test asserts nothing"
        );

        let asked: Vec<&[u8]> = vec![all[0].as_slice(), absent.as_slice(), all[1].as_slice()];
        let err = store
            .emit_set(&asked, true, None)
            .expect_err("a pack for an object the store lacks must be refused, not shortened");
        let msg = format!("{err:#}");
        assert!(
            msg.contains(&hex::encode(&absent)),
            "the refusal must name the oid it could not find; got: {msg}"
        );

        // Conditional, not unconditional: drop the absent one and the same call
        // emits both objects.
        let present: Vec<&[u8]> = vec![all[0].as_slice(), all[1].as_slice()];
        let entries = store
            .emit_set(&present, true, None)
            .expect("two objects this store does hold must still emit");
        assert_eq!(entries.len(), 2);
    }
}

/// **The zero-copy emit path: the same bytes, and none of the memory.**
///
/// [`crate::pack_walk::EntryBytes`] turned `EmitEntry.stored` from a `Vec<u8>`
/// that owned a `pread`ed copy of the entry into a 16-byte address resolved
/// against a mapping of the archive. That is a change to *where the bytes live*
/// and to **nothing else**, so the guards here are of two kinds: three that
/// require the emitted bytes to be unchanged, and one that requires the memory
/// to be gone. Either kind alone would pass on a broken change — an emitter that
/// held nothing and wrote garbage, or one that wrote perfectly and still held
/// the repository.
#[cfg(test)]
mod zero_copy_tests {
    use crate::index_layout::{ObjType, ObjectIndex as _};
    use crate::pack_walk::{EmitEntry, EntryBytes, topological_order};
    use crate::store::tests::{corpus_pack, tmpdir};
    use crate::{GitOps, GitStore};

    /// A store holding the whole corpus pack, and every oid in it.
    fn corpus(name: &str) -> (std::path::PathBuf, GitStore, Vec<Vec<u8>>) {
        let dir = tmpdir(name);
        let store = GitStore::open(&dir, "alice").unwrap();
        let (pack, _) = corpus_pack();
        store.put_pack(&pack).unwrap();
        store.absorb_pending().unwrap();
        let all = store.index().oids_in_order().unwrap();
        assert!(all.len() > 100, "the corpus is too small to prove anything");
        (dir, store, all)
    }

    /// The **old** shape of an emit set: every entry's bytes `pread` into an
    /// owned `Vec`, exactly as `emit_set` built them before 2026-08-14. The
    /// differential guards below emit this and the extent form and require the
    /// two packs to be identical.
    ///
    /// It reads through [`GitStore::read_extent`] — the `pread` path — so it
    /// shares no code with the mapping it is being compared against. A helper
    /// that resolved through `Mapped::get` would be comparing the change with
    /// itself.
    fn materialised(store: &GitStore, entries: &[EmitEntry]) -> Vec<EmitEntry> {
        entries
            .iter()
            .map(|e| {
                let mut e = e.clone();
                if let EntryBytes::Extent { offset, len } = e.stored {
                    e.stored = EntryBytes::Owned(store.read_extent(offset, len).unwrap());
                }
                e
            })
            .collect()
    }

    /// 🔴 **A full clone's pack is byte-identical before and after the change.**
    ///
    /// The strongest guard available, and the one the whole change stands on: a
    /// clone is built twice out of the same selection — once with every entry an
    /// `EntryBytes::Extent` resolved through the mapping, once with every entry
    /// `pread` into an owned `Vec` the way `emit_set` used to build it — and the
    /// two packs must be the same bytes. Not the same length, not the same
    /// object count: the same **bytes**, trailer included, which for a pack
    /// means the same headers, the same `OFS_DELTA` distances and the same sha1.
    ///
    /// A whole-repository selection, so `recompressed` is 0 and every one of the
    /// entries really is an extent — asserted, because a run in which they were
    /// all `Owned` would compare the old path with itself and pass vacuously.
    ///
    /// Seen RED by returning `snap.get(offset, len - 1)` from
    /// [`GitStore::extent`] — a mapping that hands back a *short* slice rather
    /// than `None`, which is the single most plausible way to get this wrong:
    /// "the mapped emit and the `pread` emit disagree at byte 1944: 0xe8 vs
    /// 0x3a (mapped 5651349 bytes, pread 5654037 bytes)".
    ///
    /// Seen RED a second way, by slicing `snap.get(offset + 1, len)` — an
    /// extent addressed one byte off: "resolving the emit set's stored bytes:
    /// resolving the stored bytes of 00065363b6d5f3edb20c8c17a5938503fcdcc941
    /// for emission: object type code 0 is not one git writes — refusing to
    /// guess it". That is phase 1's header parse refusing before phase 2 had
    /// written a byte of the pack, which is why the parse is there.
    #[test]
    fn a_full_clone_emits_the_identical_pack_from_extents_and_from_owned_bytes() {
        let (_dir, store, all) = corpus("zc-identical");
        let oids: Vec<&[u8]> = all.iter().map(Vec::as_slice).collect();

        let entries = store.emit_set(&oids, true, None).unwrap();
        let extents = entries
            .iter()
            .filter(|e| matches!(e.stored, EntryBytes::Extent { .. }))
            .count();
        assert_eq!(
            extents,
            entries.len(),
            "a whole-repository selection must be ALL extents — {} of {} were owned, so this \
             comparison would be the old path against itself",
            entries.len() - extents,
            entries.len()
        );

        let owned = materialised(&store, &entries);
        let (ordered_a, missing) = topological_order(entries);
        assert!(missing.is_empty(), "a full clone is closed: {missing:?}");
        let (ordered_b, _) = topological_order(owned);

        let mut mapped_pack = Vec::new();
        let report = store.emit_ordered(&ordered_a, &mut mapped_pack).unwrap();
        let mut pread_pack = Vec::new();
        store.emit_ordered(&ordered_b, &mut pread_pack).unwrap();

        assert_eq!(report.recompressed, 0, "a full clone re-deflates nothing");
        assert_eq!(report.copied, report.written);
        assert!(mapped_pack.len() > 1_000_000, "the corpus pack is tiny?");
        if mapped_pack != pread_pack {
            let at = mapped_pack
                .iter()
                .zip(pread_pack.iter())
                .position(|(a, b)| a != b);
            let i = at.unwrap_or(mapped_pack.len().min(pread_pack.len()));
            panic!(
                "the mapped emit and the `pread` emit disagree at byte {i}: {:#04x} vs {:#04x} \
                 (mapped {} bytes, pread {} bytes)",
                mapped_pack.get(i).copied().unwrap_or(0),
                pread_pack.get(i).copied().unwrap_or(0),
                mapped_pack.len(),
                pread_pack.len()
            );
        }
    }

    /// 🔴 **Every `OFS_DELTA`'s base distance still lands on its base.**
    ///
    /// The one thing phase 2 may not be parallelised around, asserted as applied
    /// output rather than as a receipt: the emitted pack is walked back, and for
    /// every `OfsDelta` entry the base its distance names must be an entry
    /// boundary in the same pack — which `PackWalk::closure` answers, and which
    /// no amount of correct-looking header encoding can satisfy by accident.
    /// `git index-pack` agrees separately in the tests beside this one; this is
    /// the guard that says *which* thing broke when it stops agreeing.
    ///
    /// The count of `OfsDelta` entries is asserted first. A pack with none would
    /// pass this vacuously, and a corpus that happened to hold none would make
    /// the whole of phase 2's serial argument untestable.
    ///
    /// Seen RED by encoding the distance as `here - base_at + 1` in
    /// `emit_pack`: "1345 of 1345 ofs-deltas name a base that is not an entry
    /// boundary — first at output offset 11". Seen RED a second, sharper way by
    /// emitting `emit_set`'s entries in their input order rather than
    /// `topological_order`'s: "entry at 3414738 deltas against archive offset
    /// 3414009, which is not in this pack — the set is not closed and was not
    /// ordered by `topological_order`".
    #[test]
    fn an_ofs_delta_still_names_its_base_by_the_right_distance() {
        let (_dir, store, all) = corpus("zc-ofs");
        let oids: Vec<&[u8]> = all.iter().map(Vec::as_slice).collect();
        let (ordered, _) = topological_order(store.emit_set(&oids, true, None).unwrap());

        let mut pack = Vec::new();
        store.emit_ordered(&ordered, &mut pack).unwrap();

        let walk = crate::pack_walk::walk(&pack, store.hash_kind().oid_len())
            .expect("our own walk must read what we emitted");
        let deltas = walk
            .entries
            .iter()
            .filter(|e| e.obj_type == ObjType::OfsDelta)
            .count();
        assert!(
            deltas > 0,
            "the corpus emitted no ofs-delta, so this proves nothing about distances"
        );

        let closure = walk.closure();
        assert!(
            closure.broken_offsets.is_empty(),
            "{} of {deltas} ofs-deltas name a base that is not an entry boundary — first at \
             output offset {}",
            closure.broken_offsets.len(),
            closure.broken_offsets.first().copied().unwrap_or(0)
        );
        assert!(
            closure.external_refs.is_empty(),
            "a full clone must name no external base"
        );
    }

    /// 🔴 **The `Owned` exception still carries a rebuilt delta's bytes.**
    ///
    /// `EntryBytes::Owned` is not vestigial: a delta whose base the request does
    /// not carry genuinely computes bytes that no extent addresses, and dropping
    /// the variant would have meant either shipping a delta the client cannot
    /// resolve or widening the request — the two failures
    /// `GitStore::emit_set`'s doc comment is written about.
    ///
    /// Asserted on applied output at both ends: the boundary entries are `Owned`
    /// and every other entry is an `Extent` (a mixed set, so the emitter is
    /// proved to handle both in one pack), and the resulting pack walks with
    /// its closure intact. The premise — that this subset really does cut delta
    /// chains — is asserted first.
    ///
    /// Seen RED by making the whole-rebuild arm of `emit_set` push
    /// `EntryBytes::Extent { offset: row.offset, len: row.len }` instead of the
    /// bytes it just built: "312 boundary entries are recompressed but 202 carry
    /// owned bytes — a rebuilt entry that points back at its stored extent ships
    /// the delta it was rebuilt to avoid". 202 and not 0, because the re-delta
    /// arm above it was untouched — which is what makes the count, and not a
    /// bare `is_some()`, the thing worth asserting.
    #[test]
    fn a_rebuilt_delta_still_carries_its_own_bytes_and_emits_beside_extents() {
        let (_dir, store, all) = corpus("zc-owned");
        let half: Vec<&[u8]> = all.iter().step_by(2).map(Vec::as_slice).collect();

        let rows = store.index().lookup_batch(&half);
        let inside: std::collections::HashSet<u64> =
            rows.iter().flatten().map(|r| r.offset).collect();
        let boundary = rows
            .iter()
            .flatten()
            .filter(|r| r.delta_base != 0 && !inside.contains(&r.delta_base))
            .count();
        assert!(
            boundary > 0,
            "this subset cuts no delta chain, so the Owned path is never reached"
        );

        let entries = store.emit_set(&half, true, None).unwrap();
        let rebuilt: Vec<&EmitEntry> = entries.iter().filter(|e| e.recompressed).collect();
        let with_bytes = rebuilt
            .iter()
            .filter(|e| e.stored.owned().is_some_and(|b| !b.is_empty()))
            .count();
        assert_eq!(
            with_bytes,
            rebuilt.len(),
            "{} boundary entries are recompressed but {with_bytes} carry owned bytes — a rebuilt \
             entry that points back at its stored extent ships the delta it was rebuilt to avoid",
            rebuilt.len()
        );
        assert_eq!(rebuilt.len(), boundary);

        // The other half of the point: this is a MIXED pack, so the emitter is
        // resolving both variants in one pass.
        let as_extent = entries
            .iter()
            .filter(|e| matches!(e.stored, EntryBytes::Extent { .. }))
            .count();
        assert_eq!(
            as_extent,
            entries.len() - rebuilt.len(),
            "every entry that was NOT rebuilt must still be an extent"
        );
        assert!(
            as_extent > 0 && !rebuilt.is_empty(),
            "the set must be mixed"
        );

        let (ordered, missing) = topological_order(entries);
        assert!(missing.is_empty(), "the emitted set is closed: {missing:?}");
        let mut pack = Vec::new();
        let report = store.emit_ordered(&ordered, &mut pack).unwrap();
        assert_eq!(report.recompressed as usize, boundary);
        let walk = crate::pack_walk::walk(&pack, store.hash_kind().oid_len())
            .expect("a mixed pack must still walk");
        assert!(
            walk.closure().broken_offsets.is_empty(),
            "a pack mixing extents and rebuilt bytes must still be self-contained"
        );
    }

    /// 🔴 **An extent past the mapping falls back to `pread` and returns the
    /// SAME bytes.**
    ///
    /// `Mapped::get` answers `None` for an extent past the mapped end, which
    /// means *"go and pread it"* and never *"there are no bytes"*. The case is
    /// real and not hypothetical: a snapshot is taken once per emission and the
    /// blob file grows whenever a push lands, so any clone running across a push
    /// reads its tail this way.
    ///
    /// Constructed deliberately rather than raced: a snapshot is taken, a second
    /// pack is pushed, and the objects of that second pack are then emitted
    /// against the **stale** snapshot. Every one of their extents is past its
    /// end, so every one takes the fallback — asserted by `Mapped::get` refusing
    /// them, so a run where the snapshot happened to cover them cannot pass
    /// quietly.
    ///
    /// Seen RED by making [`GitStore::extent`]'s `None` arm
    /// `Ok(Cow::Owned(Vec::new()))` — the "absent means no bytes" misreading
    /// [`crate::archive_map::Mapped::get`]'s own doc comment forbids: "the
    /// fallback returned 0 bytes for the extent at (5653314, 43), which the
    /// mapping does not cover; `None` means pread, not empty".
    #[test]
    fn an_extent_past_the_mapping_falls_back_and_returns_the_same_bytes() {
        let (_dir, store, _all) = corpus("zc-fallback");

        // The snapshot is taken NOW, before the second push.
        let stale = store.archive_snapshot().unwrap();
        let before = stale.len();

        let (second, _) = crate::store::tests::one_blob_pack(b"a blob pushed after the snapshot\n");
        store.put_pack(&second).unwrap();
        store.absorb_pending().unwrap();

        let after = store.archive_snapshot().unwrap().len();
        assert!(
            after > before,
            "the second push did not grow the blob file ({before} → {after}), so nothing is past \
             the stale mapping and this test proves nothing"
        );

        // Every object of the second pack sits past the stale mapping's end.
        let fresh: Vec<Vec<u8>> = store
            .index()
            .oids_in_order()
            .unwrap()
            .into_iter()
            .filter(|oid| {
                store
                    .index()
                    .lookup(oid)
                    .is_some_and(|r| r.offset >= before)
            })
            .collect();
        assert!(
            !fresh.is_empty(),
            "no object landed past the stale mapping's end"
        );

        for oid in &fresh {
            let row = store.index().lookup(oid).unwrap();
            assert!(
                stale.get(row.offset, row.len).is_none(),
                "the stale mapping must NOT cover ({}, {}) or the fallback is never taken",
                row.offset,
                row.len
            );
            let fell_back = store.extent(&stale, row.offset, row.len).unwrap();
            let preaded = store.read_extent(row.offset, row.len).unwrap();
            assert_eq!(
                fell_back.len(),
                preaded.len(),
                "the fallback returned {} bytes for the extent at ({}, {}), which the mapping does \
                 not cover; `None` means pread, not empty",
                fell_back.len(),
                row.offset,
                row.len
            );
            assert_eq!(
                fell_back.as_ref(),
                preaded.as_slice(),
                "the fallback bytes differ from the pread bytes at ({}, {})",
                row.offset,
                row.len
            );
            // A FRESH snapshot covers it, and agrees with both.
            let fresh_snap = store.archive_snapshot().unwrap();
            assert_eq!(
                fresh_snap.get(row.offset, row.len).unwrap(),
                preaded.as_slice(),
                "the remapped snapshot disagrees with the pread at ({}, {})",
                row.offset,
                row.len
            );
        }
        eprintln!(
            "{} object(s) resolved past a stale {before}-byte mapping and matched the pread",
            fresh.len()
        );
    }

    /// 🔴 **The gatling fan-out resolves the identical bytes the serial pass
    /// does** — LAW 3's primitive, over data with no lock on it.
    ///
    /// Phase 1 is the one place the serving tier fans out, and the claim that
    /// makes it safe is a claim about the *data*: the blob file is append-only
    /// and `gc` truncates nothing, so N workers reading a snapshot of it share
    /// no mutable state and need no synchronisation. This asserts the observable
    /// consequence — the same set resolved on one thread and on four gives the
    /// same slices, in the same order, pointing at the same bytes.
    ///
    /// It calls
    /// [`resolve_with`](GitStore::resolve_with) directly rather than going
    /// through `resolve_emit_payloads`, because the production threshold is 4096
    /// entries and this corpus holds 1352: through the front door the gatling
    /// arm would never run and this test would be asserting the serial path
    /// against itself. The order matters as much as the content —
    /// `gatling_for_each` self-dispatches with no barrier, so a worker finishing
    /// unit 9 before unit 3 is normal, and an implementation that collected in
    /// **completion** order would produce a pack whose entries are shuffled and
    /// whose `OFS_DELTA` distances therefore point at the wrong objects.
    ///
    /// Seen RED by having the parallel arm return
    /// `gatling_for_each(n, workers, one).into_iter().rev().collect()` — the
    /// cheapest stand-in for "results not in index order": "the fan-out resolved
    /// entry 0 to different bytes than the serial pass did (2 workers): 46 bytes
    /// vs 1933, first differing at Some(0)".
    #[test]
    fn the_fan_out_resolves_the_identical_bytes_the_serial_pass_does() {
        let (_dir, store, all) = corpus("zc-gatling");
        let oids: Vec<&[u8]> = all.iter().map(Vec::as_slice).collect();
        let (ordered, _) = topological_order(store.emit_set(&oids, true, None).unwrap());
        assert!(
            ordered.len() > 1000,
            "too few entries to spread over workers"
        );

        let snap = store.archive_snapshot().unwrap();
        let serial = store.resolve_with(&ordered, &snap, 1).unwrap();
        for workers in [2usize, 4, 8] {
            let parallel = store.resolve_with(&ordered, &snap, workers).unwrap();
            assert_eq!(
                parallel.len(),
                serial.len(),
                "the fan-out resolved {} entries against the serial pass's {} ({workers} workers)",
                parallel.len(),
                serial.len()
            );
            // Compared by hand rather than with `assert_eq!` on the slices: an
            // entry is kilobytes, and a failure that dumps two of them is a
            // failure nobody reads.
            for (i, (p, s)) in parallel.iter().zip(serial.iter()).enumerate() {
                if p.as_ref() != s.as_ref() {
                    let at = p.iter().zip(s.iter()).position(|(a, b)| a != b);
                    panic!(
                        "the fan-out resolved entry {i} to different bytes than the serial pass \
                         did ({workers} workers): {} bytes vs {}, first differing at {at:?}",
                        p.len(),
                        s.len()
                    );
                }
            }
        }

        // And the pack itself: emitted off the fan-out, walked back, closed.
        let mut pack = Vec::new();
        store.emit_ordered(&ordered, &mut pack).unwrap();
        let walk = crate::pack_walk::walk(&pack, store.hash_kind().oid_len()).unwrap();
        assert_eq!(walk.entries.len(), ordered.len());
        assert!(walk.closure().broken_offsets.is_empty());
    }

    /// 🔴 **The pack is never materialised: the whole emit set's heap is a
    /// small fraction of the pack it will write.**
    ///
    /// # What this asserts, and how completely
    ///
    /// The emit set's heap is **accounted exactly**, not sampled: an
    /// [`EmitEntry`] holds a `Vec<EmitEntry>` slot, an oid, and — for the
    /// `Owned` exception only — payload bytes. There is nothing else it can
    /// hold, so `slots + oids + owned` is the complete cost of the set, and it
    /// is compared against the bytes that set will emit. Before 2026-08-14 the
    /// `owned` term alone *was* the pack: every entry's bytes `pread` into a
    /// `Vec`, all of them live at once, because the set is built in full before
    /// [`crate::pack_walk::emit_pack`] writes a byte.
    ///
    /// Seen RED by reverting `emit_set`'s base-inside arm to
    /// `EntryBytes::Owned(self.read_extent(row.offset, row.len)?)`: "the emit
    /// set owns 5653270 bytes of payload — 100.0 % of the 5653270 bytes it will
    /// emit — which is the whole pack held in memory before the first byte goes
    /// out".
    ///
    /// # What it does NOT prove, stated plainly
    ///
    /// **It is not an RSS test, and an RSS test here would be theatre.** This
    /// began as one, asserting that `/proc/self/statm` grows by less than the
    /// pack across `emit_set`. It was **blind**: run against the broken version
    /// above, which really does allocate and fill 5 653 270 bytes, resident set
    /// size did not move by a single page — 203 366 400 before and after, and
    /// 203 182 080 → 203 182 080 in the green run. The allocator was handing
    /// back arena pages this test process already had resident. A guard that
    /// cannot tell the defect from the fix is not a weak guard, it is no guard,
    /// and it is exactly what LAW 2 says to expect of one's own new guards. It
    /// is gone; the accounting above replaces it.
    ///
    /// **It says nothing about the peak of a real clone.** This corpus is 2.8 MB
    /// against `linux.git`'s ~6.4 GB, and the 2314 MB peak this change is aimed
    /// at was measured with `perf` and `/proc`, not with a unit test.
    ///
    /// **The residual it exposes is real and is printed rather than hidden.**
    /// The `slots + oids` term is `O(objects)` and survives this change: every
    /// entry still carries its own `Vec<u8>` oid, one heap allocation each. The
    /// printed projection to 13.8 M objects is the honest size of what is left
    /// to do, and it is not small.
    #[test]
    fn an_emit_set_holds_no_payload_bytes_of_the_pack_it_will_write() {
        let (_dir, store, all) = corpus("zc-memory");
        let oids: Vec<&[u8]> = all.iter().map(Vec::as_slice).collect();

        let entries = store.emit_set(&oids, true, None).unwrap();

        let owned: u64 = entries
            .iter()
            .map(|e| e.stored.owned().map_or(0, |b| b.len() as u64))
            .sum();
        let will_emit: u64 = entries.iter().map(|e| e.stored.len()).sum();
        assert!(
            will_emit > 1_000_000,
            "the corpus emits {will_emit} bytes, too few to distinguish held from addressed"
        );
        assert_eq!(
            owned,
            0,
            "the emit set owns {owned} bytes of payload — {:.1} % of the {will_emit} bytes it will \
             emit — which is the whole pack held in memory before the first byte goes out",
            100.0 * owned as f64 / will_emit as f64
        );

        // The complete heap of the set, term by term.
        let slots = (entries.len() * std::mem::size_of::<EmitEntry>()) as u64;
        let oid_bytes: u64 = entries.iter().map(|e| e.oid.len() as u64).sum();
        let held = slots + oid_bytes + owned;
        assert!(
            held < will_emit / 8,
            "the emit set holds {held} bytes ({:.1} % of the {will_emit}-byte pack): {slots} of \
             slots, {oid_bytes} of oids, {owned} of payload",
            100.0 * held as f64 / will_emit as f64
        );
        eprintln!(
            "{} entries: {slots} B slots + {oid_bytes} B oids + {owned} B payload = {held} B held \
             for a {will_emit} B pack ({:.1} %). EmitEntry is {} B. Projected to linux.git's \
             ~13.8 M objects that residual is ~{:.2} GB — the payload copy is gone, the per-entry \
             slot and oid are NOT.",
            entries.len(),
            100.0 * held as f64 / will_emit as f64,
            std::mem::size_of::<EmitEntry>(),
            13.8e6 * (std::mem::size_of::<EmitEntry>() as f64 + 20.0) / 1e9,
        );
    }
}

/// **A `REF_DELTA` whose base is inside the same pack**, on both sides of the
/// store: ingesting one, and serving one.
///
/// # Why this shape and no other
///
/// It is not exotic and it is not hand-rolled. `git index-pack --fix-thin`
/// completes a pushed **thin** pack by *appending the base object to the pack*
/// and leaving the delta naming that base by oid — so the base lands **after**
/// its dependant, in the same pack, named the one way a pack can name something
/// that is not a position. Every repository that has been pushed to more than
/// once and not `gc`'d holds packs of this shape, which is to say: the ordinary
/// case, not a corner.
///
/// The fixture is therefore built by **git itself** (`pack-objects --thin` fed
/// to `index-pack --fix-thin`) rather than assembled here, because a hand-built
/// pack could only ever prove that this crate agrees with this crate.
///
/// Two defects met on it, one per direction:
///
/// | direction | defect | fix |
/// |---|---|---|
/// | in | `external_bases_exist` demanded every ref base be in the **store**, so a pack carrying its own base was refused | ask the pack too, and only when about to refuse |
/// | out | `emit_set` copied the stored `REF_DELTA` through, and **gitoxide cannot read one whose base is in the pack** | re-head it as an `OFS_DELTA` naming the same base by distance |
#[cfg(test)]
mod in_pack_ref_delta_tests {
    use crate::index_layout::ObjType;
    use crate::pack_walk::{DeltaBase, topological_order};
    use crate::store::tests::tmpdir;
    use crate::{GitOps, GitStore};
    use std::path::Path;
    use std::process::Command;

    fn git(dir: &Path, args: &[&str]) -> String {
        let out = Command::new("git")
            .current_dir(dir)
            .args(args)
            .env("GIT_AUTHOR_NAME", "t")
            .env("GIT_AUTHOR_EMAIL", "t@t")
            .env("GIT_COMMITTER_NAME", "t")
            .env("GIT_COMMITTER_EMAIL", "t@t")
            .env("GIT_AUTHOR_DATE", "2020-01-01T00:00:00Z")
            .env("GIT_COMMITTER_DATE", "2020-01-01T00:00:00Z")
            .output()
            .unwrap_or_else(|e| panic!("running git {args:?}: {e}"));
        assert!(
            out.status.success(),
            "git {args:?} failed: {}",
            String::from_utf8_lossy(&out.stderr)
        );
        String::from_utf8_lossy(&out.stdout).trim().to_string()
    }

    /// A pack **git wrote** that contains a `REF_DELTA` whose base is an entry
    /// of the same pack.
    ///
    /// `None` when there is no `git` on `PATH`, so the suite still runs where
    /// the oracle is absent rather than failing for the wrong reason.
    fn fix_thin_pack(scratch: &Path) -> Option<Vec<u8>> {
        if Command::new("git").arg("--version").output().is_err() {
            eprintln!("skipping: no git on PATH");
            return None;
        }
        let repo = scratch.join("src");
        std::fs::create_dir_all(&repo).unwrap();
        git(&repo, &["init", "-q", "--initial-branch=main"]);

        // Big enough, and changed little enough, that `pack-objects` really does
        // deltify the second version against the first. A two-line file would be
        // stored whole and this fixture would prove nothing — which the premise
        // assertion below refuses to let happen silently.
        let lines: Vec<String> = (0..400)
            .map(|i| format!("line {i} {}", "x".repeat(20)))
            .collect();
        std::fs::write(repo.join("f.txt"), lines.join("\n")).unwrap();
        git(&repo, &["add", "f.txt"]);
        git(&repo, &["commit", "-q", "-m", "c1"]);

        let mut changed = lines.clone();
        changed[10] = "CHANGED".to_string();
        changed.push("appended line".to_string());
        std::fs::write(repo.join("f.txt"), changed.join("\n")).unwrap();
        git(&repo, &["commit", "-q", "-a", "-m", "c2"]);

        // A THIN pack: c2's objects only, deltified against c1's, which it does
        // not carry.
        let head = git(&repo, &["rev-parse", "HEAD"]);
        let parent = git(&repo, &["rev-parse", "HEAD~1"]);
        let thin = {
            use std::io::Write as _;
            let mut child = Command::new("git")
                .current_dir(&repo)
                .args(["pack-objects", "--thin", "--revs", "--stdout"])
                .stdin(std::process::Stdio::piped())
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped())
                .spawn()
                .unwrap();
            write!(child.stdin.take().unwrap(), "{head}\n^{parent}\n").unwrap();
            let out = child.wait_with_output().unwrap();
            assert!(out.status.success(), "git pack-objects --thin failed");
            out.stdout
        };

        // `--fix-thin` appends the bases INTO the pack. The ref-deltas keep
        // naming them by oid, and now those oids ARE in the pack.
        let idx = scratch.join("fixed.idx");
        {
            use std::io::Write as _;
            let mut child = Command::new("git")
                .current_dir(&repo)
                .args([
                    "index-pack",
                    "--fix-thin",
                    "--stdin",
                    "-o",
                    idx.to_str().unwrap(),
                ])
                .stdin(std::process::Stdio::piped())
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped())
                .spawn()
                .unwrap();
            child.stdin.take().unwrap().write_all(&thin).unwrap();
            let out = child.wait_with_output().unwrap();
            assert!(
                out.status.success(),
                "git index-pack --fix-thin failed: {}",
                String::from_utf8_lossy(&out.stderr)
            );
        }
        let packdir = repo.join(".git/objects/pack");
        let pack = std::fs::read_dir(&packdir)
            .unwrap()
            .flatten()
            .map(|e| e.path())
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
            .expect("git index-pack --fix-thin wrote no pack");
        Some(std::fs::read(pack).unwrap())
    }

    /// The fixture is what it claims to be: at least one `REF_DELTA`, and every
    /// delta base resolvable **inside the pack alone**.
    ///
    /// Without this the two tests below could pass on a pack with no ref-delta
    /// in it at all, which is the vacuous green LAW 2 is about.
    fn assert_carries_its_own_ref_delta_base(pack: &[u8]) {
        let w = crate::pack_walk::walk(pack, 20).expect("git's own pack must walk");
        let refs = w
            .entries
            .iter()
            .filter(|e| e.obj_type == ObjType::RefDelta)
            .count();
        assert!(
            refs > 0,
            "the fixture has no ref-delta at all, so it cannot exercise either defect"
        );
        // Resolvable with NO external base source at all: every base a delta in
        // here names is carried in here.
        crate::resolve::resolve(
            pack,
            crate::GitHashKind::Sha1,
            0,
            &crate::resolve::NoBases,
        )
        .expect("the fixture must be self-contained, or `put_pack` is right to refuse it");
    }

    /// 🔴 **znippy ingests a pack git wrote.**
    ///
    /// `pack_walk::closure()` reports every `REF_DELTA` base on `external_refs`
    /// — it cannot do otherwise, a walk knows no oids — and
    /// `external_bases_exist` demanded each one already be in the store. A pack
    /// carrying its own base was therefore refused, which is `git index-pack
    /// --fix-thin`'s ordinary output and `gunnar.multi_pack_serve`'s failure.
    ///
    /// Asserted on applied output, not on `Ok`: every object of the pack is
    /// afterwards *in the store's index*, so a `put_pack` that returned success
    /// having stored nothing would still be red.
    ///
    /// Seen RED by restoring the old check (refuse any ref base not already in
    /// the store):
    /// *"this pack deltas against 93c4207c152fa94c3978c75b136ff5a530ca16b6,
    /// which this repository does not have — the push is refused rather than
    /// stored with a dangling base"*.
    #[test]
    fn a_pack_carrying_its_own_ref_delta_base_is_ingested_not_refused() {
        let dir = tmpdir("inpack-put");
        let Some(pack) = fix_thin_pack(&dir) else {
            return;
        };
        assert_carries_its_own_ref_delta_base(&pack);

        let store = GitStore::open(&dir.join("store"), "alice").unwrap();
        store
            .put_pack(&pack)
            .expect("a self-contained pack git itself wrote must be accepted");
        store.absorb_pending().unwrap();

        let walked = crate::pack_walk::walk(&pack, 20).unwrap();
        let stored = store.index().oids_in_order().unwrap();
        assert_eq!(
            stored.len(),
            walked.entries.len(),
            "the pack has {} entries and the store indexed {}",
            walked.entries.len(),
            stored.len()
        );
        // And specifically the base that used to be the refusal.
        let base = walked
            .entries
            .iter()
            .find_map(|e| match &e.delta_base {
                DeltaBase::Ref(oid) => Some(oid.clone()),
                _ => None,
            })
            .expect("the premise asserted there is one");
        assert!(
            store.has(&base).unwrap(),
            "the in-pack ref-delta base {} is not in the store after the push",
            hex::encode(&base)
        );
    }

    /// 🔴 **A pack this server emits is consumed by `gunnar_client`.**
    ///
    /// `gunnar_client` resolves a fetch with gitoxide, and
    /// `gix_pack::data::input::LookupRefDeltaObjectsIter` reads every
    /// `OBJ_REF_DELTA` as naming an object *the receiver already has*: it
    /// consults the local object database and the bases it has already spliced
    /// in, and never the pack it is reading. A clone's target holds nothing, so
    /// one stored ref-delta copied through is fatal for the **whole** pack.
    ///
    /// The receiver here is `gix_object::find::Never` — an object database that
    /// holds nothing, which is exactly `open_or_init_bare` at the moment a clone
    /// starts — so this is the client's own code answering, not a paraphrase of
    /// it.
    ///
    /// Three assertions, and all three are needed:
    ///
    ///  1. **gitoxide accepts it**, with an empty base source. This is the bug.
    ///  2. **stock git still accepts it.** git accepted the *broken* pack too,
    ///     so a fix that satisfied gix and broke git would otherwise ship green.
    ///  3. **the object set is exactly what was asked for**, read back out of
    ///     stock git by oid. A re-head that dropped, duplicated or corrupted an
    ///     entry passes 1 and 2 — `index-pack` files an entry under the oid it
    ///     computes from that entry's own bytes, so a wrong object is a
    ///     *different* oid and only comparing the sets can see it.
    ///
    /// Seen RED, each separately — see the commit message for the exact texts.
    #[test]
    fn an_emitted_pack_names_no_base_inside_itself_by_oid() {
        let dir = tmpdir("inpack-emit");
        let Some(pack) = fix_thin_pack(&dir) else {
            return;
        };
        assert_carries_its_own_ref_delta_base(&pack);

        let store = GitStore::open(&dir.join("store"), "alice").unwrap();
        store.put_pack(&pack).unwrap();
        store.absorb_pending().unwrap();

        let all = store.index().oids_in_order().unwrap();
        let oids: Vec<&[u8]> = all.iter().map(Vec::as_slice).collect();
        let (ordered, missing) = topological_order(store.emit_set(&oids, true, None).unwrap());
        assert!(
            missing.is_empty(),
            "an entry still names a base that is not here: {missing:?}"
        );
        let mut emitted = Vec::new();
        store.emit_ordered(&ordered, &mut emitted).unwrap();

        // ── the premise: the emitted pack really does carry a delta, or nothing
        //    below distinguishes a fix from a pack of whole objects ────────────
        let w = crate::pack_walk::walk(&emitted, 20).expect("our own walk reads what we emitted");
        let deltas = w
            .entries
            .iter()
            .filter(|e| !matches!(e.delta_base, DeltaBase::None))
            .count();
        assert!(
            deltas > 0,
            "the emitted pack has no delta at all, so it cannot show how a base is named"
        );

        // ── 1. THE BUG: no base may be named by oid ───────────────────────────
        let named_by_oid = w.closure().external_refs;
        assert!(
            named_by_oid.is_empty(),
            "a clone's pack names {} base(s) by oid — gitoxide resolves those against the \
             receiver's store and a clone's receiver is empty: {:?}",
            named_by_oid.len(),
            named_by_oid.iter().map(hex::encode).collect::<Vec<_>>()
        );

        // …and the client's own code says so, not only our reading of it.
        let out = dir.join("gix-out");
        std::fs::create_dir_all(&out).unwrap();
        let stop = std::sync::atomic::AtomicBool::new(false);
        let mut cur = std::io::Cursor::new(emitted.as_slice());
        let outcome = gix_pack::Bundle::write_to_directory(
            &mut cur,
            Some(&out),
            &mut gix_features::progress::Discard,
            &stop,
            // An object database that holds nothing — a fresh clone target.
            Some(gix_object::find::Never),
            gix_hash::Kind::Sha1,
            gix_pack::bundle::write::Options::default(),
        );
        let outcome = match outcome {
            Ok(o) => o,
            // gix 0.88 reports an `Exn`, not a `std::error::Error`.
            Err(e) => panic!(
                "gitoxide refused a pack this store emitted: {}",
                chain(&gix_error::Error::from(e))
            ),
        };
        assert_eq!(
            outcome.index.num_objects as usize,
            oids.len(),
            "gitoxide indexed {} objects for a request of {}",
            outcome.index.num_objects,
            oids.len()
        );

        // ── 2. stock git must still accept it ─────────────────────────────────
        crate::git_oracle::assert_git_accepts(
            &dir,
            "stock.git",
            &emitted,
            crate::git_oracle::Strictness::SelfContained,
        );

        // ── 3. exactly the object set that was asked for, by oid, read back out
        //       of stock git ───────────────────────────────────────────────────
        let read_back = crate::git_oracle::git_reads_back(&dir, "readback.git", &emitted).unwrap();
        assert_eq!(
            read_back.len(),
            oids.len(),
            "the pack carries {} objects, the request asked for {}",
            read_back.len(),
            oids.len()
        );
        let got: std::collections::BTreeSet<String> =
            read_back.iter().map(|(oid, _)| oid.clone()).collect();
        let want: std::collections::BTreeSet<String> = oids.iter().map(hex::encode).collect();
        assert_eq!(
            got, want,
            "the emitted pack is not the requested set — missing {:?}, unexpected {:?}",
            want.difference(&got).collect::<Vec<_>>(),
            got.difference(&want).collect::<Vec<_>>()
        );
    }

    /// **RED / GREEN perf probe** for the per-push fold cost (lane ZNIPPY-WRITE).
    ///
    /// Not part of the correctness suite (`#[ignore]`) — a manual measurement:
    /// ```text
    /// cargo test -p znippy-plugin-git --release the_incremental_push_fold_is_flat -- --ignored --nocapture
    /// ```
    /// Warms a store to `n` commits, then pushes a fixed `d`-commit delta and
    /// times `put_pack + absorb_pending` (the delta resolve is ~`3d` objects,
    /// so the number is dominated by the fold). Separately times a *forced*
    /// whole-store `refold` (graph cache removed) to isolate the fold from the
    /// resolve. The RED reading is both columns growing with `n`; the GREEN
    /// reading (after the incremental fold) is `delta_absorb_ms` flat in `n`.
    #[test]
    #[cfg(feature = "bench-kernels")]
    #[ignore = "manual perf probe; run with --features bench-kernels --ignored --nocapture"]
    fn the_incremental_push_fold_is_flat() {
        use crate::object::GitObjectKind;
        fn split(obj: &[u8]) -> (GitObjectKind, Vec<u8>) {
            let nul = obj.iter().position(|&b| b == 0).expect("canonical header");
            let kind = match std::str::from_utf8(&obj[..nul])
                .unwrap()
                .split(' ')
                .next()
                .unwrap()
            {
                "commit" => GitObjectKind::Commit,
                "tree" => GitObjectKind::Tree,
                "blob" => GitObjectKind::Blob,
                "tag" => GitObjectKind::Tag,
                other => panic!("unknown kind {other}"),
            };
            (kind, obj[nul + 1..].to_vec())
        }
        use std::time::Instant;
        let d = 8usize;
        println!("n\tcommits\tforced_refold_ms\tdelta_absorb_ms\tdelta_objs");
        for &n in &[256usize, 1024, 4096, 8192] {
            let repo = crate::bench_kernels::synth_repo(
                n + d,
                4,
                0,
                crate::object::GitHashKind::Sha256,
            );
            // synth_repo appends blob,tree,commit per commit, in order.
            let split_at = 3 * n;
            let base: Vec<(GitObjectKind, Vec<u8>)> =
                repo.objects[..split_at].iter().map(|(_, b)| split(b)).collect();
            let delta: Vec<(GitObjectKind, Vec<u8>)> =
                repo.objects[split_at..].iter().map(|(_, b)| split(b)).collect();
            let delta_objs = delta.len();
            let pack_base = crate::pack_fixture::whole_object_pack(&base);
            let pack_delta = crate::pack_fixture::whole_object_pack(&delta);

            let dir = tmpdir(&format!("fold-scale-{n}"));
            let store = GitStore::open(&dir, "alice").unwrap();
            store.put_pack(&pack_base).unwrap();
            store.absorb_pending().unwrap();

            // Isolated fold: drop the persisted graph so refold must recompute,
            // exactly what a push's token change forces.
            let _ = std::fs::remove_file(store.root().join("objects.graph"));
            let t = Instant::now();
            store.refold().unwrap();
            let forced_refold_ms = t.elapsed().as_secs_f64() * 1000.0;

            // Real incremental push.
            store.put_pack(&pack_delta).unwrap();
            let t2 = Instant::now();
            store.absorb_pending().unwrap();
            let delta_absorb_ms = t2.elapsed().as_secs_f64() * 1000.0;

            let commits = store.commit_count();
            assert_eq!(commits, n + d, "graph must hold every commit");
            println!(
                "{n}\t{commits}\t{forced_refold_ms:.2}\t{delta_absorb_ms:.2}\t{delta_objs}"
            );
        }
    }

    /// **RSS floor attribution probe** (lane ZNIPPY-WRITE). Opens `k` warm
    /// stores over the corpus pack, once with the default redb page-cache
    /// ceiling and once with a small one, and reports resident-set growth per
    /// store — the discriminator for whether the per-store floor is the redb
    /// cache. `#[ignore]`, manual:
    /// ```text
    /// cargo test -p znippy-plugin-git --release the_store_floor_rss -- --ignored --nocapture
    /// ```
    #[test]
    #[ignore = "manual RSS probe; run with --ignored --nocapture"]
    fn the_store_floor_rss() {
        use crate::arms::StoreConfig;
        fn vmrss_kb() -> u64 {
            let s = std::fs::read_to_string("/proc/self/status").unwrap_or_default();
            for l in s.lines() {
                if let Some(v) = l.strip_prefix("VmRSS:") {
                    return v.trim().trim_end_matches(" kB").trim().parse().unwrap_or(0);
                }
            }
            0
        }
        fn threads() -> usize {
            std::fs::read_dir("/proc/self/task").map(|d| d.count()).unwrap_or(0)
        }
        let pack = crate::pack_fixture::corpus().pack.clone();
        let k = 16usize;
        for (label, cache) in [("default(64MiB)", 64 << 20), ("small(1MiB)", 1 << 20)] {
            let base = vmrss_kb();
            let t0 = threads();
            let mut stores = Vec::with_capacity(k);
            for i in 0..k {
                let dir = tmpdir(&format!("floor-{label}-{i}"));
                let store = GitStore::<crate::index_layout::OneTableFourColumns>::open_with_arms(
                    &dir,
                    "alice",
                    crate::object::GitHashKind::Sha1,
                    StoreConfig::DEFAULT.with_redb_cache_bytes(cache),
                )
                .unwrap();
                store.put_pack(&pack).unwrap();
                store.absorb_pending().unwrap();
                // Warm the cache the way a served repo is.
                for r in store.refs().unwrap() {
                    if let Some(oid) = r.oid {
                        let _ = store.has(&oid);
                    }
                }
                stores.push(store);
            }
            let live = vmrss_kb();
            let per = (live.saturating_sub(base)) as f64 / k as f64 / 1024.0;
            println!(
                "FLOOR cache={label} k={k} rss_growth={:.1}MB per_store={:.2}MB threads_delta={}",
                (live.saturating_sub(base)) as f64 / 1024.0,
                per,
                threads().saturating_sub(t0)
            );
            drop(stores);
        }
    }

    /// gix errors nest their real cause; the top line alone names the operation
    /// and nothing about why. The same walk `gunnar-client::error::chain` does,
    /// and here for the same reason: without it the red above reads *"Failed to
    /// write pack"* and diagnoses nothing.
    fn chain(err: &(dyn std::error::Error + 'static)) -> String {
        let mut out = err.to_string();
        let mut cursor = err.source();
        for _ in 0..16 {
            let Some(next) = cursor else { break };
            let text = next.to_string();
            if !out.ends_with(&text) {
                out.push_str(": ");
                out.push_str(&text);
            }
            cursor = next.source();
        }
        out
    }
}