znippy-plugin-git 0.1.1

//! **`GitOps` — ours, typed, and base znippy never sees it.**
//!
//! §13.2: `ArchiveTypePlugin` is base znippy's contract and
//! [`crate::NativeGitPlugin`] implements it — base stores, the plugin annotates,
//! exactly as maven, python, rust-toolchain and media already work. `GitOps` is
//! the *other* trait: the one gunnar calls, with git's own vocabulary in it.
//!
//! §13.3: `ArchiveTypePlugin::run_command(cmd, args)` is **CLI-only string
//! dispatch** and delegates here. A server's hot path does not go through
//! stringly-typed dispatch, so nothing below takes a `&str` command or an
//! `args: &[String]`.
//!
//! ```text
//!   gunnar ──typed──► GitOps  (this file, twelve methods)
//!                       │
//!      ┌────────────────┼──────────────────┬─────────────┐
//!      ▼                ▼                  ▼             ▼
//!   ArchiveWrite    ObjectIndex          RefLog         Gc
//!   SafeWriter      ObjectReadStack      push log       NewGeneration
//!   (blob fsync,    (stree → Arrow       (Arrow IPC     (link, compact,
//!    journal row)    → redb tail)         frame = tx)    verify, rename)
//! ```
//!
//! Every arm under that diagram already existed and was already measured. This
//! file adds no storage, no index and no durability mechanism of its own — if a
//! method here looks like it is inventing one, that is a bug.
//!
//! # The one thing that is NOT here
//!
//! `run_command` still lives on the plugin in [`crate::lib`], and it now
//! delegates. Nothing git-shaped went into base znippy for any of this.

use std::collections::{BTreeMap, HashMap};
use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, RwLock};

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

use crate::arms::StoreConfig;
use crate::exploded::ExplodedStats;
use crate::exploded_arrow::ExplodedArchive;
use crate::gc::{Gc, GcReport};
use crate::graph::{assign_generations, CommitNode};
use crate::index_layout::{ObjectIndex, OneTableFourColumns};
use crate::indexer::{IndexJob, ObjectAbsorb, PushPath};
use crate::object::{GitHashKind, GitObjectKind};
use crate::pack_walk::PackWalk;
use crate::reach::ReachEntry;
use crate::read_stack::{ObjectReadStack, RebuildTriggers};
use crate::refs::{RefLog, RefUpdate};
use crate::resolve::BaseSource;

// ── the contract, now owned by `git-storage-trait` ────────────────────────────
//
// `GitOps` (the neutral ELEVEN) and its value types live in the
// `git-storage-trait` crate at this repo's root, so a gix backend
// (edda/storage-git-gix) can implement the same contract without linking Arrow
// or OpenZL. The twelfth method, `seal()`, returns `znippy_common::
// ReservedSection` (an Arrow type a gix backend has no analog for), so it is an
// **inherent** method on [`GitStore`] / [`SelectedStore`], not part of the
// trait. The re-exports below keep every existing path
// (`znippy_plugin_git::git_ops::{GitOps, Oid, …}`) alive.
pub use git_storage_trait::{Extent, GitOps, Oid, RefRow, Stored, TxId};


// ── the batch shape, which is measured and must not be papered over ───────────

/// Which path a lookup of `n` oids takes.
///
/// The number this dispatch was written for is `index_layout_bench`'s: on the bare
/// Arrow arms, `lookup_batch` with a **single** element costs 818 ns against
/// `lookup`'s 482 ns — 1.7x worse — because the batch path allocates a result
/// vector, a key vector and a second pass over them to serve one oid.
///
/// **RE-MEASURED against the shipping read stack, and the magnitude does not
/// carry over.** oden 2026-08-07, release, 1-min loadavg 4.94 (another tenant's
/// work — see LAW: never measure on a busy box, and read these as a ratio rather
/// than as absolutes), `ObjectReadStack<OneTableFourColumns>` over a real
/// repository's 2687-object pack, 20 000 iterations
/// ([`crate::store::tests::the_batch_of_one_really_is_slower_than_the_serial_path`]):
///
/// | call | per oid |
/// |---|---:|
/// | `lookup` | **628 ns** |
/// | `lookup_batch` with 1 oid | **639 ns** — 1.02x worse |
/// | `lookup_batch` with 100 oids | **161 ns** — 3.9x better |
///
/// So: the *direction* holds and the dispatch is right — a batch of one is never
/// faster — but on this stack at this size it costs 2%, not 70%. The real finding
/// is the third row: the batch path is worth **3.9x** at
/// [`BATCH_SATURATES_AT`], which is a much stronger reason to keep the two paths
/// apart than the batch-of-one penalty ever was. Both numbers are stated because
/// quoting only the 1.7x would be quoting a figure this stack does not reproduce.
///
/// The dispatch is a named function with its own guard precisely so it cannot
/// quietly become `lookup_batch` for every n.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LookupPath {
    /// One oid: `ObjectIndex::lookup`.
    Serial,
    /// Two or more: `ObjectIndex::lookup_batch`, one pass.
    Batch,
}

/// Oids at which the batch path stops getting faster. Reported, not enforced:
/// splitting a bigger batch would cost a pass and buy nothing.
pub const BATCH_SATURATES_AT: usize = 100;

/// The dispatch, in one place.
pub fn lookup_path(n: usize) -> LookupPath {
    if n == 1 {
        LookupPath::Serial
    } else {
        LookupPath::Batch
    }
}

// ── the store ─────────────────────────────────────────────────────────────────

/// A pack whose bytes are durable but whose objects the index has not absorbed.
///
/// §13.12's `indexed` bit, in the form this store needs it: membership in
/// [`PackState::pending`] *is* the bit being clear. A read that arrives while it
/// is non-empty falls back to absorbing it — slower, never wrong.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct PendingPack {
    pack_id: u64,
    offset: u64,
    len: u64,
}

impl PendingPack {
    fn from_job(j: IndexJob) -> Self {
        Self {
            pack_id: j.pack_id,
            offset: j.offset,
            len: j.len,
        }
    }
}

/// **§13.12's `indexed` bit — one bit per pack, indexed by pack ordinal.**
///
/// A plain bitset, and the two rejected alternatives say why:
///
/// * **not a hash set.** Pack ordinals are dense and small: they are assigned
///   `0..N` within **one repository** (§13.14 scopes a negotiation, and therefore
///   this store, to one repository — account-wide they would be sparse and the
///   array would grow with the account's total packs). Dense means the bit is one
///   shift and one AND with no hashing, and 10 000 packs is **1250 bytes** of
///   `words`, which lives in L1.
/// * **not a bloom filter.** A bloom's false positive would be *actively wrong*
///   here: it would report a pack absorbed when it is not, the read path would
///   stop falling back, and negotiation would be told `absent` about an object
///   whose bytes are durable — which makes a client withhold objects and lose
///   data. The bitset has no such mode. `1` means absorbed, `0` means fall back,
///   and both are exact.
///
/// **The bit is never stored.** It is derived on open by diffing two facts that
/// are already durable — the journal's extents against the index's rows — in
/// [`Absorber::adopt_journal`]. There is nothing for a derived bit to drift from.
///
/// **Why the extents sit beside it.** A `0` bit is only actionable with the
/// extent to absorb, so `extent[i]` carries it; `None` means "this store has not
/// been told about ordinal `i`", which is not the same as "not absorbed" and must
/// not be counted as work.
///
/// **Why `absorbed` is a separate fact from `extent`.** The ack path does two
/// things in this order: record the pack pending, then hand its extent to the
/// account's channel. The indexer's worker is a different thread and can reach
/// `absorb` *before* the first of those returns. With only a pending list, that
/// race ends in a bit that is set after the rows are already in and is never
/// cleared, so every later read pays a pointless fallback for the life of the
/// process. The bit is what the two writers agree on: [`PackState::note`] will
/// not queue a pack that is already done, and `absorb` will not do the work
/// twice.
///
/// Touched only under short critical sections — never across an absorb, which is
/// what the separate [`Absorber::gate`] is for. **The ack path must never block
/// behind index work.**
#[derive(Default)]
struct PackState {
    /// One bit per pack ordinal, 64 to a word. `1` = its rows are in the index.
    words: Vec<u64>,
    /// The extent of each known ordinal, so a `0` bit can be acted on.
    extent: Vec<Option<Extent>>,
    /// Known ordinals whose bit is `0` — what [`GitStore::unindexed_packs`]
    /// reports, kept as a count so the read path's check is a field load rather
    /// than a scan.
    unabsorbed: usize,
}

impl PackState {
    /// The bit.
    fn is_absorbed(&self, pack_id: u64) -> bool {
        let i = pack_id as usize;
        self.words
            .get(i / 64)
            .is_some_and(|w| w >> (i % 64) & 1 == 1)
    }

    /// Record a durable pack whose rows are not in yet. The bit **going clear**.
    ///
    /// **Idempotent, and the extent slot is the single thing that makes it so.**
    /// A second `note` for an ordinal that already has one — the ack path
    /// arriving after the drain has already absorbed the pack, or a re-queue on
    /// open racing the same pack's push — finds the slot filled and counts no new
    /// work. [`PackState::mark_absorbed`] fills that same slot, so a pack that is
    /// already done cannot be queued behind its own completion, and there is no
    /// second condition here that could disagree with the first.
    fn note(&mut self, pack_id: u64, extent: Extent) {
        let i = pack_id as usize;
        self.grow_to(i);
        if self.extent[i].is_none() {
            self.extent[i] = Some(extent);
            self.unabsorbed += 1;
        }
    }

    /// Room for ordinal `i` in both arrays. Dense ordinals are what makes this a
    /// `resize` and not an insert.
    fn grow_to(&mut self, i: usize) {
        if self.words.len() <= i / 64 {
            self.words.resize(i / 64 + 1, 0);
        }
        if self.extent.len() <= i {
            self.extent.resize(i + 1, None);
        }
    }

    /// The bit **going up**: this pack's rows are in the index.
    fn mark_absorbed(&mut self, pack_id: u64, extent: Extent) {
        let i = pack_id as usize;
        self.grow_to(i);
        if self.words[i / 64] >> (i % 64) & 1 == 0 {
            self.words[i / 64] |= 1 << (i % 64);
            if self.extent[i].is_some() {
                self.unabsorbed -= 1;
            }
        }
        self.extent[i] = Some(extent);
    }

    /// Every known pack whose bit is clear, in ordinal order.
    fn pending(&self) -> Vec<PendingPack> {
        self.extent
            .iter()
            .enumerate()
            .filter(|(i, e)| e.is_some() && !self.is_absorbed(*i as u64))
            .map(|(i, e)| {
                let (offset, len) = e.expect("filtered on Some");
                PendingPack {
                    pack_id: i as u64,
                    offset,
                    len,
                }
            })
            .collect()
    }
}

/// The derived tables: the commit graph, the tree payloads the reachability
/// bitmaps are built from, and this store's own ordinal space.
///
/// **The ordinal space is this struct's, not the projection's.** An
/// [`crate::index_layout::IndexRow::ordinal`] is a row address inside one
/// projection generation and is re-derived by every rebuild, so a bitmap built
/// over projection ordinals would silently address different objects after any
/// rebuild. The bitmaps here are over `ordinal`, which is rebuilt in the same
/// fold that rebuilds them — the two cannot drift because one call produces both.
/// **Nothing in here is a source of truth.** Every field is folded from the
/// `objects` table and §14's exploded table by [`Absorber::refold`], and both of
/// those are on disk — which is what makes a clean reopen come back with the
/// same graph it shut down with. Before the exploded table existed, `graph` and
/// `trees` were *accumulated* here as packs were absorbed and stored nowhere, so
/// a clean shutdown lost them silently: MEASURED on oden 2026-08-08, a reopened
/// store over a fully absorbed 2687-object pack had all 2687 rows and **0 of 551
/// commits**, with `reachable()` returning a commit alone instead of its closure
/// and `gc` seeing an empty live set.
#[derive(Default)]
struct Derived {
    /// Parents-before-children, generations assigned. Folded from the exploded
    /// table's commit payloads.
    graph: Vec<CommitNode>,
    /// Tree payloads by oid hex — what [`crate::reach::ObjectFacts`] wants.
    /// Folded from the exploded table's tree payloads.
    trees: HashMap<String, Vec<u8>>,
    /// oid hex → ordinal, and its inverse.
    ordinal: HashMap<String, u32>,
    oids: Vec<String>,
    /// The same inverse in **raw bytes**: ordinal `o` is
    /// `oids_raw[o * oid_len .. (o + 1) * oid_len]`, one flat allocation for the
    /// whole store.
    ///
    /// # Why both, rather than deriving one from the other per request
    ///
    /// Because deriving it per request is what this field replaces, and it was
    /// the most expensive thing on a fetch that no stage was named after.
    /// `reachable_oids` answered in oid **hex** and every serving caller
    /// immediately `hex::decode`d it straight back to the bytes it came from:
    /// one `String` plus one `Vec<u8>` per object in the answer, and the answer
    /// for the `have` side is the closure of what the client already holds —
    /// very nearly the whole repository. MEASURED on oden 2026-08-15, the
    /// `nornir` mirror (12 303 objects), a 69-object negotiated fetch: **12 112
    /// oids** through that round trip, **twice** per request (`select`'s voucher
    /// and `emit_set`'s thin bases), plus a third `to_vec` per object to key
    /// them by offset.
    ///
    /// The hex form stays because the ordinal map, the commit graph and the
    /// tree payloads are keyed by it and those are not this change's subject;
    /// the raw form is what every *serving* answer is now built out of. Both are
    /// folded from one sorted vector of raw oids, so they cannot fall out of
    /// step: see [`Absorber::refold`].
    oids_raw: Vec<u8>,
    /// The bitmaps, or empty if nothing has needed them since the last fold.
    ///
    /// **Behind an `Arc` since 2026-08-14, and that is a measurement and not a
    /// tidy-up.** [`Absorber::reach_bitmaps`] used to hand out `d.reach.clone()`
    /// — a deep copy of every commit's `RoaringBitmap` plus its oid `String` —
    /// and `crate::serve::GitServe::select` calls
    /// [`GitStore::reachable_oids`] **twice** per request (once for `want`, once
    /// to close over `have`). MEASURED on oden with `stage-probe`, the
    /// znippy-served `nornir` mirror (12 455 objects), a 4-object negotiated
    /// fetch: the `select.bitmap` stage was **75.2 % of the request's CPU** and
    /// **645 876 allocations**; at four requests sharing one warm store it was
    /// still 62 762 allocations *per request*. None of that copying is work the
    /// answer depends on — the table is immutable between folds — so the `Arc`
    /// removes it rather than dividing it across cores.
    reach: Arc<Vec<ReachEntry>>,
    /// Every commit's oid as **raw bytes**, the form
    /// [`crate::serve::GitServe::select`] tests a `want` against.
    ///
    /// `None` means a graph row's oid did not parse as hex, which `select`
    /// answers by declining the whole request. That check is why this is
    /// `Option` rather than a plain set: it used to run **per request** —
    /// `graph_snapshot()` cloned the whole `Vec<CommitNode>` and then
    /// `hex::decode`d every commit into a fresh `Vec<u8>` — and the answer
    /// cannot change between folds, so it is folded once here and borrowed
    /// after. The refusal is preserved exactly; only the arithmetic moved.
    commit_raw: Option<Arc<std::collections::HashSet<Vec<u8>>>>,
    /// Verbatim packs, so any absorbed object's content can be re-derived from
    /// the truth on demand. `(archive offset, len)`.
    packs: Vec<(u64, u64)>,
}

/// **The object-level index, and the one thing that writes into it.**
///
/// Its own allocation rather than a field group inside [`GitStore`], and §13.9-12
/// is the only reason: the drain runs on the account indexer's worker thread, so
/// whatever it absorbs into cannot be reachable solely through a `&GitStore`.
/// Splitting it out is what lets the background worker and a falling-back read
/// call **the same** absorb instead of two copies of it (LAW 5) — the alternative
/// was a second implementation on the worker side, which is precisely the
/// twinning that law forbids.
///
/// It is a **leaf**: nothing in here points back at the store, so the worker's
/// `Arc` closes no cycle and the store still drops (and joins its worker)
/// normally.
/// Entries below which [`GitStore::resolve_emit_payloads`] does **not** fan out.
///
/// A `std::thread::scope` spawn is ~10–20 µs a worker and the per-entry work it
/// would divide is a bounds check plus a varint; a 100-object fetch would pay
/// four spawns to save four microseconds of parsing. The threshold is what keeps
/// `crate::serve`'s "one core's worth per transfer" true for every request that
/// is not a bulk clone — and a bulk clone is the only request the copy this
/// replaces was ever measured on.
const FAN_OUT_AT: usize = 4096;

/// Commits that get a reachability bitmap on the **live** selection path.
///
/// # It was `usize::MAX`, and that was the single most expensive line in a fetch
///
/// The comment that stood here said a sampled table *"is for the sealed archive
/// where the cost is disk"* and that every commit must be bitmapped on the live
/// path. That was not a preference, it was forced:
/// [`GitStore::reachable_oids`] had no walk, so a `want` with no bitmap could
/// only contribute itself — a silent under-send — and `crate::serve`'s `select`
/// therefore had to refuse unless *"in the graph"* and *"has a bitmap"* were the
/// same fact.
///
/// MEASURED on oden 2026-08-14, one 4-object negotiated fetch out of the
/// znippy-served `nornir` mirror (1 581 commits, 12 455 objects): the
/// `select.bitmap` stage was **75 % of the request's CPU**, and **530 218 of its
/// 575 349 allocations were this build** — which `refold` throws away again on
/// the next push. The gix arm answered the identical fetch in 6.8 ms because it
/// *reads* a `.bitmap` instead of building one.
///
/// [`crate::reach::accumulate`] is the walk that makes the cap legal, and 512 is
/// the number [`crate::reach::ReachPolicy::default`] already uses for the sealed
/// archive — the same figure git's own bitmap selection is of the order of, and
/// deliberately not a second, different constant to keep in step.
const LIVE_REACH_COMMITS: usize = 512;

/// [`LIVE_REACH_COMMITS`], or `ZNIPPY_GIT_REACH_COMMITS`.
///
/// **An operator knob, not an arm.** It selects no implementation and changes no
/// byte of any answer: the walk covers whatever the table does not, so the cap
/// trades build cost against walk cost and nothing else. That property is the
/// whole subject of
/// `the_bounded_walk_answers_exactly_what_a_full_bitmap_table_answers`, which
/// drives this very knob to its two extremes over one store.
///
/// Read **once per process** through [`crate::arms::read_env`], so it is
/// counted with every other environment read this crate makes, and so a
/// reachability request does not cost a `getenv` — until 2026-08-21 it did,
/// once per `reach_bitmaps`/`reachable_oids` call, uncounted.
fn live_reach_policy() -> crate::reach::ReachPolicy {
    static MAX: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
    let max_commits = *MAX.get_or_init(|| {
        crate::arms::read_env(crate::arms::ENV_REACH_COMMITS)
            .and_then(|v| v.parse::<usize>().ok())
            .unwrap_or(LIVE_REACH_COMMITS)
    });
    crate::reach::ReachPolicy { max_commits }
}

/// Ceiling on phase-1 workers **per request**.
///
/// The serving tier admits ~32 concurrent transfers, and the fan-out primitive
/// has no reentrancy detection, so an uncapped `gatling_for_each` here is
/// 32 × ncores threads on a 128-core box. Four is a bound that leaves the
/// admission gate in charge of the machine.
const MAX_RESOLVE_WORKERS: usize = 4;

/// Workers for phase 1: [`MAX_RESOLVE_WORKERS`], or `ZNIPPY_GIT_EMIT_WORKERS`.
///
/// **An operator knob, not an arm.** It selects no implementation and changes no
/// byte of the emitted pack — a pack emitted at one worker and at sixteen is
/// byte-identical, because phase 1 only decides *where each entry's bytes are*
/// and phase 2 writes them in one serial order either way. `0` is read as "as
/// many as this box has cores", which is `gatling_for_each`'s own convention.
///
/// Read **once per process** through [`crate::arms::read_env`] — see
/// [`live_reach_policy`] for why; this one was a `getenv` per emitted request
/// until 2026-08-21.
fn emit_workers() -> usize {
    static WORKERS: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
    *WORKERS.get_or_init(|| {
        crate::arms::read_env(crate::arms::ENV_EMIT_WORKERS)
            .and_then(|v| v.parse::<usize>().ok())
            .unwrap_or(MAX_RESOLVE_WORKERS)
    })
}

pub(crate) struct Absorber<S: ObjectIndex = OneTableFourColumns> {
    blobs: PathBuf,
    /// Read handle on the verbatim packs, for `pread` of an extent.
    ///
    /// The **fallback**, since 2026-08-14, not the ordinary path: `map` below
    /// answers every extent the mapping covers, and this reads the ones past its
    /// end (a push that landed after the snapshot was taken).
    reader: File,
    /// The same file, mapped read-only. See [`crate::archive_map`] for why an
    /// append-only file `gc` never rewrites is safe to map, and for the ~20 % of
    /// serve cycles the `pread`-per-object shape was spending on copies.
    map: crate::archive_map::ArchiveMap,
    hash: GitHashKind,
    /// The `objects` table: stree → Arrow → redb tail.
    ///
    /// `S` is the payload layout the projection is built in
    /// ([`crate::arms::IndexArm`]) and it is a **type parameter, not a field**:
    /// this is the hot path, every lookup goes through it, and the whole reason
    /// the arms exist is to measure the differences between them. A `dyn` here
    /// would put a virtual call on every single oid.
    objects: ObjectReadStack<S>,
    /// **§14's exploded objects table** — oid → resolved type and content, built
    /// eagerly by this absorb and by nothing else.
    ///
    /// Beside the `objects` table rather than inside it because the two have
    /// different lifetimes: `objects` is the index and losing it loses the
    /// repository's addressing, while this one is derived, verifiable against the
    /// verbatim bytes and deletable at any time. They are separate files so that
    /// "droppable" is a `rm`, not a schema migration.
    ///
    /// SWAPPED 2026-08-13 from the redb `ExplodedTable` to **one Arrow IPC
    /// table**, payload in the row. redb held 16.8 GB of kernel payload in a
    /// 204 GB file — 12× — and serialised the gatling fan-out behind its one
    /// writer at 96.8% of a single core. See [`crate::exploded_arrow`].
    exploded: ExplodedArchive,
    derived: RwLock<Derived>,
    packs: Mutex<PackState>,
    /// Serialises **absorption**, which is long. Held across a whole pack.
    ///
    /// Two absorbs of one pack are duplicated *work* and nothing worse: every
    /// writer under it is keyed by oid — [`ObjectReadStack::append`] swallows an
    /// identical second row, the exploded table overwrites its own key, and the
    /// commit graph is folded from that table rather than pushed at, so it cannot
    /// double. (It could before §14's table existed, when the graph was a `Vec`
    /// accumulated per absorb; that is why this paragraph used to say the gate
    /// was load-bearing for correctness and now does not.) The gate is still what
    /// makes the ordering below hold: **the rows go in, and only then is the bit
    /// cleared**, so a read either finds the bit still clear (and waits here) or
    /// finds the rows. Neither order can produce "absent".
    gate: Mutex<()>,
}

/// One repository's git store.
///
/// | on disk | what it is | who owns the format |
/// |---|---|---|
/// | `<root>/objects.pack` | the verbatim pack bytes, appended | [`SafeWriter`] |
/// | `<root>/objects.pack.journal` | the durable extent rows | [`SafeWriter`] |
/// | `<root>/objects.tail` | the un-sealed index tail | [`ObjectReadStack`] (redb) |
/// | `<root>/objects.exploded` | §14's resolved content, **droppable** | [`ExplodedArchive`] (one Arrow IPC table) |
/// | `<root>/refs.log` | the ref transaction log | [`RefLog`] (Arrow IPC frames) |
/// | `<root>/<name>.znippy` | the sealed archive GC compacts | base znippy |
///
/// Nothing in that table is new. The store is the composition, not a fifth
/// format.
/// # Which implementation of each trait, and how a caller picks
///
/// Three of the four boxes in the diagram at the top of this file have more than
/// one implementation. [`StoreConfig`] is the choice, [`crate::arms`] documents
/// what each arm promises, and the dispatch is split deliberately:
///
/// | trait | selected by | dispatch | cost |
/// |---|---|---|---|
/// | [`ObjectIndex`] | the type parameter `S` | **static** | none — the hot path is monomorphised |
/// | [`ArchiveWrite`](crate::archive_write::ArchiveWrite) | [`StoreConfig::writer`] | `dyn`, inside [`PushPath`] as it already was | one virtual call **per pack append**, not per object |
/// | [`Gc`] | [`StoreConfig::gc`] | `dyn` | one virtual call per `gc()` run |
///
/// `S` defaults to [`OneTableFourColumns`], so `GitStore` still names exactly
/// what it named before and every existing caller compiles unchanged.
pub struct GitStore<S: ObjectIndex = OneTableFourColumns> {
    root: PathBuf,
    /// The verbatim pack bytes. The **truth** (§14).
    blobs: PathBuf,
    /// The sealed znippy archive `gc` compacts. Only GC and `seal` touch it.
    archive: PathBuf,
    /// The selected [`ArchiveWrite`](crate::archive_write::ArchiveWrite) arm
    /// plus the per-account index channel. The ack path.
    push: PushPath,
    account: String,
    hash: GitHashKind,
    /// The object-level index and the absorb behind it — **shared with the
    /// account indexer's worker**, which is how a push ends in object rows
    /// without a read having to ask for them.
    absorber: Arc<Absorber<S>>,
    refs: RefLog,
    /// Serialises the ref log's read-compare-append. A CAS that read the
    /// namespace, then appended, without holding this could be overtaken between
    /// the two and would overwrite the update it did not see.
    ref_gate: Mutex<()>,
    /// The selected [`Gc`] arm. Built once, here, and run by
    /// [`GitOps::gc`] — a `Box<dyn>` because a GC is a whole-archive
    /// compaction that happens on a maintenance timer, so one virtual call
    /// against it is unmeasurable and a third type parameter on every signature
    /// in the crate is not free to read.
    gc: Box<dyn Gc + Send + Sync>,
    /// What this store was built with, for a bench row or a config dump. A
    /// description of the arms already constructed — **nothing consults it on a
    /// serving path**.
    arms: StoreConfig,
}

impl GitStore<OneTableFourColumns> {
    /// Open (creating) the store for one repository under `root`.
    ///
    /// **The default arms, and it reads no environment to get them.** An
    /// operator who exported `ZNIPPY_GIT_WRITER=fast` does not change what this
    /// builds — a caller that never asked for a selector must not have its
    /// durability contract moved out from under it. Use
    /// [`open_from_env`] when the selection is wanted.
    ///
    /// ⚠ One variable *is* read here, and it is not an arm:
    /// `ZNIPPY_GIT_REDB_CACHE_BYTES` (see
    /// [`crate::arms::redb_cache_bytes`]). It selects no implementation,
    /// changes no durability contract and leaves the bytes on disk identical —
    /// it caps redb's page cache, whose own default is 1 GiB *per database*, of
    /// which this store opens two. The sentence above is about arms, and a
    /// memory ceiling is not one.
    pub fn open(root: &Path, account: &str) -> Result<Self> {
        Self::open_with(root, account, GitHashKind::Sha1)
    }

    /// Same, with an explicit oid width. sha256 repositories exist and the store
    /// does not get to assume otherwise; every oid it stores is
    /// [`GitHashKind::oid_len`] bytes wide and the tail refuses a mixed batch.
    ///
    /// # What a reopen recovers, and why nothing had to be stored for it
    ///
    /// A store opened over an `objects.pack` some earlier process was still
    /// indexing has to end up in the same state that process would have reached.
    /// §13.12's `indexed` bit is what says so, and it is **derived here rather
    /// than persisted**: a pack is unabsorbed *iff* its extent is in the journal
    /// and its rows are not in the index, and both of those are already durable
    /// (see [`Absorber::adopt_journal`]). The packs that come back clear are put
    /// straight onto the account's channel, so the drain finishes the interrupted
    /// work with no read having to ask — and until it does, every read falls back
    /// exactly as it would have before the crash. **Slower, never wrong**, across
    /// a restart as well as within one.
    ///
    /// The bits are set **before** the channel is fed, which is the same ordering
    /// the ack path uses for the same reason: a read that arrives in between finds
    /// the bit clear and waits, never an index that does not mention the pack.
    ///
    /// # What a reopen recovers of the DERIVED tables, and how that was fixed
    ///
    /// [`Derived::graph`] and [`Derived::trees`] come back **in full**, for every
    /// pack, whether or not anything is re-queued. They are folded from §14's
    /// exploded table ([`Absorber::refold`]), which is on disk, so a clean
    /// shutdown carries them across.
    ///
    /// This was a live correctness hole until 2026-08-08 and it is recorded
    /// rather than quietly repaired. The graph was *accumulated in RAM* as packs
    /// were absorbed and stored nowhere; a store killed mid-absorb got it back
    /// because the interrupted pack was re-queued and re-resolved, but a store
    /// that shut down **cleanly** re-queued nothing and came back with an empty
    /// graph. MEASURED on oden 2026-08-08: 2687 rows and **0 of 551 commits**,
    /// `reachable` on a commit returning that commit alone instead of its
    /// closure — a *quiet* wrong answer — and `gc` seeing an empty live set. The
    /// fix is the table, because the payloads the graph is folded from had no
    /// home; [`tests::a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable`]
    /// is what proves it, over a real process that exits normally.
    ///
    /// The exploded table is still **droppable**: if its row count is short of
    /// the `objects` table's, [`Absorber::adopt_journal`] declares no pack
    /// absorbed and every one of them is re-queued and re-exploded. Absent means
    /// rebuild, never wrong (§13.12, applied unchanged to a derived table).
    ///
    /// **Also the default arms, and also no environment read** — see
    /// [`open`](GitStore::open).
    pub fn open_with(root: &Path, account: &str, hash: GitHashKind) -> Result<Self> {
        // The default ARMS, and the environment's CEILING: the one variable
        // this constructor reads, because an operator capping a server's
        // footprint must not have to pick a different constructor to do it.
        // See [`crate::arms::redb_cache_bytes`].
        Self::open_with_arms(
            root,
            account,
            hash,
            StoreConfig::DEFAULT.with_redb_cache_bytes(crate::arms::redb_cache_bytes()?),
        )
    }
}

/// Where a stored entry's delta base sits, relative to the request being served.
///
/// The `Outside` arm carries the base in **both** namings because the two delta
/// forms know only one each — an `OFS_DELTA` its archive offset, a `REF_DELTA`
/// its oid — and [`GitStore::emit_set`]'s thin arm has to ask "does the client
/// hold this" in whichever one the entry speaks.
enum BaseOf {
    /// Not a delta at all. There is nothing to place it after and nothing to
    /// re-head; the stored entry is copied as it stands.
    Whole,
    /// A delta whose base **is** in the request, at this archive offset.
    ///
    /// The offset is carried rather than recomputed because a `REF_DELTA` does
    /// not know it — it names its base by oid, and its own `delta_base` column
    /// is `0` by construction — and the emitter needs it twice: as the
    /// `EmitEntry::delta_base` that [`crate::pack_walk::topological_order`]
    /// orders on, and as the base an in-pack ref-delta is re-headed to point at.
    Inside { at: u64 },
    /// A delta whose base the request does not carry.
    Outside {
        /// The base's archive offset, `0` when the entry named an oid instead.
        offset: u64,
        /// The base's oid, `None` when the entry named an offset instead — or
        /// when the entry is too short to read one out of.
        oid: Option<Vec<u8>>,
    },
}

/// The bases a **thin** pack may name without carrying them: what the receiver
/// already held before this transfer, keyed both ways.
///
/// Built by [`GitStore::client_bases`], which is where the argument for trusting
/// it is written down.
struct ExternalBases {
    /// The held object's oid, keyed by its archive offset — the coordinate an
    /// `OFS_DELTA` names its base in, and the only reason this map exists.
    by_offset: HashMap<u64, OidKey>,
    /// The same objects keyed the way a `REF_DELTA` names them.
    held: std::collections::HashSet<OidKey>,
}

/// **A raw oid as a fixed-size, `Copy` hash key — 20 bytes for SHA-1, 32 for
/// SHA-256, and no heap allocation either way.**
///
/// The two maps in [`ExternalBases`] are repository-sized: the client's `have`
/// closure on an incremental fetch is very nearly every object in the store.
/// Keying them by `Vec<u8>` is one heap allocation per entry per request —
/// 12 112 of them for a 69-object fetch, measured on oden 2026-08-15 — for a
/// question (*does the client hold this base?*) that never looks at more than
/// `oid_len` bytes.
///
/// Padded to the widest hash rather than parameterised by width: a store holds
/// exactly one hash kind, so every key in one set has the same real length and
/// two different oids can never collide through the padding. The length is
/// carried anyway, because a key that silently compared equal across widths is
/// the kind of wrong that only shows up in a mixed-hash mirror.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct OidKey {
    bytes: [u8; 32],
    len: u8,
}

impl OidKey {
    /// The key for `oid`, or a refusal for one wider than any git hash.
    fn new(oid: &[u8]) -> Result<Self> {
        let len = oid.len();
        if len == 0 || len > 32 {
            bail!("{len} is not the width of any git object id (1..=32 bytes)");
        }
        let mut bytes = [0u8; 32];
        bytes[..len].copy_from_slice(oid);
        Ok(OidKey {
            bytes,
            len: len as u8,
        })
    }

    /// The oid back, at its real width.
    fn as_slice(&self) -> &[u8] {
        &self.bytes[..self.len as usize]
    }
}

impl<S: ObjectIndex + 'static> GitStore<S> {
    /// **The one constructor**, with every arm named.
    ///
    /// `S` picks the [`ObjectIndex`] layout at compile time; `arms` picks the
    /// [`ArchiveWrite`](crate::archive_write::ArchiveWrite) and [`Gc`]
    /// implementations at run time. [`open`](GitStore::open) and
    /// [`open_with`](GitStore::open_with) are this function with
    /// [`StoreConfig::DEFAULT`], which is why nothing about them moved.
    ///
    /// `arms.index` is **not** consulted here — `S` is the index arm, and a
    /// value that disagreed with the type would be a second source of truth.
    /// [`open_from_env`] is the one place the two are tied together, and it does
    /// it by choosing `S` from the value.
    ///
    /// ```no_run
    /// use znippy_plugin_git::arms::{GcArm, StoreConfig, WriterArm};
    /// use znippy_plugin_git::index_layout::PackedPayload;
    /// use znippy_plugin_git::{GitHashKind, GitStore};
    ///
    /// // A rebuildable import mirror: the ceiling writer, the packed payload
    /// // layout, and an in-place compaction.
    /// let arms = StoreConfig::DEFAULT
    ///     .with_writer(WriterArm::Fast)
    ///     .with_gc(GcArm::CompactInPlace);
    /// let store = GitStore::<PackedPayload>::open_with_arms(
    ///     std::path::Path::new("/srv/repo"),
    ///     "rickard",
    ///     GitHashKind::Sha1,
    ///     arms,
    /// )?;
    /// # Ok::<(), anyhow::Error>(())
    /// ```
    pub fn open_with_arms(
        root: &Path,
        account: &str,
        hash: GitHashKind,
        arms: StoreConfig,
    ) -> Result<Self> {
        std::fs::create_dir_all(root)
            .with_context(|| format!("creating the store root {}", root.display()))?;
        let blobs = root.join("objects.pack");
        let writer = arms.writer.create(&blobs)?;
        // `None` for an arm that keeps no durable ack log — see
        // [`crate::arms::WriterArm::journal`]. Everything below that touches the
        // journal is written against that `Option` rather than assuming one.
        let journal = arms.writer.journal(&blobs);
        // **One ceiling for both databases**, carried in on the config rather
        // than read here: `StoreConfig::from_env` reads it with the arms, and
        // `open`/`open_with` put the environment's value on `DEFAULT` before
        // getting here, so every constructor still honours the variable and
        // `arms.to_string()` prints what was actually applied. Not an arm — see
        // [`crate::arms::redb_cache_bytes`] — redb's own default is 1 GiB *per
        // database*.
        let cache_bytes = arms.redb_cache_bytes;
        let objects = ObjectReadStack::<S>::open(
            &root.join("objects.tail"),
            RebuildTriggers::default(),
            cache_bytes,
        )?;
        let _ = cache_bytes; // no page cache to size: the blob is pread, the index is a slice
        let exploded = ExplodedArchive::open(&root.join("objects.exploded"))?;
        let absorber = Arc::new(Absorber::<S>::open(&blobs, hash, objects, exploded)?);

        // **The `indexed` bit, derived.** Every extent this archive has ever
        // acked, diffed against the rows the index holds. `SafeWriter::create`
        // above appends to that journal and never truncates it, which is what
        // makes the older extents still be here to diff against.
        //
        // An arm with no journal has nothing to derive the bit from, and that is
        // the honest consequence of choosing it rather than a gap: a
        // `FastWriter` store has no durable record that a pack was acked, so a
        // reopen re-queues nothing and there is nothing for it to re-queue.
        let acked = match journal.as_deref() {
            Some(p) if p.exists() => crate::archive_write::read_journal(p).with_context(|| {
                format!(
                    "reading {} — it is the durable half of the indexed bit and a store \
                         cannot be opened without knowing which packs it owes index work for",
                    p.display()
                )
            })?,
            _ => Vec::new(),
        };
        let requeue = absorber.adopt_journal(&acked)?;

        // **The wiring.** The absorber goes into the push path, which hands it to
        // every account indexer it starts, so the drain behind `push_pack` ends
        // in object rows instead of one pack row. Nothing about the ack changes:
        // `append` still returns after the two fsyncs and this is all downstream
        // of that (§13.9).
        let push = PushPath::with_absorber(
            writer,
            &blobs,
            journal,
            absorber.clone() as Arc<dyn ObjectAbsorb>,
        )?;
        let refs = RefLog::new(root.join("refs.log"));
        let archive = root.join("repository.znippy");

        let store = Self {
            root: root.to_path_buf(),
            blobs,
            archive,
            push,
            account: account.to_string(),
            hash,
            absorber,
            refs,
            ref_gate: Mutex::new(()),
            gc: arms.gc.create(),
            arms,
        };
        // The interrupted work, back on the channel it fell off. Nothing reads
        // here and nothing blocks: this is the same 24-byte handoff a push makes,
        // and the drain picks it up on its own thread (§13.9-11).
        let indexer = store.push.indexer(&store.account);
        for p in &requeue {
            indexer.submit(IndexJob {
                pack_id: p.pack_id,
                offset: p.offset,
                len: p.len,
            })?;
        }
        // A store reopened over an existing tail already knows its objects; the
        // derived tables are rebuilt from them rather than from a side-car,
        // because a side-car is a second truth.
        store.refold()?;
        Ok(store)
    }

    /// The arms this store was built with. A description of what was
    /// constructed, not a switch — nothing on a serving path reads it.
    pub fn arms(&self) -> StoreConfig {
        self.arms
    }

    /// The selected writer's name, as it goes on a bench row.
    pub fn writer_name(&self) -> &'static str {
        self.push.name()
    }

    /// What the selected writer's `append` actually promises. Printed next to a
    /// throughput, because a throughput without it is not a comparison.
    pub fn writer_durability(&self) -> &'static str {
        self.push.durability()
    }

    /// The selected [`Gc`] arm.
    pub(crate) fn gc_arm(&self) -> &dyn Gc {
        self.gc.as_ref()
    }

    /// Where the verbatim packs are.
    pub fn blobs_path(&self) -> &Path {
        &self.blobs
    }

    /// The sealed archive GC operates on.
    pub fn archive_path(&self) -> &Path {
        &self.archive
    }

    pub fn root(&self) -> &Path {
        &self.root
    }

    pub fn hash_kind(&self) -> GitHashKind {
        self.hash
    }

    /// The `objects` table, for a caller that wants the index directly (the
    /// bench does).
    pub fn index(&self) -> &ObjectReadStack<S> {
        &self.absorber.objects
    }

    /// Objects in the `objects` table.
    pub fn object_count(&self) -> usize {
        self.absorber.objects.len()
    }

    /// Commits in the graph.
    pub fn commit_count(&self) -> usize {
        self.absorber
            .derived
            .read()
            .expect("derived lock")
            .graph
            .len()
    }

    /// **One object's resolved content** — §14's exploded table, in one lookup.
    ///
    /// Not one of the twelve and deliberately not on [`GitOps`]: the twelve are
    /// the wire vocabulary and they hand back **stored** bytes ([`GitOps::get`]),
    /// which for a delta entry is a delta. This is the other question — *what
    /// does this object contain* — and it is the one §14's table exists to make
    /// a point lookup instead of a chain walk.
    ///
    /// It absorbs any pending pack first, for the same reason
    /// [`GitOps::has`] does: a store with durable bytes it has not indexed cannot
    /// answer "absent" without lying.
    pub fn content(&self, oid: Oid<'_>) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
        if self.unindexed_packs() > 0 {
            self.absorb_pending()?;
        }
        self.absorber.resolved(oid)
    }

    /// §14's table, counted: rows, rows written, reads it served, reads that fell
    /// through to re-deriving from the verbatim truth.
    ///
    /// Applied output. Both content paths return identical bytes, so `served`
    /// against `rederived` is the only thing that can say which one ran.
    pub fn exploded_stats(&self) -> ExplodedStats {
        self.absorber.exploded.engine_stats()
    }

    /// Where §14's table lives. Deleting this file while no store holds it is the
    /// sanctioned way to drop it; the next open re-explodes every pack.
    pub fn exploded_path(&self) -> &Path {
        self.absorber.exploded.path()
    }

    /// Every live row of one kind, with its payload — what the graph fold reads,
    /// exposed so a measurement can ask **the table** how many bytes it holds
    /// rather than dividing a file size by what the pack claimed.
    pub fn exploded_of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        self.absorber.exploded.of_kind(kind)
    }

    /// The commit graph as it stands, generations included. A snapshot: the
    /// caller holds no lock and the next fold replaces it wholesale.
    pub fn graph_snapshot(&self) -> Vec<CommitNode> {
        self.absorber
            .derived
            .read()
            .expect("derived lock")
            .graph
            .clone()
    }

    /// Packs whose objects are not in the index yet — §13.12's bit, counted.
    pub fn unindexed_packs(&self) -> usize {
        self.absorber.unindexed_packs()
    }

    /// This account's indexer: the thread that drains the channel a push queues
    /// on. `is_indexed(pack_id)` on it is §13.12's bit at the pack level, and it
    /// is set only once that pack's **objects** are in the index.
    pub fn indexer(&self) -> std::sync::Arc<crate::indexer::AccountIndexer> {
        self.push.indexer(&self.account)
    }

    /// Block until the background drain has absorbed everything this store has
    /// pushed. What a maintenance tick, a seal or a test uses instead of racing
    /// it.
    pub fn wait_indexed(&self) {
        self.push.pool().wait_caught_up();
    }

    /// **The indexer's half of the split**, and not one of the twelve.
    ///
    /// Runs on the account indexer's worker (§13.9), and on a read that arrives
    /// before that worker got there — the same function either way, which is why
    /// there is no such thing as a background index that behaves differently from
    /// the fallback one.
    pub fn absorb_pending(&self) -> Result<usize> {
        self.absorber.absorb_pending()
    }

    /// **The object ids a stored pack introduced**, and not one of the twelve.
    ///
    /// [`GitOps::put_pack`] cannot answer this and never will: the walk it does
    /// reads pack *entries*, and a pack entry does not carry its oid — that is
    /// why [`crate::pack_walk::Closure`] can only name the `REF_DELTA` bases and
    /// says so. The oids exist once the indexer has absorbed the pack, so this
    /// absorbs first rather than racing the drain, and is therefore the one
    /// caller that deliberately pays §13.9's deferral back.
    ///
    /// **Only call it when something needs the set.** A push whose policy has no
    /// gate over objects must not pay this, which is the whole reason it is a
    /// separate call and not a field on `TxId`.
    ///
    /// The extent *is* the identity: §14 stores a pack verbatim as one
    /// contiguous append, so "introduced by this pack" and "indexed at an offset
    /// inside this pack's extent" are the same set, with no pack column to keep
    /// in step. Bases the push deltas against live in *other* extents and are
    /// excluded by construction — which is what a thin push needs, and what
    /// reading a written `.idx` back could never do.
    ///
    /// Cost is one pass over the index, so it is O(objects in the repository),
    /// not O(objects pushed). Recorded rather than hidden: a per-pack ordinal
    /// range would make it O(pushed), and needs the indexer to keep one.
    pub fn oids_in_extent(&self, extent: Extent) -> Result<Vec<Vec<u8>>> {
        self.absorb_pending()
            .context("absorbing the pending index work before enumerating a pack's objects")?;
        let (start, len) = extent;
        let end = start.saturating_add(len);
        let all = self.index().oids_in_order()?;
        let refs: Vec<&[u8]> = all.iter().map(Vec::as_slice).collect();
        let extents = self.index().extents_batch(&refs);
        Ok(all
            .into_iter()
            .zip(extents)
            .filter_map(|(oid, at)| match at {
                Some((offset, _)) if offset >= start && offset < end => Some(oid),
                _ => None,
            })
            .collect())
    }

    /// **Everything needed to emit a pack for EXACTLY `oids`** — nothing added,
    /// nothing dropped.
    ///
    /// Not one of the twelve. It sits beside [`absorb_pending`](Self::absorb_pending)
    /// for the same reason those do: it is about how storage is *shaped*, not
    /// about what a caller stores or reads.
    ///
    /// # 🔴 It used to add the delta bases, and that shipped broken clones
    ///
    /// Until 2026-08-11 this closed the request over its delta bases: an
    /// `OFS_DELTA` names its base by position, so a base outside the set cannot
    /// be encoded, and *adding the base* looked like the answer that is always
    /// safe. It is not, and the failure is silent on the server:
    ///
    /// ```text
    /// git clone --bare --single-branch --branch base <url>
    ///     fatal: did not receive expected object 8601ec33920b7d701c9887fa04916501136d6e90
    ///     fatal: fetch-pack: invalid index-pack output
    /// event="git.upload_pack.served" objects=33773      ← the server said SUCCESS
    /// ```
    ///
    /// Measured on the `h2h-linear-sha1-2048c-1024f-16k` fixture: `base` reaches
    /// **31 805** objects, the delta-base closure added **1 968** more — 33 773,
    /// exactly the count the server logged — and 507 of those additions were
    /// **trees**. A tree that is not reachable from the wants drags its own
    /// children in as a *requirement*, because `git index-pack
    /// --check-self-contained-and-connected` (which is what a clone runs, and
    /// which sets `strict`) walks every received object's links and then demands
    /// each one exist. Those 507 trees named **204** objects the pack did not
    /// contain. `8601ec33…` is one of them.
    ///
    /// So a delta base pulled into a narrowed request is not free: it satisfies
    /// the *pack format* and violates *connectivity*. Closing over the added
    /// bases' children in turn does terminate — the same fixture reaches a fixed
    /// point at 34 534 objects — but it is an over-send by construction, and on
    /// a `--filter=blob:none` fetch it adds back exactly the blobs the filter
    /// excluded, which is the defect
    /// `emit_pack_emits_exactly_the_set_it_is_given_and_never_its_closure`
    /// exists to forbid.
    ///
    /// # What it does instead: the base decides how the entry is encoded
    ///
    /// The set is never widened. For each requested object:
    ///
    /// * **base inside the request** → the stored entry is copied byte for byte,
    ///   which is the whole point of this engine and is what a full clone does
    ///   for every single object;
    /// * **base outside the request** → the entry is **rebuilt** from §14's
    ///   resolved content and marked
    ///   [`EmitEntry::recompressed`](crate::pack_walk::EmitEntry::recompressed) —
    ///   as a computed delta against a base the request *does* carry where one
    ///   can be found (see [`Self::in_request_chain_ancestor`] and
    ///   [`crate::delta`]), and whole where it cannot.
    ///
    /// That is what stock `pack-objects` does — it reuses a stored delta only
    /// when the base is also being packed — and it keeps the receipt a fact:
    /// `copied` and `recompressed` add up to the objects written, and
    /// `recompressed` is **0 for a whole-repository clone** because such a
    /// selection contains every base. The cost lands on the boundary a narrowed
    /// request cuts through and nowhere else: 1 968 of 33 773 entries, 5.9 %, on
    /// the fixture above.
    ///
    /// `ofs_delta_ok` is the client's `ofs-delta` capability. An old client that
    /// cannot parse one is not refused any more: the entry is re-headed as a
    /// `REF_DELTA` naming its base by oid, **carrying the same compressed delta
    /// payload**, so that arm stays a copy too. Only a base outside the request
    /// costs a rebuild.
    ///
    /// # `thin_haves` — the boundary entry that does NOT have to be rebuilt
    ///
    /// `Some(tips)` says the client consented to a thin pack *and* the
    /// negotiation found common ground; [`crate::serve::GitStore::emit_oids`] is
    /// where the two halves are required together, because either alone is a
    /// failed fetch rather than a smaller one. Given it, a base outside the
    /// request is checked against [`the client's own objects`](GitStore::client_bases),
    /// and one the client already holds is named by oid in a `REF_DELTA` that
    /// carries the **stored delta stream unchanged**. That entry counts as
    /// `copied`, because it is: nothing was inflated and nothing was deflated.
    ///
    /// `None` — a clone, or a client that did not ask for a thin pack — behaves
    /// exactly as before, and a **full clone is byte-identical either way**,
    /// because a whole-repository request has no base outside itself for any of
    /// this to apply to.
    ///
    /// It does nothing for the *clone* half of the boundary cost, and cannot: a
    /// clone's receiver holds nothing, so there is no external base to name.
    /// That half is the arm below it — [`crate::delta`] computes a delta against
    /// something the pack **does** carry, which is the only move left when there
    /// is no receiver to lean on. The two do not overlap: `thin_haves` reaches a
    /// boundary entry first and costs nothing when it fires, so a fetch stays a
    /// pure copy and only a clone pays for a computed delta.
    ///
    /// # An oid this repository does not hold is REFUSED
    ///
    /// It used to be skipped, which is the same silent-under-send shape as the
    /// closure defect above wearing a different hat: N asked, N-1 emitted,
    /// `PackStats` reporting success, and the client the first to know. gunnar's
    /// in-memory arm has always refused it, so the skip also meant the two
    /// engines behind one contract disagreed about what a missing object means.
    /// A partial-clone filter does not depend on the skip and cannot: a filter
    /// *removes* oids from the selection, it never adds one the store lacks.
    ///
    /// # Cost
    ///
    /// O(objects requested). The previous shape read `oids_in_order()` and
    /// `lookup_batch` over the **whole repository** on every call, to build an
    /// offset→oid map it needed only for the closure; with no closure the
    /// selection's own offsets answer "is this base in the request", so that
    /// scan is gone and no reverse index has to exist.
    pub fn emit_set(
        &self,
        oids: &[Oid<'_>],
        ofs_delta_ok: bool,
        thin_haves: Option<&[Oid<'_>]>,
    ) -> Result<Vec<crate::pack_walk::EmitEntry>> {
        use crate::index_layout::{ObjType, ObjectIndex as _};
        use std::collections::hash_map::Entry;

        self.absorb_pending()
            .context("absorbing pending index work before emitting a pack")?;

        // Deduplicate, preserving the caller's order. A repeated oid in the
        // request must not become a repeated entry in the pack.
        //
        // **A map and not a set**, keyed oid → position in `asked` (which is the
        // position in `rows`). The membership question is the one it always was;
        // what the position buys is the base's *archive offset* for a
        // `REF_DELTA`, which names its base by oid and can learn where that base
        // sits no other way. It is a `usize` per requested object on a vector
        // this function already builds — not a second pass and not a second
        // index lookup.
        let mut asked: Vec<&[u8]> = Vec::with_capacity(oids.len());
        let mut seen: HashMap<&[u8], usize> = HashMap::with_capacity(oids.len());
        for oid in oids {
            if let Entry::Vacant(slot) = seen.entry(*oid) {
                slot.insert(asked.len());
                asked.push(*oid);
            }
        }

        let rows = self.index().lookup_batch(&asked);

        // **One mapping for the whole emit set.** Only the header probes below
        // read through it — the entries themselves are recorded as extents and
        // resolved by [`Self::resolve_emit_payloads`], which takes its own
        // snapshot of the same [`ArchiveMap`](crate::archive_map::ArchiveMap) and
        // therefore the same `Arc` unless a push landed in between.
        let snap = self.archive_snapshot()?;

        // The two questions "is this base in the request" are asked in the two
        // coordinate systems the two delta forms use, and both are answered off
        // the request itself rather than off the index.
        //
        // **Offset → position in the request**, not a bare set of offsets. The
        // membership question is the same one it always was; what the position
        // buys is the base's *oid*, which the re-delta arm below needs in order
        // to read the base's content. It is the same single pass over `rows`
        // that built the set, so it costs a `usize` per requested object and no
        // extra work — and it is emphatically not the whole-repository
        // offset→oid scan the old delta-base closure used to do.
        let in_set_offsets: HashMap<u64, usize> = rows
            .iter()
            .enumerate()
            .filter_map(|(i, r)| r.as_ref().map(|r| (r.offset, i)))
            .collect();
        let oid_len = self.hash_kind().oid_len();

        // The base of the last boundary entry that was re-deltified, kept so a
        // run of entries hanging off one ancestor inflates it once. One slot and
        // not an LRU: measured on `h2h-linear-sha1-2048c-1024f-16k`, 961
        // boundary entries name 855 distinct stored bases, so there is no reuse
        // worth a cache that could hold a repository's worth of objects — and
        // this crate is under a live RSS budget.
        let mut base_content: Option<(u64, Vec<u8>)> = None;

        // What the client already holds, built **lazily and at most once**. A
        // clone never asks for it (`thin_haves` is `None`), and a fetch every one
        // of whose bases is inside the request never reaches the line that builds
        // it either — only a request that actually cuts a delta chain pays, which
        // is the only request it can help.
        let mut client_bases: Option<ExternalBases> = None;

        let mut out = Vec::with_capacity(asked.len());
        for (oid, row) in asked.iter().zip(rows.iter()) {
            // **An oid this repository does not hold is REFUSED, not dropped.**
            //
            // This used to `continue`, and that is the same silent-under-send
            // shape as the closure defect above: the request asks for N objects,
            // the pack carries N-1, the receipt says success and the client is
            // the first to find out. It also made the two engines disagree about
            // what a missing object *means* — gunnar's in-memory arm has always
            // said *"emit_pack was asked for X, which this store does not hold;
            // a pack emitter that skipped it would send a pack no client can
            // close"* — so a benchmark across the two was comparing two
            // contracts.
            //
            // No legitimate caller relies on the skip. Every set that reaches
            // here is built out of this store's own index: `select` answers from
            // the reachability bitmaps, and gunnar's own walk resolves each id
            // against the store before it selects it. A partial-clone filter is
            // no exception and is the case worth naming, because it is the one
            // that *looks* like it might be: `--filter=blob:none` **removes**
            // oids from the selection, it never adds one the store lacks, so a
            // filtered request is a smaller set of objects that are all present
            // and this refusal never fires on it.
            let Some(row) = row.as_ref() else {
                bail!(
                    "this pack was asked for {}, which this repository does not hold; emitting \
                     the rest would be a short pack reported as a success, so this refuses \
                     instead",
                    hex::encode(oid)
                );
            };

            // Which base this entry names, if any, and whether the request holds
            // it. A `REF_DELTA`'s `delta_base` column is 0 by construction —
            // `DeltaBase::as_offset` reports 0 for a ref base and 0 is also the
            // no-base sentinel — so its base is read out of the entry's own
            // bytes, directly after the type/size varint.
            //
            // A base that is *outside* the request is carried in **both**
            // coordinate systems rather than as a bare `false`, because that is
            // what the thin arm below needs: an `OFS_DELTA` knows only where its
            // base sits, a `REF_DELTA` knows only what it is called, and
            // "does the client hold it" has to be asked in whichever one the
            // entry speaks.
            let base = match row.obj_type {
                ObjType::OfsDelta => match in_set_offsets.get(&row.delta_base) {
                    Some(_) => BaseOf::Inside {
                        at: row.delta_base,
                    },
                    None => BaseOf::Outside {
                        offset: row.delta_base,
                        oid: None,
                    },
                },
                ObjType::RefDelta => {
                    let head = self.extent(&snap, row.offset, row.len.min(64))?;
                    let (_, _, n) = crate::pack_walk::type_and_size_of(&head)?;
                    match head.get(n..n + oid_len) {
                        // In the request — **and where**. `rows[i]` is `None`
                        // only for an oid this store does not hold, which the
                        // refusal above turns into an error the moment the loop
                        // reaches it; until then the base is treated as outside,
                        // which is the answer that cannot be wrong.
                        Some(b) => match seen
                            .get(b)
                            .and_then(|i| rows[*i].as_ref())
                            .map(|r| r.offset)
                        {
                            Some(at) => BaseOf::Inside { at },
                            None => BaseOf::Outside {
                                offset: 0,
                                oid: Some(b.to_vec()),
                            },
                        },
                        // A truncated entry names no base this can read. It is
                        // not "inside", and it is not offered to the client
                        // either — it falls through to the whole rebuild, which
                        // is the answer that cannot be wrong.
                        None => BaseOf::Outside {
                            offset: 0,
                            oid: None,
                        },
                    }
                }
                _ => BaseOf::Whole,
            };

            if let BaseOf::Whole | BaseOf::Inside { .. } = base {
                let inside_at = match base {
                    BaseOf::Inside { at } => Some(at),
                    _ => None,
                };
                // **The whole-clone arm, and it reads not one byte here.** The
                // entry is recorded as its address in the archive; the bytes are
                // resolved against the mapping at emit time and go straight from
                // the page cache to the wire. This line used to be a `pread`
                // into a fresh `Vec` — 13.8 M of each on a `linux.git` clone, all
                // of them held at once, which is the 2314 MB peak RSS
                // [`crate::pack_walk::EntryBytes`] documents.
                let extent = crate::pack_walk::EntryBytes::Extent {
                    offset: row.offset,
                    len: row.len,
                };
                let (stored, obj_type, delta_base) = match (row.obj_type, inside_at) {
                    // An `OFS_DELTA` for a client that cannot read one: re-head it
                    // as a `REF_DELTA` naming the same base by oid. The compressed
                    // payload is the delta stream either way, so this is still a
                    // copy in the `recompressed` sense — nothing is inflated — but
                    // the concatenation of a new header and an old payload exists
                    // nowhere on disk, so it cannot be an extent and is `Owned`.
                    (ObjType::OfsDelta, Some(_)) if !ofs_delta_ok => {
                        let base_oid = self.oid_at_offset(row.delta_base, &asked, &rows)?;
                        let bytes = self.extent(&snap, row.offset, row.len)?;
                        (
                            crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(
                                &bytes, &base_oid,
                            )?),
                            ObjType::RefDelta,
                            0,
                        )
                    }
                    // 🚨 **A stored `REF_DELTA` whose base is IN THIS PACK is
                    // re-headed as an `OFS_DELTA`, and that is not an
                    // optimisation.**
                    //
                    // gitoxide — which is `gunnar_client`, `gunnar`'s own
                    // `receive-pack`, and every other gix-based receiver — reads
                    // every `OBJ_REF_DELTA` in an incoming pack as naming an
                    // object *the receiver already has*.
                    // `gix_pack::data::input::LookupRefDeltaObjectsIter` consults
                    // the local object database and the bases it has already
                    // spliced in, and **nowhere else**: never inside the pack it
                    // is reading. A base that is in the pack and not in the
                    // receiver's store is `input::Error::NotFound`, and the fetch
                    // dies with `The object <base> could not be decoded or wasn't
                    // found` — with the whole pack rejected, on a clone whose
                    // target is empty by definition, so *every* such entry is
                    // fatal. That is `S-003` / gitoxide#2882, and `gunnar-wire`'s
                    // `Selection::client_has` carries the same rule for gunnar's
                    // own emitter.
                    //
                    // Stock git resolves it happily, which is why this survived
                    // five green `git clone` runs against the store that could
                    // not serve one `gunnar_client` fetch (`gunnar.rewrite`
                    // round 0 and `gunnar.year` week 0, all four znippy columns,
                    // 2026-08-14 — the server logged the same request `served`).
                    //
                    // The entry keeps its compressed delta stream byte for byte;
                    // only the header changes, from "base named by oid" to "base
                    // named by distance", which is the naming every receiver in
                    // existence resolves inside the pack. This is where such an
                    // entry comes from in the first place: `git index-pack
                    // --fix-thin` appends the base into the pack and leaves the
                    // ref-delta naming it by oid, so any repository pushed to
                    // more than once holds them.
                    //
                    // # What it costs, stated rather than glossed
                    //
                    // `Owned`, so this is the one arm that gives up the zero-copy
                    // property for an entry it does **not** rebuild: the
                    // compressed delta stream is memcpy'd once, because a new
                    // header concatenated with an old payload exists nowhere on
                    // disk and no extent can address it. Bounded by the
                    // compressed size of the ref-deltas whose base is in the
                    // request — not by the repository, and not by the pack.
                    //
                    // It could be avoided: `emit_pack` re-encodes every header
                    // from `EmitEntry::obj_type` while `header_len` finds the
                    // payload from the STORED bytes' own type, so an `Extent`
                    // over the untouched ref-delta declared as an `OfsDelta`
                    // would emit the identical pack with no copy at all. That is
                    // deliberately not done: it makes an `EmitEntry` disagree
                    // with the bytes it points at, and the entry model being
                    // trustworthy is worth more than the memcpy. If the RSS
                    // budget ever says otherwise, the honest form is a third
                    // `EntryBytes` variant that carries the new header beside
                    // the extent, not a silent disagreement.
                    (ObjType::RefDelta, Some(at)) if ofs_delta_ok => {
                        let bytes = self.extent(&snap, row.offset, row.len)?;
                        (
                            crate::pack_walk::EntryBytes::Owned(Self::as_ofs_delta(
                                &bytes, oid_len, row.offset, at,
                            )?),
                            ObjType::OfsDelta,
                            at,
                        )
                    }
                    // Everything else copies. A `REF_DELTA` left as one here is
                    // the `!ofs_delta_ok` client, which cannot be handed a
                    // distance at all — and which is stock git older than 1.4.4,
                    // never a gix receiver, since gix has always advertised
                    // `ofs-delta`.
                    _ => (extent, row.obj_type, row.delta_base),
                };
                out.push(crate::pack_walk::EmitEntry {
                    oid: oid.to_vec(),
                    stored,
                    obj_type,
                    uncompressed_size: row.uncompressed_size,
                    delta_base,
                    offset: row.offset,
                    recompressed: false,
                    deltified: false,
                });
                continue;
            }

            // ── the base is outside the request ──────────────────────────────
            //
            // **Thin first.** If the client consented to a thin pack and the
            // negotiation vouches that it already holds this base, the entry
            // goes out as a `REF_DELTA` naming that base by oid, carrying the
            // **same compressed delta stream**. Nothing is inflated, nothing is
            // deflated, and the pack is smaller than the whole object by exactly
            // the margin the delta was worth. That is what `caps.thin` has always
            // been the allowance for.
            //
            // An `OFS_DELTA` cannot survive this: it names its base by a
            // backwards distance *within the pack*, and the base is not in the
            // pack. Re-heading it as a `REF_DELTA` is the same header swap the
            // no-`ofs-delta` client path does, over an unchanged payload.
            if let Some(haves) = thin_haves {
                if client_bases.is_none() {
                    client_bases = Some(self.client_bases(haves)?);
                }
                let ext = client_bases
                    .as_ref()
                    .expect("the client's bases were just built");
                let base_oid: Option<Vec<u8>> = match &base {
                    // Unreachable: the arm above `continue`s on both. Matched
                    // rather than `unreachable!()` so that a future arm falling
                    // through here declines the thin path instead of panicking
                    // inside a serving thread.
                    BaseOf::Whole | BaseOf::Inside { .. } => None,
                    BaseOf::Outside {
                        oid: Some(named), ..
                    } => ext
                        .held
                        .contains(&OidKey::new(named)?)
                        .then(|| named.clone()),
                    BaseOf::Outside {
                        offset,
                        oid: None,
                    } => ext.by_offset.get(offset).map(|k| k.as_slice().to_vec()),
                };
                if let Some(base_oid) = base_oid {
                    // A stored `REF_DELTA` already names its base by oid, so its
                    // bytes go out untouched and stay an extent. An `OFS_DELTA`
                    // has to be re-headed, and the result is new bytes.
                    let stored = if row.obj_type == ObjType::OfsDelta {
                        let bytes = self.extent(&snap, row.offset, row.len)?;
                        crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(&bytes, &base_oid)?)
                    } else {
                        crate::pack_walk::EntryBytes::Extent {
                            offset: row.offset,
                            len: row.len,
                        }
                    };
                    out.push(crate::pack_walk::EmitEntry {
                        oid: oid.to_vec(),
                        stored,
                        obj_type: ObjType::RefDelta,
                        uncompressed_size: row.uncompressed_size,
                        // The base is NOT in this pack, so nothing may be ordered
                        // after it: `topological_order` reads 0 as "names no base
                        // in here", which for an external base is the truth.
                        delta_base: 0,
                        offset: row.offset,
                        // Copied. The payload never left the archive's bytes.
                        recompressed: false,
                        deltified: false,
                    });
                    continue;
                }
            }

            // The base is not being sent and the client cannot supply it, so the
            // stored bytes cannot go out as they are. Two answers remain, and
            // the first is worth ~4× the second on the wire.
            let (kind, body) = self.content(oid)?.ok_or_else(|| {
                anyhow!(
                    "{} is a delta whose base is outside this request, so it must be sent whole, \
                     and this repository cannot produce its content — neither §14's exploded \
                     table nor a re-derivation from the verbatim packs answered",
                    hex::encode(oid)
                )
            })?;

            // ── re-delta against a base the request DOES carry ────────────────
            //
            // The candidate comes from the entry's own stored delta chain: the
            // nearest ancestor that is inside the request. That is the packer's
            // original similarity judgement, already made and already written
            // down — no window, no sort, no scan of the repository.
            let candidate = if crate::delta::enabled() {
                self.in_request_chain_ancestor(row.delta_base, row.offset, &in_set_offsets)?
            } else {
                None
            };
            if let Some(base_at) = candidate {
                let at = *in_set_offsets
                    .get(&base_at)
                    .expect("the ancestor was found by lookup in this very map");
                // Reuse the last base if this entry hangs off the same one.
                if base_content.as_ref().is_none_or(|(o, _)| *o != base_at) {
                    let base_oid = asked[at];
                    let (_, base_body) = self.content(base_oid)?.ok_or_else(|| {
                        anyhow!(
                            "{} is in this request and is the delta base chosen for {}, and this \
                             repository cannot produce its content",
                            hex::encode(base_oid),
                            hex::encode(oid)
                        )
                    })?;
                    base_content = Some((base_at, base_body));
                }
                let (_, base_body) = base_content.as_ref().expect("just filled");

                // `None` is the encoder declining — an unrelated base, or a
                // delta that came out no smaller than the object. Falling
                // through to the whole rebuild is then strictly right.
                if let Some(d) = crate::delta::delta(base_body, &body)? {
                    let stored = Self::ofs_delta_entry(&d, row.offset, base_at)?;
                    // A client with no `ofs-delta` is served the same computed
                    // delta under a `REF_DELTA` head naming the base by oid —
                    // the identical header swap the base-inside arm above does,
                    // over a payload that is already built.
                    // **The `Owned` exception, in its primary shape.** These
                    // bytes were computed here and exist on no disk, so no
                    // extent addresses them; carrying them on the entry is
                    // correct and is bounded by the boundary a narrowed request
                    // cuts through (961 of 33 773 entries on the fixture below),
                    // never by the repository.
                    let (stored, obj_type, delta_base) = if ofs_delta_ok {
                        (
                            crate::pack_walk::EntryBytes::Owned(stored),
                            ObjType::OfsDelta,
                            base_at,
                        )
                    } else {
                        (
                            crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(
                                &stored, asked[at],
                            )?),
                            ObjType::RefDelta,
                            0,
                        )
                    };
                    out.push(crate::pack_walk::EmitEntry {
                        oid: oid.to_vec(),
                        stored,
                        obj_type,
                        uncompressed_size: row.uncompressed_size,
                        delta_base,
                        offset: row.offset,
                        // Inflated and re-deflated, so NOT a copy — and the
                        // receipt says which of the two rebuilds ran.
                        recompressed: true,
                        deltified: true,
                    });
                    continue;
                }
            }

            // Ship the object whole rather than dragging an unreachable base —
            // and its children — into a clone.
            out.push(crate::pack_walk::EmitEntry {
                oid: oid.to_vec(),
                // Inflated and re-deflated from §14's resolved content: the
                // `Owned` exception again, and again bounded by the boundary.
                stored: crate::pack_walk::EntryBytes::Owned(Self::whole_entry(kind, &body)?),
                obj_type: crate::serve::resolved_type(kind),
                uncompressed_size: body.len() as u64,
                // Whole: it names no base, and nothing may order it after one.
                delta_base: 0,
                offset: row.offset,
                recompressed: true,
                deltified: false,
            });
        }
        Ok(out)
    }

    /// **PHASE 1 of emitting: resolve every entry's stored bytes, in parallel.**
    ///
    /// Emission splits in two, and the split is not a preference — it is where
    /// the ordering dependency actually is:
    ///
    /// | phase | what it does | why |
    /// |---|---|---|
    /// | 1, **here**, parallel | address → bytes for every entry | the archive is immutable, so N workers over disjoint (or shared) extents need **no lock at all** |
    /// | 2, [`crate::pack_walk::emit_pack`], serial | assign output offsets, encode `OFS_DELTA` distances, write, hash | an `OFS_DELTA` names its base by **distance back in the output pack**, so entry *n* cannot be encoded until every earlier entry's byte length is known |
    ///
    /// Phase 2 is not parallelisable and no attempt is made: the distance chain
    /// is a genuine serial dependency, and a running sha1 is another.
    ///
    /// # No lock, and why that is a fact rather than an intention
    ///
    /// [`Self::adopt_journal`] states the storage property this rests on — *"the
    /// blob file is append-only and `gc` truncates nothing, so a retired pack's
    /// bytes are still exactly where the journal says they are"*. Every byte a
    /// snapshot covers is therefore immutable for the life of the file, so the
    /// workers below share one `&Mapped` and read it concurrently with no
    /// synchronisation of any kind. There is no mutex here; if one were needed
    /// the data model would be wrong.
    ///
    /// # 🚨 This is a fan-out inside the serving tier, and `crate::serve`'s
    /// header says that tier is serial
    ///
    /// It did, and the reason it gave is still true and still respected: the
    /// constellation's fan-out primitive has no reentrancy detection, so an
    /// unbounded fan-out per request inside an admission gate that already runs
    /// 32 transfers spawns W² threads. That is why this is **bounded twice**:
    ///
    ///  * it does not fan out at all below [`FAN_OUT_AT`] entries, so every fetch
    ///    small enough for thread-spawn to dominate runs on the calling thread
    ///    exactly as before, and
    ///  * it never exceeds [`MAX_RESOLVE_WORKERS`] (override with
    ///    `ZNIPPY_GIT_EMIT_WORKERS`), so the worst case is 32 × 4 and not
    ///    32 × ncores.
    ///
    /// # What the fan-out is worth — stated honestly
    ///
    /// Resolving a mapped extent is a bounds check and a pointer, so the work
    /// this parallelises is **not** the resolve itself. It is the two things
    /// beside it: the first-touch **page fault** on each entry's header (the
    /// grammar is parsed here, which is what forces that fault, and it validates
    /// the entry before phase 2 commits a byte to the wire), and the `pread`
    /// fallback for any extent the mapping does not cover. On a warm page cache
    /// with the whole archive mapped, phase 1 is nearly free and the fan-out
    /// buys correspondingly little — the 20 % of cycles this change is aimed at
    /// is recovered by *not copying*, not by threads. I could not measure the
    /// fan-out's own contribution separately: oden was at loadavg 25.9 while
    /// this was written (2026-08-14), and a shared box cannot answer a question
    /// that fine.
    pub(crate) fn resolve_emit_payloads<'a>(
        &self,
        entries: &'a [crate::pack_walk::EmitEntry],
        snap: &'a crate::archive_map::Mapped,
    ) -> Result<Vec<std::borrow::Cow<'a, [u8]>>>
    where
        Self: Sync,
    {
        let workers = if entries.len() < FAN_OUT_AT {
            1
        } else {
            emit_workers()
        };
        self.resolve_with(entries, snap, workers)
    }

    /// [`Self::resolve_emit_payloads`] with the worker count named rather than
    /// decided — `1` runs inline on the calling thread.
    ///
    /// Split out so the fan-out is **testable at a size a test fixture can
    /// reach**. The production threshold is 4096 entries and the corpus these
    /// guards run on holds 2687, so a test that went through the front door
    /// would exercise the serial arm every time and the gatling arm never;
    /// `the_fan_out_resolves_the_identical_bytes_the_serial_pass_does` calls
    /// this with 1 and with 4 over the same set and requires the two to agree
    /// byte for byte.
    pub(crate) fn resolve_with<'a>(
        &self,
        entries: &'a [crate::pack_walk::EmitEntry],
        snap: &'a crate::archive_map::Mapped,
        workers: usize,
    ) -> Result<Vec<std::borrow::Cow<'a, [u8]>>>
    where
        Self: Sync,
    {
        use znippy_zoomies::gatling_forkjoin::gatling_for_each;

        let one = |i: usize| -> Result<std::borrow::Cow<'a, [u8]>> {
            let e = &entries[i];
            let bytes = match &e.stored {
                // The bytes live on the entry — the rebuilt-delta exception.
                // Borrowed, never cloned: `EntryBytes::Owned` is already the
                // only allocation this path makes and copying it would double it.
                crate::pack_walk::EntryBytes::Owned(v) => std::borrow::Cow::Borrowed(v.as_slice()),
                crate::pack_walk::EntryBytes::Extent { offset, len } => {
                    self.extent(snap, *offset, *len)?
                }
            };
            // Parse the entry's own header now. Two things fall out: a malformed
            // entry is refused **before** phase 2 has written any of the pack,
            // and the first cache line of every entry is touched here — on
            // whichever worker got it — rather than one at a time on the serial
            // writer.
            crate::pack_walk::type_and_size_of(&bytes).with_context(|| {
                format!(
                    "resolving the stored bytes of {} for emission",
                    hex::encode(&e.oid)
                )
            })?;
            Ok(bytes)
        };

        if workers <= 1 {
            return (0..entries.len()).map(one).collect();
        }
        gatling_for_each(entries.len(), workers, one)
            .into_iter()
            .collect()
    }

    /// **Both phases, one call** — the only way a caller in this crate turns an
    /// ordered emit set into pack bytes.
    ///
    /// It is three lines, and it is a function rather than three lines because
    /// LAW 5's *fix by construction* applies exactly here: the snapshot has to
    /// outlive the slices taken from it, and a caller that took its own snapshot
    /// per entry, or dropped it between the phases, would be writing a pack out
    /// of a mapping that no longer exists. Routing `emit_oids` and every test
    /// through one writer makes that impossible to get wrong twice.
    pub(crate) fn emit_ordered(
        &self,
        ordered: &[crate::pack_walk::EmitEntry],
        out: &mut dyn std::io::Write,
    ) -> Result<crate::pack_walk::EmitReport>
    where
        Self: Sync,
    {
        let snap = self
            .archive_snapshot()
            .context("mapping the archive to emit")?;
        let payloads = self
            .resolve_emit_payloads(ordered, &snap)
            .context("resolving the emit set's stored bytes")?;
        crate::pack_walk::emit_pack(
            ordered,
            self.hash_kind(),
            out,
            &|i| Ok(payloads[i].as_ref()),
        )
    }

    /// **The nearest ancestor of `from`'s stored delta chain that is inside the
    /// request**, or `None` when the chain leaves the request and never comes
    /// back.
    ///
    /// `from` is the archive offset of a base already known to be *outside* the
    /// request; `entry_at` is the offset of the entry that named it.
    ///
    /// # Why this cannot make a cycle, and why that is not a comment but an
    /// invariant
    ///
    /// The answer is required to satisfy `answer < entry_at`. In a well-formed
    /// pack a base always precedes the delta that names it, so every edge in the
    /// stored graph already points from a higher offset to a lower one; a new
    /// edge with the same property keeps the emitted graph a DAG **by
    /// construction**, whatever else the request contains. That matters because
    /// the obvious better candidate — the entry's own delta *children*, one
    /// revision away instead of two to nine — is exactly the set that would
    /// close a cycle: a child's stored base is this entry, so pointing this
    /// entry at the child makes `topological_order` unable to order either, and
    /// `emit_pack` refuses the set. Measured cost of the restriction on
    /// `h2h-linear-sha1-2048c-1024f-16k`: ~1 470 bytes per boundary entry
    /// against ~314 for the cyclic candidate, and 5 800 for shipping it whole.
    ///
    /// An `OFS_DELTA` decreases the offset every hop, so the bound is free
    /// there; a `REF_DELTA` names its base by oid and could point anywhere, so
    /// the comparison is made rather than assumed.
    ///
    /// # Cost
    ///
    /// One 64-byte `pread` per hop, no inflate, no allocation sized by an
    /// object — the same walk and the same ceiling as
    /// [`crate::serve::GitStore::resolved_type_at`], for the same reason: a
    /// pushed pack's back-references are not under this server's control, so a
    /// cycle must terminate rather than spin. Measured hop counts on the fixture
    /// above: 2 for 392 of the 859 that resolve, 3 for 260, and 9 at the worst.
    fn in_request_chain_ancestor(
        &self,
        from: u64,
        entry_at: u64,
        in_request: &HashMap<u64, usize>,
    ) -> Result<Option<u64>> {
        use crate::index_layout::{ObjType, ObjectIndex as _};

        /// git's own `pack.depth` ceiling, doubled — the same bound
        /// `resolved_type_at` carries and for the same reason.
        const MAX_LINKS: usize = 100;
        /// Enough of an entry to read its type/size varint plus either an
        /// `OFS_DELTA` distance or a `REF_DELTA` oid.
        const HEADER_PROBE: u64 = 64;

        let oid_len = self.hash_kind().oid_len();
        let mut at = from;
        for _ in 0..MAX_LINKS {
            if at != 0 && at < entry_at && in_request.contains_key(&at) {
                return Ok(Some(at));
            }
            if at == 0 {
                return Ok(None);
            }
            let head = self.read_extent(at, HEADER_PROBE)?;
            let (t, _, n) = crate::pack_walk::type_and_size_of(&head)?;
            at = match t {
                // A whole object with no base: the chain ends here, and it was
                // not in the request.
                ObjType::Commit | ObjType::Tree | ObjType::Blob | ObjType::Tag => {
                    return Ok(None)
                }
                ObjType::OfsDelta => {
                    let (distance, _) = crate::pack_walk::ofs_distance_of(&head[n..])?;
                    match at.checked_sub(distance) {
                        Some(next) if next != 0 => next,
                        // A base before the start of the archive is a corrupt
                        // back-reference. The whole rebuild is always a correct
                        // answer, so this declines rather than failing a clone
                        // over an entry it was only ever going to optimise.
                        _ => return Ok(None),
                    }
                }
                ObjType::RefDelta => {
                    let Some(base_oid) = head.get(n..n + oid_len) else {
                        return Ok(None);
                    };
                    match self.index().lookup(base_oid) {
                        Some(row) => row.offset,
                        None => return Ok(None),
                    }
                }
            };
        }
        Ok(None)
    }

    /// An `OFS_DELTA` pack entry carrying `d`, deflated.
    ///
    /// The distance written here is the **archive** distance, which is the same
    /// coordinate every stored entry's distance is in;
    /// [`crate::pack_walk::emit_pack`] rewrites it into output coordinates for
    /// every entry it emits, copied or not, so this only has to be well-formed
    /// and self-consistent — [`crate::pack_walk::header_len`] parses it back to
    /// find where the payload starts.
    ///
    /// `Compression::fast` for the same reason [`Self::whole_entry`] uses it,
    /// and the trade is far better here: the buffer being deflated is the delta,
    /// which on the measured fixture is ~1.4 KB against the object's 16 KiB, so
    /// this path deflates roughly a **tenth** of the bytes the whole rebuild it
    /// replaces does.
    fn ofs_delta_entry(d: &[u8], entry_at: u64, base_at: u64) -> Result<Vec<u8>> {
        use crate::index_layout::ObjType;
        use std::io::Write as _;

        let distance = entry_at.checked_sub(base_at).ok_or_else(|| {
            anyhow!(
                "a computed delta at archive offset {entry_at} names a base at {base_at}, which \
                 is after it — an ofs-delta distance is backwards and this would not encode"
            )
        })?;
        let mut out = Vec::with_capacity(d.len() / 2 + 32);
        crate::pack_walk::encode_type_and_size(&mut out, ObjType::OfsDelta, d.len() as u64);
        crate::pack_walk::encode_ofs_distance(&mut out, distance);
        let mut z = flate2::write::ZlibEncoder::new(out, flate2::Compression::fast());
        z.write_all(d)?;
        Ok(z.finish()?)
    }

    /// **The objects the receiver already held before this transfer**, in both
    /// coordinate systems a delta can name a base in.
    ///
    /// `haves` is the negotiated common **tips**, exactly as
    /// [`git_storage_trait::GitServe::emit_pack`] defines them, and the closure
    /// over them is this store's own — the contract deliberately passes tips
    /// rather than the walk's exclusion set, because the closure is
    /// repository-sized and the engine can rebuild it from its own bitmaps.
    ///
    /// # This is a VOUCHER, and the caller has to be able to stand behind it
    ///
    /// Every object named here may be pointed at by a delta the pack does not
    /// carry. If the receiver turns out **not** to hold one, its
    /// `index-pack --fix-thin` dies with `pack has N unresolved deltas`, so a
    /// caller that cannot vouch must not pass `caps.thin`. The case that matters
    /// is a **partial clone**: a `blob:none` client's `have` tips reach blobs it
    /// was deliberately never sent, so the closure over-states what it holds and
    /// a filtered fetch must not be marked thin.
    ///
    /// # Cost
    ///
    /// One bitmap union plus one point lookup per held object, and it happens at
    /// most once per request — and only on a request that has a base outside
    /// itself to place. The `by_offset` map exists because an `OFS_DELTA` knows
    /// its base only as an archive offset; building it over what the *client*
    /// holds rather than over the repository is what keeps this off the
    /// whole-repository scan the closure used to do.
    ///
    /// **And it is spent in raw bytes now, end to end.** This used to ask
    /// [`GitOps::reachable`] — hex out of the ordinal space, `hex::decode`d
    /// straight back to bytes, then a third `to_vec` per object to key the
    /// offset map, three heap allocations per held object for a set the size of
    /// the client's history. It asks [`GitStore::reachable_raw`] instead: one
    /// buffer for the closure, one hash table of [`OidKey`]s over it, and
    /// nothing per object.
    fn client_bases(&self, haves: &[Oid<'_>]) -> Result<ExternalBases> {
        use crate::index_layout::ObjectIndex as _;

        let held = self.reachable_raw(haves, &[]).context(
            "closing over the negotiated common tips, to learn which bases the client can supply \
             for itself",
        )?;
        let refs: Vec<&[u8]> = held.iter().collect();
        let rows = self.index().lookup_batch(&refs);
        let mut by_offset = HashMap::with_capacity(refs.len());
        let mut keys = std::collections::HashSet::with_capacity(refs.len());
        for (oid, row) in refs.iter().zip(rows.iter()) {
            let key = OidKey::new(oid)?;
            if let Some(row) = row {
                by_offset.insert(row.offset, key);
            }
            keys.insert(key);
        }
        Ok(ExternalBases {
            by_offset,
            held: keys,
        })
    }

    /// The oid of the entry at `offset`, looked up **in the request** — which is
    /// the only place it can be, because this is called for a base the request
    /// was just shown to contain.
    ///
    /// Linear over the request rather than a map, because it runs only on the
    /// `ofs-delta`-less client path: a capability no git since 1.4.4 omits, and
    /// one that has to be *advertised absent* to reach here at all.
    fn oid_at_offset(
        &self,
        offset: u64,
        asked: &[&[u8]],
        rows: &[Option<crate::index_layout::IndexRow>],
    ) -> Result<Vec<u8>> {
        asked
            .iter()
            .zip(rows)
            .find(|(_, r)| r.as_ref().is_some_and(|r| r.offset == offset))
            .map(|(oid, _)| oid.to_vec())
            .ok_or_else(|| {
                anyhow!(
                    "the entry at archive offset {offset} was checked to be in this request and \
                     then could not be found in it"
                )
            })
    }

    /// Re-head a stored `OFS_DELTA` as a `REF_DELTA` naming `base_oid`.
    ///
    /// The compressed delta stream is copied unchanged; only the header differs,
    /// because the two forms differ in **how the base is named** and in nothing
    /// else. This is why a client without `ofs-delta` costs bytes and not CPU.
    fn as_ref_delta(stored: &[u8], base_oid: &[u8]) -> Result<Vec<u8>> {
        use crate::index_layout::ObjType;
        let (t, stated_size, n) = crate::pack_walk::type_and_size_of(stored)?;
        if t != ObjType::OfsDelta {
            bail!("only an ofs-delta can be re-headed as a ref-delta, this entry is {t:?}");
        }
        let (_, d) = crate::pack_walk::ofs_distance_of(stored.get(n..).unwrap_or(&[]))?;
        let mut out = Vec::with_capacity(stored.len() + base_oid.len());
        crate::pack_walk::encode_type_and_size(&mut out, ObjType::RefDelta, stated_size);
        out.extend_from_slice(base_oid);
        out.extend_from_slice(stored.get(n + d..).unwrap_or(&[]));
        Ok(out)
    }

    /// Re-head a stored `REF_DELTA` as an `OFS_DELTA` naming the base that sits
    /// at `base_at` — **the inverse of [`Self::as_ref_delta`]**, and the fix for
    /// the one pack shape gitoxide cannot read.
    ///
    /// The compressed delta stream is copied unchanged; only the header differs,
    /// because the two forms differ in how the base is *named* and in nothing
    /// else. The reason this has to exist is in the caller: a `REF_DELTA` whose
    /// base is inside the same pack is resolved by stock git and refused by
    /// every gix-based receiver.
    ///
    /// # The distance written here is a placeholder, and that is by design
    ///
    /// Exactly as [`Self::ofs_delta_entry`] documents: the distance an
    /// `OFS_DELTA` carries is relative to a position in the pack it is *in*, so
    /// [`crate::pack_walk::emit_pack`] re-encodes it for every entry from the
    /// output offsets it is assigning, and [`crate::pack_walk::header_len`]
    /// parses whatever is written here only to find where the payload starts. It
    /// therefore has to be well-formed and self-consistent, and nothing more.
    ///
    /// It cannot simply be `entry_at - base_at`: a `REF_DELTA` is allowed to
    /// name a base that comes **after** it in the pack, and `git index-pack
    /// --fix-thin` produces exactly that — it appends the base at the end. So a
    /// forward reference encodes as `1`, and `emit_pack` writes the real
    /// backwards distance once `topological_order` has put the base first.
    fn as_ofs_delta(
        stored: &[u8],
        oid_len: usize,
        entry_at: u64,
        base_at: u64,
    ) -> Result<Vec<u8>> {
        use crate::index_layout::ObjType;
        let (t, stated_size, n) = crate::pack_walk::type_and_size_of(stored)?;
        if t != ObjType::RefDelta {
            bail!("only a ref-delta can be re-headed as an ofs-delta, this entry is {t:?}");
        }
        let payload = stored.get(n + oid_len..).ok_or_else(|| {
            anyhow!(
                "a ref-delta entry at archive offset {entry_at} is {} bytes, which is not even its \
                 header plus a {oid_len}-byte base oid",
                stored.len()
            )
        })?;
        let mut out = Vec::with_capacity(payload.len() + 32);
        crate::pack_walk::encode_type_and_size(&mut out, ObjType::OfsDelta, stated_size);
        crate::pack_walk::encode_ofs_distance(
            &mut out,
            entry_at.checked_sub(base_at).filter(|d| *d > 0).unwrap_or(1),
        );
        out.extend_from_slice(payload);
        Ok(out)
    }

    /// A whole (non-delta) pack entry for `body`: type/size header, then the
    /// body deflated.
    ///
    /// The **only** place in this crate that deflates on a serving path, and the
    /// only reason it exists is a delta whose base the request does not contain.
    /// `Compression::fast` deliberately: these bytes are re-compressed because
    /// they *cannot* be copied, so the trade is wire size against a clone's
    /// latency, and the entries this runs on are a small boundary fraction of any
    /// real request.
    fn whole_entry(kind: crate::object::GitObjectKind, body: &[u8]) -> Result<Vec<u8>> {
        use std::io::Write as _;
        let mut out = Vec::with_capacity(body.len() / 2 + 32);
        crate::pack_walk::encode_type_and_size(
            &mut out,
            crate::serve::resolved_type(kind),
            body.len() as u64,
        );
        let mut z = flate2::write::ZlibEncoder::new(out, flate2::Compression::fast());
        z.write_all(body)?;
        Ok(z.finish()?)
    }

    /// Take the absorb gate and hold it.
    ///
    /// The only way to observe "a read that arrived before the drain" as a
    /// **state** rather than as a race: with this held, the worker is parked at
    /// the top of its absorb, so the pack is provably durable, provably queued
    /// and provably not indexed. Its guard is in this module's tests, and nothing
    /// on a serving path takes it.
    #[cfg(test)]
    pub(crate) fn hold_absorb_gate(&self) -> std::sync::MutexGuard<'_, ()> {
        self.absorber.gate.lock().expect("absorb gate poisoned")
    }

    /// `pread` an extent out of the verbatim archive, into a fresh `Vec`.
    ///
    /// **The fallback and the owning path.** A caller that can hold a borrow
    /// wants [`Self::archive_snapshot`] plus [`Self::extent`] instead: this one
    /// costs a syscall, an allocation and a kernel→user copy, which is the shape
    /// [`crate::archive_map`] exists to get off the emit path.
    pub(crate) fn read_extent(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
        self.absorber.read_extent(offset, len)
    }

    /// One mapping of the verbatim archive, to be taken **once per operation**
    /// and sliced N times. See [`crate::archive_map::ArchiveMap::snapshot`].
    pub(crate) fn archive_snapshot(&self) -> Result<Arc<crate::archive_map::Mapped>> {
        self.absorber.map.snapshot()
    }

    /// **The one resolver**: an extent's bytes, borrowed from `snap` when it
    /// covers them and `pread` into a fresh buffer when it does not.
    ///
    /// `Cow::Borrowed` is the ordinary answer and the whole point — no syscall,
    /// no allocation, no copy, just a bounds-checked slice of the page cache.
    /// `Cow::Owned` happens for an extent past the mapped end, which means the
    /// blob file grew after the snapshot was taken (a push landed mid-clone).
    /// `Mapped::get` returning `None` is *"go and pread it"* and never *"there
    /// are no bytes"*, so the two arms return the **same bytes** and differ only
    /// in what they cost — which is the property
    /// `an_extent_past_the_mapping_falls_back_and_returns_the_same_bytes`
    /// asserts.
    pub(crate) fn extent<'m>(
        &self,
        snap: &'m crate::archive_map::Mapped,
        offset: u64,
        len: u64,
    ) -> Result<std::borrow::Cow<'m, [u8]>> {
        match snap.get(offset, len) {
            Some(b) => Ok(std::borrow::Cow::Borrowed(b)),
            None => Ok(std::borrow::Cow::Owned(self.read_extent(offset, len)?)),
        }
    }

    /// Record a durable pack as pending index work.
    ///
    /// The **bit going clear**, and the ack path's last act before it returns.
    pub(crate) fn queue(&self, pack_id: u64, extent: Extent) -> Result<()> {
        self.absorber.queue(pack_id, extent)
    }

    fn reach_bitmaps(&self) -> Result<Arc<Vec<ReachEntry>>> {
        self.absorber.reach_bitmaps()
    }

    /// [`Absorber::reach_bitmaps_with`], for the differential guard.
    pub(crate) fn reach_bitmaps_with(
        &self,
        policy: crate::reach::ReachPolicy,
        cache: bool,
    ) -> Result<Arc<Vec<ReachEntry>>> {
        self.absorber.reach_bitmaps_with(policy, cache)
    }

    /// The commit oid set, raw, shared. See [`Derived::commit_raw`].
    pub(crate) fn commit_oids_raw(
        &self,
    ) -> Result<Option<Arc<std::collections::HashSet<Vec<u8>>>>> {
        self.absorber.commit_oids_raw()
    }

    fn refold(&self) -> Result<()> {
        self.absorber.refold()
    }
}

impl<S: ObjectIndex> Absorber<S> {
    fn open(
        blobs: &Path,
        hash: GitHashKind,
        objects: ObjectReadStack<S>,
        exploded: ExplodedArchive,
    ) -> Result<Self> {
        let reader =
            File::open(blobs).with_context(|| format!("opening {} for reads", blobs.display()))?;
        Ok(Self {
            blobs: blobs.to_path_buf(),
            reader,
            map: crate::archive_map::ArchiveMap::new(blobs),
            hash,
            objects,
            exploded,
            derived: RwLock::new(Derived::default()),
            packs: Mutex::new(PackState::default()),
            gate: Mutex::new(()),
        })
    }

    /// Packs whose objects are not in the index yet — §13.12's bit, counted.
    fn unindexed_packs(&self) -> usize {
        self.packs.lock().expect("pack state lock").unabsorbed
    }

    /// The bit going clear: this pack is durable and its objects are not in.
    ///
    /// Skips a pack the drain already absorbed — the ack path queues *after* it
    /// has handed the extent to the channel, so the worker can legitimately be
    /// finished before this runs, and clearing a bit for a pack that is already
    /// indexed would leave it clear for ever.
    fn queue(&self, pack_id: u64, extent: Extent) -> Result<()> {
        let mut s = self
            .packs
            .lock()
            .map_err(|_| anyhow!("pack state poisoned"))?;
        s.note(pack_id, extent);
        Ok(())
    }

    /// **The bit, derived from the two durable facts, on open.**
    ///
    /// `journal` is every row this archive's journal holds, in append order.
    /// [`acked_packs`](crate::archive_write::acked_packs) is what turns that into
    /// the packs: its index *is* the pack ordinal — dense by construction, and
    /// re-derived the same way by every later open. For each one, the question is
    /// only *does this pack have any rows in the index*, which
    /// [`ObjectReadStack::extents_with_rows`] answers in one early-terminating
    /// scan rather than per pack.
    ///
    /// Returns the packs whose bit came up **clear**: durable bytes, no rows. The
    /// caller re-queues them on the account's channel, which is what makes a
    /// pack that was mid-absorb when the process died get absorbed by the next
    /// one instead of being durable-but-invisible for ever.
    ///
    /// The verbatim extents are also handed to [`Derived::packs`] — *all* of
    /// them, not only the unabsorbed ones. That is what lets [`BaseSource`]
    /// re-derive a delta base's content from the truth after a restart; without
    /// it a reopened store knows the rows of a pack it can no longer find the
    /// bytes of.
    ///
    /// # The one case where "no rows" does not mean "never absorbed" — and the
    /// third durable fact that now tells them apart
    ///
    /// [`GitOps::gc`] is the only thing that removes rows. A pack **all** of
    /// whose objects it finds dead therefore ends with a journal extent and no
    /// rows, which is byte for byte the state a pack that was in flight when the
    /// machine died leaves behind. This diff cannot tell them apart, and it must
    /// not guess: re-queueing the first resurrects every dead object, and *not*
    /// re-queueing the second loses a pack whose bytes were acked.
    ///
    /// So `gc` says which it is, durably, before it drops a row:
    /// [`retire_packs`](crate::archive_write::retire_packs) appends a tombstone
    /// naming the retired pack's offset, and a pack named by one is never
    /// re-queued here. The bit is still derived from durable facts only — there
    /// are now three of them (the journal's pack rows, the journal's tombstones,
    /// the index's rows) and none of them is a cache of another.
    ///
    /// A **partly** dead pack is not tombstoned and still has rows, so it takes
    /// the `absorbed` branch exactly as it always did.
    ///
    /// A tombstoned pack that *does* still have rows is a `gc` that was killed
    /// between the tombstone and the drop. It is treated as absorbed, which is
    /// what its rows say: nothing is lost, and the next `gc` finishes the job.
    ///
    /// Its extent still goes to [`Derived::packs`] either way. The blob file is
    /// append-only and `gc` truncates nothing, so a retired pack's bytes are
    /// still exactly where the journal says they are — and a live object may
    /// legitimately be a delta against a base that is itself unreachable, so
    /// dropping the extent from that list would break the resolve rather than
    /// tidy it.
    ///
    /// # The second durable fact: §14's exploded table
    ///
    /// A pack counts as absorbed only if the exploded table is **whole**, which
    /// is `exploded.rows() >= objects.len()` — one row per object, because §14's
    /// table is eager and skips nothing. If it is short, the table has been
    /// dropped (or a process died between the two commits) and **no** pack counts
    /// as absorbed: every one is re-queued and re-exploded from the verbatim
    /// truth. That is the whole of "droppable": deleting the file is a supported
    /// operation whose only cost is one re-ingest, and §13.12's rule holds
    /// unchanged over it — absent means fall back, never means wrong.
    ///
    /// `>=` and not `==` on purpose. A resolve that failed part way leaves rows
    /// for the objects it got to, and a `gc` retires both tables together, so the
    /// table can legitimately hold *more* than the index names; only being
    /// **short** is evidence of a gap.
    fn adopt_journal(&self, journal: &[Extent]) -> Result<Vec<PendingPack>> {
        if journal.is_empty() {
            return Ok(Vec::new());
        }
        // The packs, in ordinal order, and the offsets a `gc` retired. A
        // tombstone is a row and not a pack, so it holds no ordinal.
        let packs = crate::archive_write::acked_packs(journal);
        let retired = crate::archive_write::retired_offsets(journal);
        if packs.is_empty() {
            return Ok(Vec::new());
        }
        // Under `Graph` the table is short **by design** — commits and trees
        // only — so the row-count comparison would call every store dropped and
        // re-queue every pack on every open. It is weakened to "not empty",
        // which still catches the case this rule exists for (the file deleted),
        // and is stated rather than hidden: making it exact needs a durable
        // commit-and-tree count, which the index does not publish today.
        let exploded_is_whole = match self.exploded.policy() {
            crate::exploded_arrow::ExplodePolicy::Graph => {
                self.objects.len() == 0 || self.exploded.rows()? > 0
            }
            _ => self.exploded.rows()? >= self.objects.len() as u64,
        };
        let has_rows = if exploded_is_whole {
            self.objects.extents_with_rows(&packs)?
        } else {
            vec![false; packs.len()]
        };
        {
            let mut d = self
                .derived
                .write()
                .map_err(|_| anyhow!("derived poisoned"))?;
            d.packs = packs.clone();
        }
        let mut s = self
            .packs
            .lock()
            .map_err(|_| anyhow!("pack state poisoned"))?;
        let mut requeue = Vec::new();
        for (i, (&extent, &absorbed)) in packs.iter().zip(&has_rows).enumerate() {
            let pack_id = i as u64;
            if absorbed {
                s.mark_absorbed(pack_id, extent);
            } else if retired.contains(&extent.0) {
                // A `gc` dropped every row this pack had and said so durably
                // *before* dropping them. There is no index work owed for it,
                // ever: re-queueing it is precisely how its dead objects would
                // come back. The bit goes **up** rather than being left clear,
                // because a clear bit is what makes a read fall back to
                // absorbing — which is the same resurrection by another door.
                s.mark_absorbed(pack_id, extent);
            } else {
                s.note(pack_id, extent);
                requeue.push(PendingPack {
                    pack_id,
                    offset: extent.0,
                    len: extent.1,
                });
            }
        }
        Ok(requeue)
    }

    /// **The indexer's half of the split**, and not one of the twelve.
    ///
    /// Reads each pending pack's verbatim bytes back out of the archive,
    /// resolves them to oids ([`crate::resolve`]), appends the rows to the
    /// `objects` table and re-folds the derived tables. This is what §13.9-12
    /// puts behind the channel; it runs after the ack, never before one.
    ///
    /// Two callers, one function: the account indexer's worker (through
    /// [`ObjectAbsorb`]) and a read that arrived before that worker did. The
    /// second is §13.12's fallback in its literal form — *slower, never wrong*.
    fn absorb_pending(&self) -> Result<usize> {
        // Nothing to do, and — this is the point — **no gate taken**: the common
        // case is a store whose drain has kept up, and it must not queue behind
        // one that is running.
        if self.unindexed_packs() == 0 {
            return Ok(0);
        }
        let _gate = self
            .gate
            .lock()
            .map_err(|_| anyhow!("absorb gate poisoned"))?;
        let jobs: Vec<PendingPack> = {
            let s = self
                .packs
                .lock()
                .map_err(|_| anyhow!("pack state poisoned"))?;
            s.pending()
        };
        let mut absorbed = 0usize;
        for job in &jobs {
            self.absorb_gated(job)?;
            absorbed += 1;
        }
        Ok(absorbed)
    }

    /// One pack, **with [`Absorber::gate`] already held**.
    ///
    /// The order at the end is the correctness argument: the rows are in the
    /// index *before* the bit is cleared. A read that finds the bit still clear
    /// blocks on the gate and then finds the rows; a read that finds it set finds
    /// the rows too. There is no interleaving in which a durable object answers
    /// "absent" — which during negotiation would make a client withhold objects
    /// and lose data.
    fn absorb_gated(&self, job: &PendingPack) -> Result<()> {
        {
            let s = self
                .packs
                .lock()
                .map_err(|_| anyhow!("pack state poisoned"))?;
            if s.is_absorbed(job.pack_id) {
                // Already in, by whoever reached it first. Re-resolving is now
                // *correct* — the exploded table is keyed by oid and the graph is
                // folded from it rather than pushed at, so a second absorb of one
                // pack changes nothing (that idempotence is what makes dropping
                // the table and re-queueing every pack a safe operation). It is
                // still a whole pack re-read, re-inflated and re-hashed for
                // nothing, so it is skipped.
                return Ok(());
            }
        }
        self.absorb_one(job).map_err(|e| {
            // The entry stays pending by construction — it was never removed —
            // so the bytes are durable, the pack is still un-indexed, and reads
            // keep falling back rather than answering absent.
            e.context(format!(
                "absorbing pack {} at ({}, {}) — its bytes are durable and it stays \
                 un-indexed; reads will keep falling back rather than answer absent",
                job.pack_id, job.offset, job.len
            ))
        })?;
        let mut s = self
            .packs
            .lock()
            .map_err(|_| anyhow!("pack state poisoned"))?;
        s.mark_absorbed(job.pack_id, (job.offset, job.len));
        Ok(())
    }

    /// One pack: verbatim bytes in, `objects` rows **and** §14's exploded rows
    /// out, from **one** walk and **one** resolve.
    ///
    /// The exploded rows are not a second pass and not a second resolver (LAW 5):
    /// [`crate::resolve::resolve_walked`] hands every object it produces to the
    /// sink as it produces it, including the blob payloads it is about to drop,
    /// so the side table falls out of the pass that was already being made.
    ///
    /// Ordering is the correctness argument, and it is the same one the drain
    /// makes one level up: **the exploded rows are committed before the fold that
    /// reads them**, and the fold runs before this returns, so there is no
    /// interleaving in which the graph is folded over a table that does not yet
    /// hold this pack's commits.
    fn absorb_one(&self, job: &PendingPack) -> Result<()> {
        let bytes = self.read_extent(job.offset, job.len)?;
        let walked = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
        let rows = crate::resolve::resolve_walked(
            &bytes,
            &walked,
            self.hash,
            job.offset,
            self,
            &self.exploded,
        )?;
        // Down before anything folds over it.
        self.exploded.flush()?;

        let entries: Vec<crate::index_layout::IndexEntry> =
            rows.iter().map(|r| r.index_entry()).collect();
        self.objects.append(&entries)?;

        {
            let mut d = self
                .derived
                .write()
                .map_err(|_| anyhow!("derived poisoned"))?;
            // A reopen already listed every acked extent (`adopt_journal`), so
            // this is the first absorb of a pack pushed in *this* process — or a
            // re-queued one that is already listed. `find` is what reads this
            // list, so a duplicate would not be wrong, only unbounded.
            if !d.packs.contains(&(job.offset, job.len)) {
                d.packs.push((job.offset, job.len));
            }
        }
        // The graph and the tree payloads are folded from the exploded table, not
        // pushed at from here. That is what makes a re-absorb idempotent — which
        // it has to be, because dropping the table re-queues every pack.
        self.refold()?;
        let _ = self.objects.maybe_rebuild()?;
        Ok(())
    }

    /// Recompute everything derived from the object set: the commit graph, the
    /// tree payloads, the ordinal space, the generation numbers, and (lazily) the
    /// bitmaps.
    ///
    /// **The graph is folded from §14's exploded table, not accumulated.** That
    /// is the whole fix for the clean-reopen hole: commit and tree payloads are
    /// on disk, so this call produces the same graph after a restart as before
    /// one, and it is called at the end of [`GitStore::open_with_arms`] for
    /// exactly that reason. It also makes a re-absorb idempotent — the old
    /// accumulate-into-a-`Vec` shape doubled every commit if a pack was absorbed
    /// twice, which is why dropping the table and re-queueing everything was not
    /// a safe operation before this.
    ///
    /// **Generations are stored and must be recomputed in any fold** (§13,
    /// decided). A generation number is `1 + max(parents)`, so it is invalidated
    /// the moment history's shape changes — which is exactly what absorbing a
    /// pack does. [`assign_generations`] is the one writer of that number and it
    /// is called here, over the whole graph, rather than incrementally: an
    /// incremental update would have to know which descendants moved, and being
    /// wrong about that produces an ancestry cutoff that skips real ancestors.
    ///
    /// # Cost
    ///
    /// One range scan of the exploded table's commits and one of its trees, per
    /// absorbed pack. It is the same order as the `tail_oids` scan below, which
    /// this function already made on every absorb, and it reads **no blob
    /// payload** — that is what the kind index in
    /// [`crate::exploded`] is for.
    ///
    /// ## And that order is the whole repository, on every push
    ///
    /// Said plainly, because "the same order as the scan below" reads like a
    /// reassurance and is not one: **nothing here is proportional to the push.**
    /// Every call rebuilds every tree payload, every commit node, every
    /// generation number, a `HashMap<String, u32>` over every oid in the
    /// repository and a sorted `Vec<String>` of the same — for a push of one
    /// commit exactly as for a push of ten thousand.
    ///
    /// MEASURED on oden 2026-08-11, release, quiet box (`some avg10` 0.00). The
    /// **same** 100-commit increment, pushed onto the same history seeded to
    /// four different depths, server CPU for the push:
    ///
    /// | objects already in the repository | CPU |
    /// |---:|---:|
    /// | 9 614 | 0.12 s |
    /// | 35 350 | 0.23 s |
    /// | 103 988 | 0.45 s |
    /// | 206 925 | 0.77 s |
    ///
    /// That is a straight line: **≈88 ms fixed, plus ≈3.3 µs per object already
    /// stored**, and the fit predicts the four points to within 20 ms. It is
    /// also why a push is slower than the ack path suggests — §13.9 puts the
    /// index behind a channel, but `put_refs` checks every ref target with
    /// `has`, `has` calls `lookup_one`, and `lookup_one` absorbs pending packs
    /// first. The drain is off the *ack* and squarely on the *push*.
    ///
    /// This is not fixed here and the numbers are recorded so the next attempt
    /// starts from them rather than from a guess. The obvious lever — fold
    /// incrementally — is the one the paragraph above rules out for
    /// generations, so it needs its own argument and its own gate.
    fn refold(&self) -> Result<()> {
        // Read out of redb *before* taking the write lock: the fold is the long
        // part and nothing that reads `derived` should queue behind it.
        let commits = self.exploded.of_kind(GitObjectKind::Commit)?;
        let trees = self.exploded.of_kind(GitObjectKind::Tree)?;

        let mut d = self
            .derived
            .write()
            .map_err(|_| anyhow!("derived poisoned"))?;
        d.trees = trees
            .into_iter()
            .map(|(oid, payload)| (hex::encode(oid), payload))
            .collect();
        let graph: Vec<CommitNode> = commits
            .iter()
            .map(|(oid, payload)| {
                let h = crate::object::parse_commit(payload);
                CommitNode {
                    oid: hex::encode(oid),
                    parents: h.parents,
                    tree: h.tree,
                    committer_time: h.committer_time,
                    generation: 0,
                }
            })
            .collect();
        d.graph = assign_generations(graph);
        // The same commit set `select` refuses a `want` against, in the raw form
        // it tests. Folded once here instead of once per request; see the field.
        // A row that does not decode makes the whole set `None`, which is the
        // decline `select` used to reach by returning early out of its own loop.
        d.commit_raw = d
            .graph
            .iter()
            .map(|n| hex::decode(&n.oid).ok())
            .collect::<Option<std::collections::HashSet<Vec<u8>>>>()
            .map(Arc::new);

        // Ordinals: oid-lexicographic rank over every object in the index. Same
        // ordering rule the Arrow projection uses, computed here so the bitmaps
        // and the ordinals they address are produced by one call.
        //
        // **Sorted as bytes, then hex-encoded — not hex-encoded and then
        // sorted.** The two orders are the same order (hex is monotone over
        // fixed-width oids, which is why this is not a behaviour change), and
        // taking it in this direction is what lets the raw and the hex form come
        // out of ONE sort: `d.oids_raw` is the concatenation of the same
        // sequence `d.oids` spells in hex, so ordinal `o` names the same object
        // in both by construction rather than by two agreeing computations
        // (LAW 5).
        let mut raw = self.tail_oids()?;
        raw.sort_unstable();
        let mut oids_raw = Vec::with_capacity(raw.iter().map(Vec::len).sum());
        for oid in &raw {
            oids_raw.extend_from_slice(oid);
        }
        let oids: Vec<String> = raw.iter().map(hex::encode).collect();
        d.ordinal = oids
            .iter()
            .enumerate()
            .map(|(i, o)| (o.clone(), i as u32))
            .collect();
        d.oids = oids;
        d.oids_raw = oids_raw;
        // The bitmaps are over the ordinal space that just changed.
        d.reach = Arc::default();
        Ok(())
    }

    /// Every oid in the index, from the tail — which redb keeps in oid order, so
    /// this is already the ordinal order.
    fn tail_oids(&self) -> Result<Vec<Vec<u8>>> {
        self.objects.oids_in_order()
    }

    /// The bitmaps, built on first use after a fold and cached until the next.
    fn reach_bitmaps(&self) -> Result<Arc<Vec<ReachEntry>>> {
        self.reach_bitmaps_with(live_reach_policy(), true)
    }

    /// [`Absorber::reach_bitmaps`] with the policy **named** and the cache
    /// optional.
    ///
    /// `cache` is false for the differential guard, which builds the table at two
    /// very different caps over one store and requires the two to answer
    /// identically; letting either poison `d.reach` would make the second arm
    /// read the first one's table and the comparison would be of a thing against
    /// itself.
    fn reach_bitmaps_with(
        &self,
        policy: crate::reach::ReachPolicy,
        cache: bool,
    ) -> Result<Arc<Vec<ReachEntry>>> {
        {
            let d = self
                .derived
                .read()
                .map_err(|_| anyhow!("derived poisoned"))?;
            if cache && !d.reach.is_empty() {
                // A refcount bump. This was `d.reach.clone()` — one deep copy of
                // every commit's roaring bitmap, per call, twice per request.
                return Ok(Arc::clone(&d.reach));
            }
            if d.graph.is_empty() {
                return Ok(Arc::default());
            }
        }
        let mut d = self
            .derived
            .write()
            .map_err(|_| anyhow!("derived poisoned"))?;
        let facts = crate::reach::ObjectFacts {
            ordinal: &d.ordinal,
            trees: &d.trees,
            oid_len: self.hash.oid_len(),
        };
        let built = Arc::new(crate::reach::build_reach(&d.graph, &facts, policy));
        if cache {
            // `Arc::new` once, then a refcount bump into the cache — the second
            // full copy this line used to make (`d.reach = built.clone()`) is
            // gone too.
            d.reach = Arc::clone(&built);
        }
        Ok(built)
    }

    /// The commit oid set, raw, as folded. See [`Derived::commit_raw`].
    fn commit_oids_raw(&self) -> Result<Option<Arc<std::collections::HashSet<Vec<u8>>>>> {
        Ok(self
            .derived
            .read()
            .map_err(|_| anyhow!("derived poisoned"))?
            .commit_raw
            .clone())
    }

    // `ordinal_space()` lived here: `d.oids.clone()`, a fresh `Vec<String>` of
    // every oid in the store, handed to `reachable_oids` twice per request. It
    // is gone rather than left unused — its one caller now borrows `d.oids`
    // under the read guard it was already going to take.

    /// `pread` an extent out of the verbatim archive, **into uninitialised
    /// capacity** — no zero-fill before the read.
    ///
    /// This used to be `vec![0u8; len]` + `read_exact_at`, and under emit
    /// concurrency the zeroing was not a detail: profiled on oden 2026-08-12
    /// (gunnar serving 32 concurrent clones, perf on the server pid), ~44 % of
    /// serve CPU was memory zeroing/copying — `memset` 18.1 % plus kernel
    /// page-zeroing 7.8 % — against ~12 % for one clone, because 32 threads
    /// churning pack-scale buffers recycle mimalloc freelist blocks that must
    /// each be memset before `pread` immediately overwrites every byte.
    ///
    /// Soundness: no reference to uninitialised memory is ever formed. The
    /// bytes are written through a raw pointer into the `Vec`'s spare
    /// capacity, `EINTR` retries, a short read refuses, and `set_len` runs
    /// only after every one of `len` bytes has been written.
    fn read_extent(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
        use std::os::fd::AsRawFd;
        let want = usize::try_from(len).context("extent length overflows usize")?;
        let mut buf: Vec<u8> = Vec::with_capacity(want);
        let fd = self.reader.as_raw_fd();
        let mut filled = 0usize;
        while filled < want {
            let n = unsafe {
                libc::pread(
                    fd,
                    buf.as_mut_ptr().add(filled).cast(),
                    want - filled,
                    i64::try_from(offset + filled as u64).context("extent offset overflows off_t")?,
                )
            };
            if n < 0 {
                let err = std::io::Error::last_os_error();
                if err.kind() == std::io::ErrorKind::Interrupted {
                    continue;
                }
                return Err(err).with_context(|| {
                    format!("reading ({offset}, {len}) out of {}", self.blobs.display())
                });
            }
            if n == 0 {
                bail!(
                    "short read at ({offset}, {len}) out of {}: {filled} of {want} bytes before \
                     EOF — the archive is truncated relative to its index",
                    self.blobs.display()
                );
            }
            filled += n as usize;
        }
        // Every byte of `want` is now initialised.
        unsafe { buf.set_len(want) };
        Ok(buf)
    }
}

/// **The drain's object-level ingress**, and the reason a push ends in object
/// rows without anybody reading first.
///
/// The account indexer holds this as an `Arc<dyn ObjectAbsorb>` and calls it once
/// per drained job, on its own thread, after the ack. It is the *same* absorb a
/// falling-back read takes, gate and all (LAW 5): there is no background copy of
/// this logic to drift.
impl<S: ObjectIndex> ObjectAbsorb for Absorber<S> {
    fn absorb(&self, job: IndexJob) -> Result<()> {
        let _gate = self
            .gate
            .lock()
            .map_err(|_| anyhow!("absorb gate poisoned"))?;
        self.absorb_gated(&PendingPack::from_job(job))
    }
}

impl<S: ObjectIndex + 'static> GitStore<S> {
    /// The push path, for the two `put` halves.
    pub(crate) fn push_path(&self) -> &PushPath {
        &self.push
    }

    pub(crate) fn account(&self) -> &str {
        &self.account
    }

    pub(crate) fn ref_log(&self) -> &RefLog {
        &self.refs
    }

    pub(crate) fn ref_gate(&self) -> &Mutex<()> {
        &self.ref_gate
    }

    /// The closure check's index half: a `REF_DELTA`'s base has to exist —
    /// **in the store, or in this very pack**.
    ///
    /// This is the one part of receive-pack that reads `objects.oid` (§13's
    /// table), and on the path a real push takes it reads nothing else.
    ///
    /// # The second half of that sentence used to be missing, and it refused
    /// packs git itself writes
    ///
    /// [`crate::pack_walk::PackWalk::closure`] cannot do better on its own:
    /// a walk reports what the bytes say, and the bytes of a `REF_DELTA` name a
    /// base by **oid**, which is a fact about resolved content that no walk can
    /// know. So every ref base came back on `external_refs` and every one of
    /// them was required to be in the store already.
    ///
    /// A `REF_DELTA` naming a base *inside the same pack* is ordinary and git
    /// writes it constantly: `git index-pack --fix-thin` completes a pushed thin
    /// pack by **appending the base object to the pack** and leaving the delta
    /// naming it by oid. Every such pack was refused with *"this pack deltas
    /// against X, which this repository does not have"*, on a store that was
    /// being handed a perfectly ordinary, self-contained packfile — the base was
    /// entry 1 of 25 in the pack's own `.idx` (`gunnar.multi_pack_serve`,
    /// 2026-08-14). The same missing check is `S-003` / gitoxide#2882 in
    /// `LookupRefDeltaObjectsIter`; it is the same defect wearing the receiving
    /// hat instead of the sending one.
    ///
    /// # What it costs, and where
    ///
    /// **Nothing on any push that was going to succeed before.** Every base
    /// found in the store short-circuits exactly as it did, and a pack with no
    /// ref-delta at all never gets past the first loop.
    ///
    /// The pack is resolved only when the alternative is *refusing it*, and then
    /// the resolve is the cheapest honest answer available: an entry's oid is
    /// knowable only by applying its delta chain, which is precisely what
    /// [`crate::resolve::resolve_walked`] does — the same resolver the absorb is
    /// about to run over the same pack anyway, reused rather than twinned
    /// (LAW 5). A push that is genuinely thin against a base nobody has still
    /// fails, one resolve later, and still names the oid.
    pub(crate) fn external_bases_exist(&self, bytes: &[u8], w: &PackWalk) -> Result<()> {
        let c = w.closure();
        if !c.broken_offsets.is_empty() {
            bail!(
                "this pack is corrupt: {} delta base offset(s) do not land on an entry — first {}",
                c.broken_offsets.len(),
                c.broken_offsets[0]
            );
        }
        // The fast path, unchanged: a base the store already holds is settled
        // without reading one byte of the pack.
        let mut unheld: Vec<&[u8]> = Vec::new();
        for oid in &c.external_refs {
            if self.lookup_one(oid)?.is_none() {
                unheld.push(oid.as_slice());
            }
        }
        if unheld.is_empty() {
            return Ok(());
        }

        // Only now, and only because the alternative is refusing the push: does
        // the pack supply these itself? `resolve_walked` answers with the
        // store behind it, so a base that really is external is still found the
        // cheap way and only a genuinely absent one fails.
        let resolved = crate::resolve::resolve_walked(
            bytes,
            w,
            self.hash_kind(),
            0,
            &*self.absorber,
            &crate::exploded::NoSink,
        )
        .with_context(|| {
            format!(
                "this pack deltas against {} object(s) this repository does not have, so it was \
                 resolved to find out whether the pack carries them itself — first {}",
                unheld.len(),
                hex::encode(unheld[0])
            )
        })?;
        let in_pack: std::collections::HashSet<&[u8]> =
            resolved.iter().map(|r| r.oid.as_slice()).collect();
        for oid in unheld {
            if !in_pack.contains(oid) {
                bail!(
                    "this pack deltas against {}, which this repository does not have and which \
                     the pack does not carry either — the push is refused rather than stored with \
                     a dangling base",
                    hex::encode(oid)
                );
            }
        }
        Ok(())
    }

    /// One oid, on the **serial** path. See [`lookup_path`].
    pub(crate) fn lookup_one(&self, oid: Oid<'_>) -> Result<Option<crate::index_layout::IndexRow>> {
        if self.unindexed_packs() > 0 {
            self.absorb_pending()?;
        }
        Ok(self.absorber.objects.lookup(oid))
    }

    /// The current ref namespace, as the log folds to it.
    pub(crate) fn ref_state(&self) -> Result<BTreeMap<String, crate::refs::RefState>> {
        self.refs.current()
    }

    /// Reachability over the graph, as roaring bitmaps, in **this store's**
    /// ordinal space.
    ///
    /// Returns `(want ∪ closure) − (have ∪ closure)` as oid hex. Every object —
    /// commit, tree and blob — is in the bitmaps, because
    /// [`crate::reach::build_reach`] walks the trees; that is why this can answer
    /// with objects rather than only with commits.
    ///
    /// **Hex is for the maintenance path, not the serving one.** The answer is
    /// computed in raw bytes by [`GitStore::reachable_raw_with`] and encoded
    /// here, at one `String` per object; every caller that serves a request asks
    /// for the raw form instead. See [`Derived::oids_raw`] for what that cost.
    pub(crate) fn reachable_oids(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> Result<Vec<String>> {
        Ok(self
            .reachable_raw_with(want, have, live_reach_policy(), true)?
            .iter()
            .map(hex::encode)
            .collect())
    }

    /// [`GitStore::reachable_oids`] with the bitmap policy **named** and the
    /// cache optional.
    ///
    /// One body, not two (LAW 5): both are [`GitStore::reachable_raw_with`], so
    /// the guarded path and the shipped path are the same code.
    #[cfg(test)]
    pub(crate) fn reachable_oids_with(
        &self,
        want: &[Oid<'_>],
        have: &[Oid<'_>],
        policy: crate::reach::ReachPolicy,
        cache: bool,
    ) -> Result<Vec<String>> {
        Ok(self
            .reachable_raw_with(want, have, policy, cache)?
            .iter()
            .map(hex::encode)
            .collect())
    }

    /// `(want ∪ closure) − (have ∪ closure)`, in **raw oid bytes**, in one
    /// allocation.
    ///
    /// The answer every serving caller wants and the one the bitmaps naturally
    /// produce: an ordinal indexes straight into [`Derived::oids_raw`], so
    /// nothing is encoded, decoded or allocated per object on the way out.
    pub(crate) fn reachable_raw(
        &self,
        want: &[Oid<'_>],
        have: &[Oid<'_>],
    ) -> Result<git_storage_trait::OidList> {
        self.reachable_raw_with(want, have, live_reach_policy(), true)
    }

    /// [`GitStore::reachable_raw`] with the bitmap policy **named** and the
    /// cache optional.
    ///
    /// It exists because the answer is defined to be **independent of the
    /// policy** — [`crate::reach::accumulate`] covers whatever the table does
    /// not — and a property like that can only be tested by driving the policy.
    /// A test that could not reach it would be asserting that one code path
    /// equals itself. `cache: false` keeps one arm's table from being read by
    /// the next, which is the same trap in a different hat.
    pub(crate) fn reachable_raw_with(
        &self,
        want: &[Oid<'_>],
        have: &[Oid<'_>],
        policy: crate::reach::ReachPolicy,
        cache: bool,
    ) -> Result<git_storage_trait::OidList> {
        if self.unindexed_packs() > 0 {
            self.absorb_pending()?;
        }
        // `reach_bitmaps` may take the derived WRITE lock to build, so it runs
        // to completion before the read guard below is taken. Getting that order
        // wrong is a self-deadlock, not a slow path.
        let bitmaps = self.reach_bitmaps_with(policy, cache)?;
        let by_commit: HashMap<&str, &roaring::RoaringBitmap> = bitmaps
            .iter()
            .map(|e| (e.commit.as_str(), &e.bitmap))
            .collect();
        // **One read guard, held across the whole answer, and nothing cloned out
        // of it.** This used to be `ordinal_space()?` (a `Vec<String>` of every
        // oid in the store) plus `.ordinal.clone()` (a `HashMap<String, u32>` of
        // the same, again) — ~2 × 12 455 `String` allocations per call and two
        // calls per request, for two tables that cannot change between folds.
        // Borrowing them is not a lock held longer than the work: the guard is a
        // reader, folds take the writer, and a fold that lands mid-answer would
        // invalidate the ordinals this loop is translating.
        let d = self
            .absorber
            .derived
            .read()
            .map_err(|_| anyhow!("derived poisoned"))?;
        let space = &d.oids_raw;
        let ordinal = &d.ordinal;
        let oid_len = self.hash.oid_len();

        // The graph by oid, for [`crate::reach::accumulate`]'s commit walk. Built
        // per call and not cached: it is one `HashMap` of borrows over rows the
        // read guard is already holding, and caching it would be a third
        // derivation to keep in step with a fold.
        let graph: HashMap<&str, &CommitNode> =
            d.graph.iter().map(|n| (n.oid.as_str(), n)).collect();
        let facts = crate::reach::ObjectFacts {
            ordinal,
            trees: &d.trees,
            oid_len,
        };

        let mut union = roaring::RoaringBitmap::new();
        for oid in want {
            let hex = hex::encode(oid);
            match by_commit.get(hex.as_str()) {
                // A commit with a bitmap: its whole closure, in one OR.
                Some(bm) => union |= *bm,
                // **No bitmap. This is the case that used to be a wrong answer.**
                //
                // It contributed the object itself and nothing else — see
                // `crate::reach::accumulate`'s header for what that cost. Now a
                // commit the graph holds is WALKED, bounded by the first
                // bitmapped commit behind it, and only a `want` the store does
                // not hold at all is still refused.
                None => {
                    if graph.contains_key(hex.as_str()) {
                        crate::reach::accumulate(
                            hex.as_str(),
                            &by_commit,
                            &graph,
                            &facts,
                            &mut union,
                        );
                    } else if let Some(&o) = ordinal.get(hex.as_str()) {
                        // Not a commit at all — a tag or a blob asked for
                        // directly. Just it, as before.
                        union.insert(o);
                    } else {
                        bail!(
                            "want {hex} is not in this repository — a negotiation must not be \
                             answered from a partial set"
                        );
                    }
                }
            }
        }
        let mut had = roaring::RoaringBitmap::new();
        for oid in have {
            let hex = hex::encode(oid);
            if let Some(bm) = by_commit.get(hex.as_str()) {
                had |= *bm;
            } else if graph.contains_key(hex.as_str()) {
                // The same walk on the exclude side, and it must be the same
                // walk: an under-counted `have` over-sends (harmless, wasteful)
                // but an under-counted `want` under-sends, and answering the two
                // sides by different rules is how a `want − have` stops being a
                // subtraction of like for like.
                crate::reach::accumulate(hex.as_str(), &by_commit, &graph, &facts, &mut had);
            } else if let Some(&o) = ordinal.get(hex.as_str()) {
                had.insert(o);
            }
            // An unknown `have` is normal: the client may have objects we do not.
            // It contributes nothing, which is the safe direction — we send more,
            // never less.
        }

        let delta = union - had;
        // One allocation for the whole answer, sized before the loop. The
        // ordinal space is a flat buffer, so each object is a `copy_from_slice`
        // of `oid_len` bytes into it and nothing else.
        let mut out = git_storage_trait::OidList::with_capacity(delta.len() as usize, oid_len);
        for o in delta {
            let at = o as usize * oid_len;
            let raw = space
                .get(at..at + oid_len)
                .ok_or_else(|| anyhow!("ordinal {o} is outside this store's ordinal space"))?;
            out.push(raw)?;
        }
        Ok(out)
    }

    /// Every object reachable from every ref: what a GC must keep.
    pub(crate) fn live_set(&self) -> Result<std::collections::HashSet<String>> {
        let refs = self.ref_state()?;
        let tips: Vec<Vec<u8>> = refs
            .values()
            .filter_map(|s| s.target.as_deref())
            .filter_map(|t| hex::decode(t).ok())
            .collect();
        if tips.is_empty() {
            bail!(
                "this repository has no ref pointing at an object, so every object in it would be \
                 dead. A GC that would delete everything is refused: name a ref first"
            );
        }
        let borrowed: Vec<Oid<'_>> = tips.iter().map(|t| t.as_slice()).collect();
        let mut live: std::collections::HashSet<String> =
            self.reachable_oids(&borrowed, &[])?.into_iter().collect();
        // The peeled targets of annotated tags are reachable too, and a tag
        // object is not a commit so it has no bitmap of its own.
        for s in refs.values() {
            if let Some(p) = &s.peeled {
                live.insert(p.clone());
            }
            if let Some(t) = &s.target {
                live.insert(t.clone());
            }
        }
        Ok(live)
    }

    /// Drop every row the live set does not name. The index is append-only for
    /// everything except this.
    ///
    /// **Both tables, in one call.** §14's exploded table is retired with the
    /// `objects` table because its row count against that one's is what says
    /// whether it is whole ([`Absorber::adopt_journal`]): dropping from one and
    /// not the other would leave every later open re-exploding the entire
    /// repository. The count returned is still the index's — that is what a
    /// [`GcReport`] means by a dropped row.
    pub(crate) fn drop_dead_rows(&self, live: &std::collections::HashSet<String>) -> Result<u64> {
        let dropped = self
            .absorber
            .objects
            .retain(&|oid: &[u8]| live.contains(&hex::encode(oid)))?;
        self.absorber
            .exploded
            .retain(&|oid: &[u8]| live.contains(&hex::encode(oid)))?;
        // **Everything folded from those two tables, refolded.** The commit
        // graph, the tree payloads, the ordinal space and the bitmaps are all
        // derivations of the rows that just went, and a derivation that still
        // names a dropped oid is *wrong* rather than merely stale — a `reachable`
        // over a graph holding dead commits selects objects the index can no
        // longer serve. This is the same [`Absorber::refold`] every absorb and
        // every open already calls, over the table this call just changed;
        // nothing about how the graph is rebuilt is touched here.
        //
        // The bitmaps go with it: they address **ordinals**, and dropping rows
        // renumbers the ordinal space, so `refold` clears them and the next
        // `reach_bitmaps()` builds them over the new numbering. Carrying them
        // across would be carrying an index into a different array.
        self.absorber.refold()?;
        Ok(dropped)
    }

    /// **The journal's half of a GC: retire every pack whose objects are now all
    /// dead, before a single row is dropped.**
    ///
    /// Returns the offsets retired.
    ///
    /// # Why this exists
    ///
    /// §13.12's `indexed` bit is derived on open as *extent in the journal, rows
    /// not in the index* ([`Absorber::adopt_journal`]). Dropping every row of a
    /// pack produces that state exactly, so without this call the next open
    /// re-queues the pack and re-absorbs the objects a GC just decided were dead.
    /// A **partly** dead pack keeps rows and was never affected; this is only
    /// about the all-dead case.
    ///
    /// # What it does not do
    ///
    /// It does not truncate `objects.pack`. The blob file is append-only and a
    /// pack in the middle of it cannot be cut out without moving every extent
    /// after it — which would invalidate every index row in the archive. The dead
    /// pack's bytes stay; what changes is that nothing will ever index them
    /// again. Reclaiming those bytes is a rewrite of the blob file, and it is not
    /// this function.
    ///
    /// An arm with no journal ([`FastWriter`](crate::archive_write::FastWriter))
    /// has nothing to retire and nothing that re-queues, so it returns empty.
    ///
    /// # Cost
    ///
    /// One [`ObjectReadStack::extents_with_rows`] scan plus one batch lookup of
    /// the live set. Both are already the shape a GC pays elsewhere, and both run
    /// once per `gc()`, not per pack.
    pub(crate) fn retire_dead_packs(
        &self,
        live: &std::collections::HashSet<String>,
    ) -> Result<Vec<u64>> {
        let Some(journal) = self.arms.writer.journal(&self.blobs) else {
            return Ok(Vec::new());
        };
        if !journal.exists() {
            return Ok(Vec::new());
        }
        let rows = crate::archive_write::read_journal(&journal)?;
        let packs = crate::archive_write::acked_packs(&rows);
        let already = crate::archive_write::retired_offsets(&rows);
        if packs.is_empty() {
            return Ok(Vec::new());
        }

        // Which packs hold any row at all right now. A pack with none is either
        // already retired or was never absorbed — neither is this call's
        // business, and calling the second one dead would lose it.
        let occupied = self.absorber.objects.extents_with_rows(&packs)?;

        // Which packs hold a row that survives. Placed by binary search over the
        // pack starts, the same way the scan above places a row.
        let mut order: Vec<usize> = (0..packs.len()).filter(|&i| packs[i].1 > 0).collect();
        order.sort_unstable_by_key(|&i| packs[i].0);
        let starts: Vec<u64> = order.iter().map(|&i| packs[i].0).collect();
        let live_oids: Vec<Vec<u8>> = live.iter().filter_map(|h| hex::decode(h).ok()).collect();
        let borrowed: Vec<&[u8]> = live_oids.iter().map(|o| o.as_slice()).collect();
        let mut has_live = vec![false; packs.len()];
        for (offset, _) in self
            .absorber
            .objects
            .extents_batch(&borrowed)
            .into_iter()
            .flatten()
        {
            let p = starts.partition_point(|&s| s <= offset);
            if p == 0 {
                continue;
            }
            let i = order[p - 1];
            if offset < packs[i].0 + packs[i].1 {
                has_live[i] = true;
            }
        }

        let dead: Vec<u64> = (0..packs.len())
            .filter(|&i| occupied[i] && !has_live[i] && !already.contains(&packs[i].0))
            .map(|i| packs[i].0)
            .collect();
        // Durable before the rows go. The one ordering this function has.
        crate::archive_write::retire_packs(&journal, &dead)?;
        Ok(dead)
    }

    /// **Write generation 0** at [`archive_path`](Self::archive_path), carrying
    /// the verbatim packs this store has acked plus `reserved`.
    ///
    /// The whole of the decision-making lives in
    /// [`seal_generation_zero`](crate::archive_write::seal_generation_zero); this
    /// is the three paths it needs, taken off the same fields the push path and
    /// the GC read, so a seal cannot address a different blob file or a different
    /// journal from the one that acked the packs.
    pub(crate) fn seal_archive(
        &self,
        reserved: Vec<ReservedSection>,
    ) -> Result<crate::archive_write::SealReport> {
        crate::archive_write::seal_generation_zero(
            &self.blobs,
            self.arms.writer.journal(&self.blobs).as_deref(),
            &self.archive,
            reserved,
        )
    }

    /// The sections a sealed archive carries: the ref log, the commit graph and
    /// the bitmaps.
    pub(crate) fn reserved_sections(&self) -> Result<Vec<ReservedSection>> {
        use znippy_common::{GUNNAR_GRAPH_MODULE, GUNNAR_REACH_MODULE};
        let mut out = vec![self.refs.seal_section()?];
        let d = self
            .absorber
            .derived
            .read()
            .map_err(|_| anyhow!("derived poisoned"))?;
        if !d.graph.is_empty() {
            out.push(ReservedSection::arrow(
                GUNNAR_GRAPH_MODULE,
                crate::graph::graph_schema(),
                vec![crate::graph::build_graph_batch(&d.graph)?],
            ));
        }
        drop(d);
        let reach = self.reach_bitmaps()?;
        if !reach.is_empty() {
            out.push(ReservedSection::arrow(
                GUNNAR_REACH_MODULE,
                crate::reach::reach_schema(),
                vec![crate::reach::build_reach_batch(&reach)?],
            ));
        }
        Ok(out)
    }
}

// ── the runtime selector ──────────────────────────────────────────────────────

/// **A [`GitStore`] whose index arm was chosen at run time.**
///
/// The [`ObjectIndex`] arm is a *type*, so a value can only choose it by naming
/// every monomorphisation. This enum is that list, and it is an enum rather than
/// a `Box<dyn GitOps>` for two reasons:
///
/// * a boxed trait object can only offer the twelve, and the arms differ in
///   things the twelve deliberately do not expose — how many Arrow bytes the
///   projection actually materialised, which writer is on the ack path. A bench
///   or an operator dump needs those, and a guard needs them to assert that the
///   selector selected **on applied output** rather than on a label;
/// * there is no allocation and no vtable: the `match` is one predictable branch
///   at the top of a call, and everything under it — the `stree` probe, the
///   Arrow gather, the redb tail — is the same fully static code
///   [`GitStore::<S>::open_with_arms`](GitStore::open_with_arms) produces,
///   because `S` is known inside each arm.
///
/// # What it costs, stated rather than hidden
///
/// One branch per [`GitOps`] call — **not per object**. `extents(&[1000 oids])`
/// is one branch and a thousand monomorphised lookups behind it. The binary
/// carries three copies of the store, which is the price of picking a type at
/// run time and there is no version of this that does not pay it.
///
/// A caller that knows its arm at compile time should **not** come through here:
/// [`GitStore::open`] and [`GitStore::open_with_arms`] hand back a concrete
/// store with no branch at all, and that includes every caller that wants the
/// default.
pub enum SelectedStore {
    OneTableFourColumns(GitStore<OneTableFourColumns>),
    FourTables(GitStore<crate::index_layout::FourTables>),
    PackedPayload(GitStore<crate::index_layout::PackedPayload>),
}

/// One expression, evaluated against whichever concrete store this is. The
/// delegation below is generated from it so there is no second copy of any
/// method body (LAW 5).
macro_rules! on_arm {
    ($self:ident, $s:ident => $body:expr) => {
        match $self {
            SelectedStore::OneTableFourColumns($s) => $body,
            SelectedStore::FourTables($s) => $body,
            SelectedStore::PackedPayload($s) => $body,
        }
    };
}

impl SelectedStore {
    /// The arms this store was built with.
    pub fn arms(&self) -> StoreConfig {
        on_arm!(self, s => s.arms())
    }

    /// The selected writer's name, as it goes on a bench row.
    pub fn writer_name(&self) -> &'static str {
        on_arm!(self, s => s.writer_name())
    }

    /// What the selected writer's `append` promises.
    pub fn writer_durability(&self) -> &'static str {
        on_arm!(self, s => s.writer_durability())
    }

    /// **Arrow IPC bytes the selected index arm actually materialised.**
    ///
    /// The applied output that distinguishes one layout from another: a packed
    /// 25-byte column, four columns in one section and four independent
    /// sections are three different numbers for the same objects. This is what
    /// a guard reads to prove the selector selected, because a *name* would
    /// prove only that a name was copied.
    pub fn index_ipc_bytes(&self) -> usize {
        on_arm!(self, s => s.index().ipc_bytes())
    }

    /// **The name the built projection reports about itself** —
    /// `OneTableFourColumns`, `FourTables` or `PackedPayload`.
    ///
    /// The other half of the applied-output pair [`index_ipc_bytes`] starts:
    /// the byte count separates the arms once objects are in the store, and
    /// this separates them from the instant it is built, including on an empty
    /// store where all three materialise nothing. It is the value
    /// [`crate::arms::IndexArm::projection_name`] exists to be compared with —
    /// what the layout calls itself, never what the selector asked for — so a
    /// server logging it is stating what it built rather than repeating its own
    /// environment back.
    ///
    /// [`index_ipc_bytes`]: SelectedStore::index_ipc_bytes
    pub fn index_name(&self) -> &'static str {
        on_arm!(self, s => s.index().projection_name())
    }

    /// The oids a pushed extent introduced — [`GitStore::oids_in_extent`],
    /// delegated.
    ///
    /// Not one of the twelve (a pack's extent is a znippy concept), so it does
    /// not arrive with the [`GitOps`] impl below and has to be forwarded by
    /// hand. A server that reports the objects a push introduced needs it, and
    /// a server that picked its index arm at run time still needs it.
    pub fn oids_in_extent(&self, extent: Extent) -> Result<Vec<Vec<u8>>> {
        on_arm!(self, s => s.oids_in_extent(extent))
    }

    /// Objects in the `objects` table.
    pub fn object_count(&self) -> usize {
        on_arm!(self, s => s.object_count())
    }

    /// Block until the background drain has absorbed everything pushed.
    pub fn wait_indexed(&self) {
        on_arm!(self, s => s.wait_indexed())
    }

    /// Rebuild the projection, so the layout under test is what answers rather
    /// than the redb tail.
    pub fn rebuild_projection(&self) -> Result<()> {
        on_arm!(self, s => s.index().rebuild())
    }

    /// The twelfth method — inherent, not on [`GitOps`], because it returns
    /// Arrow `ReservedSection`s a gix backend has no analog for.
    pub fn seal(&self) -> Result<Vec<ReservedSection>> {
        on_arm!(self, s => s.seal())
    }
}

impl GitOps for SelectedStore {
    fn put(&self, pack: &[u8], refs: &[RefUpdate]) -> Result<TxId> {
        on_arm!(self, s => s.put(pack, refs))
    }
    fn put_pack(&self, bytes: &[u8]) -> Result<TxId> {
        on_arm!(self, s => s.put_pack(bytes))
    }
    fn put_refs(&self, updates: &[RefUpdate]) -> Result<TxId> {
        on_arm!(self, s => s.put_refs(updates))
    }
    fn get(&self, oid: Oid<'_>) -> Result<Option<Stored>> {
        on_arm!(self, s => s.get(oid))
    }
    fn has(&self, oid: Oid<'_>) -> Result<bool> {
        on_arm!(self, s => s.has(oid))
    }
    fn size(&self, oid: Oid<'_>) -> Result<Option<u64>> {
        on_arm!(self, s => s.size(oid))
    }
    fn extents(&self, oids: &[Oid<'_>]) -> Result<Vec<Option<Extent>>> {
        on_arm!(self, s => s.extents(oids))
    }
    fn refs(&self) -> Result<Vec<RefRow>> {
        on_arm!(self, s => s.refs())
    }
    fn update_ref(&self, name: &str, old: Option<Oid<'_>>, new: Option<Oid<'_>>) -> Result<TxId> {
        on_arm!(self, s => s.update_ref(name, old, new))
    }
    fn put_refs_cas(&self, edits: &[git_storage_trait::RefCas<'_>]) -> Result<TxId> {
        on_arm!(self, s => s.put_refs_cas(edits))
    }
    fn reachable(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> Result<Vec<Vec<u8>>> {
        on_arm!(self, s => s.reachable(want, have))
    }
    fn gc(&self) -> Result<GcReport> {
        on_arm!(self, s => s.gc())
    }
}

/// The reading contract, delegated the same way [`GitOps`] is.
///
/// Every method is one predictable branch over a monomorphised body — the same
/// arrangement, and the same cost argument, [`SelectedStore`] makes for the
/// eleven. It is here rather than in [`crate::serve`] only because the macro
/// that generates it lives in this module.
impl crate::serve::GitServe for SelectedStore {
    fn read(&self, oid: Oid<'_>) -> Result<Option<(crate::index_layout::ObjType, Vec<u8>)>> {
        on_arm!(self, s => crate::serve::GitServe::read(s, oid))
    }
    fn header(&self, oid: Oid<'_>) -> Result<Option<(crate::index_layout::ObjType, u64)>> {
        on_arm!(self, s => crate::serve::GitServe::header(s, oid))
    }
    fn sizes(&self, oids: &[Oid<'_>]) -> Result<Vec<Option<u64>>> {
        on_arm!(self, s => crate::serve::GitServe::sizes(s, oids))
    }
    fn head(&self) -> Result<Option<RefRow>> {
        on_arm!(self, s => crate::serve::GitServe::head(s))
    }
    fn set_head(&self, target: &str) -> Result<TxId> {
        on_arm!(self, s => crate::serve::GitServe::set_head(s, target))
    }
    /// `objects` is the **exact set to emit**, not tips to close over — the
    /// contract's word, and this forwarder uses it so that the name cannot drift
    /// back into `want` one arm at a time. See [`crate::serve::GitServe::emit_pack`].
    fn emit_pack(
        &self,
        objects: &[Oid<'_>],
        have: &[Oid<'_>],
        caps: &git_storage_trait::Caps,
        out: &mut dyn std::io::Write,
    ) -> Result<git_storage_trait::PackStats> {
        on_arm!(self, s => crate::serve::GitServe::emit_pack(s, objects, have, caps, out))
    }
    fn select(
        &self,
        want: &[Oid<'_>],
        have: &[Oid<'_>],
    ) -> Result<Option<git_storage_trait::ReachSet>> {
        on_arm!(self, s => crate::serve::GitServe::select(s, want, have))
    }
}

/// **Open a store with all three arms chosen at run time.**
///
/// The one place [`crate::arms::IndexArm`] — a value — is turned into `S` — a
/// type. See [`SelectedStore`] for what the resulting handle costs.
pub fn open_selected(
    root: &Path,
    account: &str,
    hash: GitHashKind,
    arms: StoreConfig,
) -> Result<SelectedStore> {
    use crate::arms::IndexArm;
    use crate::index_layout::{FourTables, PackedPayload};
    Ok(match arms.index {
        IndexArm::OneTableFourColumns => {
            SelectedStore::OneTableFourColumns(GitStore::<OneTableFourColumns>::open_with_arms(
                root, account, hash, arms,
            )?)
        }
        IndexArm::FourTables => SelectedStore::FourTables(GitStore::<FourTables>::open_with_arms(
            root, account, hash, arms,
        )?),
        IndexArm::PackedPayload => SelectedStore::PackedPayload(
            GitStore::<PackedPayload>::open_with_arms(root, account, hash, arms)?,
        ),
    })
}

/// **The env-driven front door**: read the three variables **once**, here, and
/// build the store they name.
///
/// This is the whole runtime selector a server or a bench needs — one call, one
/// triple of `getenv`s, and from then on the store holds built objects and reads
/// nothing. There is deliberately no per-operation lookup of any kind: a
/// `std::env::var` inside a serving loop would be one syscall per object served,
/// which is a real defect that was found and fixed in gunnar the day before this
/// was written. [`crate::arms::env_reads`] counts every read this crate makes so
/// that "once, at construction" is an assertion rather than a claim.
///
/// An unset variable takes the shipping default, so a process with a clean
/// environment gets exactly [`GitStore::open`]'s combination.
pub fn open_from_env(root: &Path, account: &str, hash: GitHashKind) -> Result<SelectedStore> {
    let arms = StoreConfig::from_env()?;
    open_selected(root, account, hash, arms)
}

impl<S: ObjectIndex> Absorber<S> {
    /// **One object's resolved content, and §14's table is what answers.**
    ///
    /// Two paths, and they are not interchangeable even though they return
    /// identical bytes:
    ///
    /// 1. **the exploded table** — one point lookup, no inflate, no delta chain;
    /// 2. **re-derived from the verbatim truth** — find which pack holds the
    ///    object, read the whole pack back, resolve it. Correct and slow, and it
    ///    exists because the table is droppable: a store whose table has been
    ///    deleted, or whose re-explosion has not caught up, must still answer.
    ///
    /// The counters are the only way to tell them apart. Identical bytes come
    /// back either way, so a byte comparison proves nothing about which path ran
    /// — that is the identity-value trap in its exact form, and it is why
    /// [`crate::exploded::ExplodedStats::rederived`] is bumped here rather than
    /// inferred.
    ///
    /// On the [`Absorber`] rather than on [`GitStore`] because that is where the
    /// absorb runs: the background worker has no store handle, and a base lookup
    /// that needed one would put the whole store behind the drain.
    fn resolved(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
        // (1) §14's table.
        if let Some(hit) = self.exploded.content(oid)? {
            return Ok(Some(hit));
        }
        // (2) the verbatim truth, re-derived.
        let Some(row) = self.objects.lookup(oid) else {
            return Ok(None);
        };
        let found = {
            let d = self
                .derived
                .read()
                .map_err(|_| anyhow!("derived poisoned"))?;
            d.packs
                .iter()
                .find(|(o, l)| row.offset >= *o && row.offset < o + l)
                .copied()
        };
        let Some((pack_offset, pack_len)) = found else {
            return Ok(None);
        };
        self.exploded.note_rederived();
        let bytes = self.read_extent(pack_offset, pack_len)?;
        let walked = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
        // Through the sink, not through `Resolved::payload`: the resolver drops a
        // blob payload as soon as nothing in the pack deltas against it, so
        // reading that field answered `None` for most blobs — the commonest
        // object there is. The sink sees it before it is dropped.
        let capture = crate::exploded::CaptureOne::new(oid);
        crate::resolve::resolve_walked(
            &bytes,
            &walked,
            self.hash,
            pack_offset,
            &crate::resolve::NoBases,
            &capture,
        )?;
        Ok(capture.take())
    }
}

/// A thin pack's external delta base, answered from §14's exploded table.
///
/// Before that table existed this had to read and re-resolve a whole pack per
/// base, and it deliberately cached nothing — because a cache here *would* have
/// been the undecided table. It is decided now (eager, 2026-08-08), so the cache
/// is the table and this is one point lookup.
///
/// **The refusal is unchanged**: a base this cannot produce still comes back
/// `None`, and [`crate::resolve::ResolveError::missing_base`] refuses the pack by
/// name rather than half-resolving it.
impl<S: ObjectIndex> BaseSource for Absorber<S> {
    fn content(&self, oid: &[u8]) -> Option<(GitObjectKind, Vec<u8>)> {
        self.resolved(oid).ok().flatten()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::archive_write::{read_journal, SafeWriter};
    use crate::object::GitObjectKind;
    use crate::store::tests::{one_blob_pack, real_pack, tmpdir};
    use std::collections::HashSet;
    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomicOrdering};
    use std::time::{Duration, Instant};
    use znippy_zoomies::background::Job;
    use znippy_zoomies::gatling_forkjoin::gatling_for_each;

    fn loadavg() -> String {
        std::fs::read_to_string("/proc/loadavg")
            .unwrap_or_default()
            .split_whitespace()
            .take(3)
            .collect::<Vec<_>>()
            .join(" ")
    }

    /// **The ordinal space has two spellings and they must be ONE sequence.**
    ///
    /// `Derived::oids` (hex) is what the ordinal map, the commit graph and the
    /// tree payloads are keyed by; `Derived::oids_raw` (flat bytes) is what
    /// every serving answer is built out of, because it is what the bitmaps'
    /// ordinals index into. They are folded from one sorted vector so they
    /// cannot disagree — this asserts the property that argument claims, at
    /// every ordinal, because an off-by-one between them is a server that names
    /// the wrong object in a pack and no exit code can see it.
    ///
    /// Seen RED by folding the raw form in reverse: 3 341 of 3 341 ordinals
    /// named a different object, and the closure comparison below disagreed as
    /// well.
    #[test]
    fn the_raw_ordinal_space_and_the_hex_one_are_the_same_sequence() {
        let dir = tmpdir("ordinal-space-two-spellings");
        let (pack, rows) = real_pack();
        let store = GitStore::open(&dir, "rickard").unwrap();
        store.put_pack(&pack).unwrap();
        store.wait_indexed();

        let oid_len = store.hash_kind().oid_len();
        let (hex, raw) = {
            let d = store.absorber.derived.read().unwrap();
            (d.oids.clone(), d.oids_raw.clone())
        };
        assert!(
            !hex.is_empty(),
            "the fixture pack ({} entries) folded to an EMPTY ordinal space, so every \
             comparison below would hold vacuously",
            rows.len()
        );
        assert_eq!(
            raw.len(),
            hex.len() * oid_len,
            "the raw ordinal space is {} bytes for {} oids of {oid_len} bytes",
            raw.len(),
            hex.len()
        );
        let mut checked = 0usize;
        for (o, h) in hex.iter().enumerate() {
            assert_eq!(
                &raw[o * oid_len..(o + 1) * oid_len],
                hex::decode(h).unwrap().as_slice(),
                "ordinal {o} is {h} in the hex space and something else in the raw one"
            );
            checked += 1;
        }
        assert_eq!(checked, hex.len());

        // And the answer built out of each: the raw closure IS the hex closure,
        // for a real tip rather than for an empty request.
        let tip = store
            .graph_snapshot()
            .into_iter()
            .max_by_key(|c| c.generation)
            .expect("a graph");
        let tip_raw = hex::decode(&tip.oid).unwrap();
        let flat = store.reachable_raw(&[&tip_raw], &[]).unwrap();
        let hexed = store.reachable_oids(&[&tip_raw], &[]).unwrap();
        assert!(
            flat.len() > 1,
            "the tip's closure is {} object(s) — too small to tell an ordering bug from a \
             coincidence",
            flat.len()
        );
        assert_eq!(
            flat.iter().map(hex::encode).collect::<Vec<_>>(),
            hexed,
            "the serving answer and the maintenance answer name different objects"
        );
        assert!(flat.contains(&tip_raw), "a closure without its own tip");
        eprintln!(
            "load {}; {} ordinals agree in both spellings, tip closure {} objects",
            loadavg(),
            checked,
            flat.len()
        );
    }

    /// Distinct commits in a resolved pack — what the graph must hold exactly
    /// once.
    fn commits_in(rows: &[crate::resolve::Resolved]) -> usize {
        rows.iter()
            .filter(|r| r.kind == GitObjectKind::Commit)
            .map(|r| r.oid.clone())
            .collect::<HashSet<Vec<u8>>>()
            .len()
    }

    /// **A push ends in object rows, and no read asked for them.**
    ///
    /// This is the seam this file wires: before it, `put` produced a *pack* row
    /// on the account indexer's channel and the object rows appeared only when
    /// somebody read (or called `absorb_pending` by hand). Here nothing is read
    /// between `put_pack` and `wait_indexed` — the only thing that could have
    /// built the rows is the drain.
    ///
    /// Asserted on applied output: one row per pack entry, every oid resolving
    /// out of the index, and the pack's indexed bit up. `absorb_failures` is
    /// asserted too, because a drain that absorbed nothing and recorded the
    /// failure would otherwise look like a drain that was never called.
    ///
    /// Seen RED by making the drain skip the absorber
    /// (`match absorber.as_deref()` → `match None::<&dyn ObjectAbsorb>` in
    /// `indexer::index_worker`): "the drain built no object rows: 0 of 2687".
    /// Restored.
    ///
    /// MEASURED on oden, 2026-08-08, release, three runs at **1-minute loadavg
    /// 5.77** (another tenant's work — oden is shared, so this is not an idle
    /// box): 2687 objects out of a 5 653 302-byte real pack absorbed in
    /// 161–164 ms, **16 421–16 673 object rows/s**, behind an ack of 29.1–29.4 ms.
    /// An earlier set at loadavg 16.94 gave 15 528 rows/s, so the figure moves
    /// about 7% with the box.
    ///
    /// Read it as a *floor* rather than as the engine's speed: the drain here is
    /// one pack, so it never uses more than one absorb, and the 162 ms covers a
    /// full inflate-and-delta-apply of every entry, a SHA-1 per object, the redb
    /// append and a whole-tail re-fold. Nothing about it has been optimised and
    /// no arm of it has been isolated. The rate is printed with the loadavg of
    /// its own run so a later number is never compared against a different
    /// machine state by accident.
    #[test]
    fn a_push_ends_in_object_rows_and_no_read_asked_for_them() {
        let dir = tmpdir("drain-rows");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, rows) = real_pack();

        let t = Instant::now();
        let tx = store.put_pack(&pack).unwrap();
        let ack = t.elapsed();
        let pack_id = tx.pack_id.expect("a pack push assigns an id");

        // Nothing is read here. The drain is the only thing that can move.
        store.wait_indexed();
        let drained = t.elapsed();

        assert_eq!(
            store.object_count(),
            rows.len(),
            "the drain built no object rows: {} of {}",
            store.object_count(),
            rows.len()
        );
        assert_eq!(store.unindexed_packs(), 0, "the bit is still clear");
        assert!(
            store.indexer().is_indexed(pack_id),
            "the pack's indexed bit is not up after the drain"
        );
        assert_eq!(
            store.indexer().absorb_failures(),
            0,
            "the drain recorded an absorb failure: {:?}",
            store.indexer().last_absorb_error()
        );

        // Every oid answers **out of the rows**, not out of a fallback: the
        // index is asked directly, so nothing can absorb behind this loop.
        for (i, r) in rows.iter().enumerate() {
            let row = store
                .index()
                .lookup(&r.oid)
                .unwrap_or_else(|| panic!("object {i} {} has no row", hex::encode(&r.oid)));
            assert_eq!(
                (row.offset, row.len),
                (tx.extent.unwrap().0 + r.offset, r.len)
            );
            assert_eq!(row.uncompressed_size, r.uncompressed_size);
        }
        assert_eq!(
            store.absorb_pending().unwrap(),
            0,
            "the drain left work for a read to do"
        );
        assert_eq!(
            store.commit_count(),
            commits_in(&rows),
            "the commit graph does not match the pack"
        );

        let secs = (drained - ack).as_secs_f64();
        eprintln!(
            "load {}; {} objects in {} bytes: ack {:.0} µs, drain {:.1} ms, {:.0} object rows/s",
            loadavg(),
            rows.len(),
            pack.len(),
            ack.as_secs_f64() * 1e6,
            (drained - ack).as_secs_f64() * 1e3,
            rows.len() as f64 / secs.max(1e-9),
        );
    }

    // ── the bit across a restart ─────────────────────────────────────────────

    /// Where the child of the kill guard finds its fixture and leaves its marker.
    const KILL_DIR: &str = "GUNNAR_KILL_MID_ABSORB_DIR";

    /// **The other half of [`a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack`]:
    /// the process that dies.**
    ///
    /// It runs as a child of that test and it **aborts on purpose**, which is why
    /// it is `#[ignore]`d: `SIGABRT` from inside a test binary is a failing test
    /// run, and the point is for the *parent* to see the signal.
    ///
    /// The kill lands where it has to land — after `put_pack` has returned, so
    /// the bytes and the journal row are on the platter, and before any row is in
    /// the index, which the held absorb gate makes a **state** rather than a
    /// hope. Nothing is simulated: no injected error, no fault flag, no early
    /// return. The process is gone.
    #[test]
    #[ignore = "spawned by a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack; it aborts on purpose"]
    fn the_child_that_dies_between_durable_bytes_and_indexed_rows() {
        let Ok(dir) = std::env::var(KILL_DIR) else {
            // Run by hand with `--ignored` and no fixture: there is nothing to
            // kill over, so do nothing rather than abort somebody's test run.
            return;
        };
        let dir = PathBuf::from(dir);
        let pack = std::fs::read(dir.join("fixture.pack")).expect("the parent's fixture");

        let store = GitStore::open(&dir, "rickard").expect("open");
        // The drain is parked in front of its absorb for as long as this lives,
        // so no row can land before the abort below.
        let _gate = store.hold_absorb_gate();
        let tx = store.put_pack(&pack).expect("the ack path");
        assert_eq!(tx.pack_id, Some(0), "a fresh archive starts at ordinal 0");
        assert_eq!(store.unindexed_packs(), 1, "the pack is not queued");
        assert_eq!(store.index().len(), 0, "a row landed before the kill");
        std::fs::write(dir.join("ready"), b"durable, not indexed").expect("marker");

        // Not `panic!`, not `exit`: the process stops here with a live gate, a
        // live worker and a live redb handle, which is what a machine that dies
        // mid-absorb leaves behind.
        std::process::abort();
    }

    /// **A store reopened over a pack that was still being absorbed re-queues
    /// it, answers correctly the whole way through, and gets its derived tables
    /// back.**
    ///
    /// The interruption is a **real** one: a child process is spawned, pushes the
    /// pack, and is killed by `SIGABRT` between "bytes durable" and "rows
    /// indexed". The parent then asserts what the kill left on disk — one journal
    /// row, the pack verbatim, **zero** index rows — before opening a store over
    /// it, so the recovery is being tested against the state it claims to
    /// recover from and not against a lucky one.
    ///
    /// Three things are then asserted, all on applied output:
    ///
    /// 1. **the pack is re-queued** — the account indexer of the *new* process
    ///    publishes ordinal 0, which it can only do for a job somebody submitted;
    /// 2. **a read is correct throughout** — `has` is asked the instant the store
    ///    is open, before any drain has been waited for, and must answer `true`;
    /// 3. **`commit_count()` and `reachable()` come back** — the derived tables
    ///    are rebuilt by the same absorb, so the graph holds exactly the pack's
    ///    commits and a tip's closure is non-empty.
    ///
    /// Seen RED by making `open_with` re-queue nothing (`absorber
    /// .adopt_journal(&acked)?` → an empty `Vec`, which is what this store did
    /// before this change): "a durable object read `absent` from a store reopened
    /// over the pack that holds it — the crash-recovery bit did not re-queue
    /// anything". Restored.
    ///
    /// Seen RED a second time by inverting the diff in
    /// [`Absorber::adopt_journal`] (`if absorbed` → `if !absorbed`), so a pack
    /// with **no** rows has its bit set and a pack with rows is re-queued: the
    /// same first assertion fires, because a set bit over an empty index is
    /// exactly the wrong answer a bloom filter's false positive would have
    /// produced — the read stops falling back and says `absent` about an object
    /// whose bytes are durable. Restored.
    ///
    /// **Why the assertions are not immediately after the reopen.** The trap the
    /// drain guards already record applies here too, mirrored: right after
    /// `GitStore::open` the drain may legitimately have finished the re-queued
    /// pack already, so `unindexed_packs() == 1` is not assertable and is not
    /// asserted. What *is* assertable at that instant is the answer to a read,
    /// which must be `true` whichever side of the drain it lands on — and that is
    /// exactly the property §13.12 promises.
    ///
    /// **And it still holds now that a `gc` can retire a pack's journal row.**
    /// The tombstone tells "indexed then garbage-collected" from "never
    /// indexed", and this pack is the second: nothing retired it, so it must
    /// still be re-queued. Seen RED by making `adopt_journal` treat *every* pack
    /// as retired (`retired.contains(&extent.0)` → `true`): "a durable object
    /// read `absent` from a store reopened over the pack that holds it — the
    /// crash-recovery bit did not re-queue anything". Restored.
    #[test]
    fn a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack() {
        use std::os::unix::process::ExitStatusExt;

        let dir = tmpdir("kill-mid-absorb");
        let (pack, rows) = real_pack();
        std::fs::write(dir.join("fixture.pack"), &pack).unwrap();

        let status = std::process::Command::new(std::env::current_exe().unwrap())
            .args([
                "--exact",
                "--ignored",
                "--nocapture",
                "git_ops::tests::the_child_that_dies_between_durable_bytes_and_indexed_rows",
            ])
            .env(KILL_DIR, &dir)
            .status()
            .expect("spawn the child that dies");
        assert_eq!(
            status.signal(),
            Some(6),
            "the child did not die by SIGABRT — it {status:?}, so nothing was interrupted"
        );
        assert!(
            dir.join("ready").exists(),
            "the child never reached the kill point"
        );

        // ── what the kill left on disk ───────────────────────────────────────
        let blobs = dir.join("objects.pack");
        let acked = read_journal(&SafeWriter::journal_path(&blobs)).unwrap();
        assert_eq!(
            acked.len(),
            1,
            "the pack's extent is not durable: {acked:?}"
        );
        assert_eq!(acked[0].1, pack.len() as u64);
        let on_disk = std::fs::read(&blobs).unwrap();
        assert_eq!(
            &on_disk[acked[0].0 as usize..(acked[0].0 + acked[0].1) as usize],
            &pack[..],
            "the verbatim bytes did not survive the kill"
        );
        {
            // Opened and dropped before the store takes redb's file lock.
            let tail = ObjectReadStack::<OneTableFourColumns>::open(
                &dir.join("objects.tail"),
                RebuildTriggers::default(),
                crate::arms::DEFAULT_REDB_CACHE_BYTES,
            )
            .unwrap();
            assert_eq!(
                tail.len(),
                0,
                "the kill landed after the rows were indexed, not between the bytes and the rows \
                 — this guard would then be testing nothing"
            );
        }

        // ── the reopen ───────────────────────────────────────────────────────
        let store = GitStore::open(&dir, "rickard").unwrap();
        assert!(
            store.has(&rows[0].oid).unwrap(),
            "a durable object read `absent` from a store reopened over the pack that holds it — \
             the crash-recovery bit did not re-queue anything"
        );

        store.wait_indexed();
        assert_eq!(
            store.object_count(),
            rows.len(),
            "the re-queued pack's objects never reached the index — left: {}, right: {}",
            store.object_count(),
            rows.len()
        );
        assert!(
            store.indexer().is_indexed(0),
            "the re-queued pack's indexed bit never went up in the new process"
        );
        assert_eq!(store.unindexed_packs(), 0);
        assert_eq!(store.absorb_pending().unwrap(), 0);
        for r in rows.iter().take(512) {
            let row = store
                .index()
                .lookup(&r.oid)
                .unwrap_or_else(|| panic!("{} has no row after recovery", hex::encode(&r.oid)));
            assert_eq!((row.offset, row.len), (acked[0].0 + r.offset, r.len));
        }

        // ── the derived tables, which were the second symptom ────────────────
        assert_eq!(
            store.commit_count(),
            commits_in(&rows),
            "the commit graph did not come back: {} of {}",
            store.commit_count(),
            commits_in(&rows)
        );
        let tip = store
            .graph_snapshot()
            .into_iter()
            .max_by_key(|c| c.generation)
            .expect("a graph");
        let tip_raw = hex::decode(&tip.oid).unwrap();
        let closure = store.reachable(&[&tip_raw], &[]).unwrap();
        assert!(
            !closure.is_empty(),
            "reachable() came back empty after the reopen"
        );
        assert!(closure.contains(&tip_raw));
        eprintln!(
            "load {}; killed mid-absorb: {} journal extent(s) recovered {} object rows, {} \
             commits, closure of the tip {} objects",
            loadavg(),
            acked.len(),
            store.object_count(),
            store.commit_count(),
            closure.len(),
        );
    }

    /// **A clean reopen re-queues nothing, and the next push gets a fresh
    /// ordinal.**
    ///
    /// The other side of the diff. A store whose pack was fully absorbed before
    /// it closed must come back with the bit **set**: re-absorbing would be a
    /// whole pack re-resolved for nothing, and — because the commit graph is a
    /// `Vec` — it is the same doubling the drain's own guard exists to prevent.
    ///
    /// "Re-queued nothing" is asserted deterministically rather than by a timing
    /// window: the account indexer in this process publishes a pack only for a
    /// job somebody submitted, so `is_indexed(0)` being **false** after
    /// `wait_indexed` is proof that open submitted none.
    ///
    /// The second half is the ordinal. Pack ids are the journal's row numbers, so
    /// a reopened store must resume after them. Asserted on applied output rather
    /// than on the id alone: the newly pushed pack's object has to be findable,
    /// and it is exactly what goes missing when a fresh pack is handed an
    /// absorbed pack's ordinal — `queue` skips it, the drain skips it, and its
    /// durable bytes are never indexed by anybody.
    ///
    /// Seen RED by making `Absorber::adopt_journal` treat every journal extent as
    /// unabsorbed (`let has_rows = vec![false; journal.len()]`): "a fully
    /// absorbed pack was re-queued on open — the diff is not exact".
    ///
    /// Seen RED a second time by `AtomicU64::new(packs_already_acked(...)?)` →
    /// `AtomicU64::new(0)` in `PushPath::with_absorber`: "a reopened store handed
    /// a fresh pack the ordinal of an absorbed one — left: Some(0), right:
    /// Some(1)". With that same edit **and the id assertion deleted**, the
    /// applied-output one goes too — "the second push's object never reached the
    /// index — its ordinal collided with an absorbed pack's, so every writer
    /// skipped it" — which is what says this guard is about the objects and not
    /// about a counter. Restored.
    #[test]
    fn a_clean_reopen_requeues_nothing_and_the_next_push_gets_a_fresh_ordinal() {
        let dir = tmpdir("clean-reopen");
        let (pack, rows) = real_pack();
        let (second, blob_oid) = one_blob_pack(b"a push that arrives after the restart");

        {
            let store = GitStore::open(&dir, "rickard").unwrap();
            store.put_pack(&pack).unwrap();
            store.wait_indexed();
            assert_eq!(
                store.object_count(),
                rows.len(),
                "the first run did not index"
            );
        }

        let store = GitStore::open(&dir, "rickard").unwrap();
        store.wait_indexed();
        assert!(
            !store.indexer().is_indexed(0),
            "a fully absorbed pack was re-queued on open — the diff is not exact"
        );
        assert_eq!(store.unindexed_packs(), 0);
        assert_eq!(
            store.absorb_pending().unwrap(),
            0,
            "the reopen left index work"
        );
        assert_eq!(
            store.object_count(),
            rows.len(),
            "the reopened store lost rows"
        );
        assert!(store.has(&rows[0].oid).unwrap());

        let tx = store.put_pack(&second).unwrap();
        assert_eq!(
            tx.pack_id,
            Some(1),
            "a reopened store handed a fresh pack the ordinal of an absorbed one"
        );
        store.wait_indexed();
        assert!(
            store.has(&blob_oid).unwrap(),
            "the second push's object never reached the index — its ordinal collided with an \
             absorbed pack's, so every writer skipped it"
        );
        assert_eq!(store.object_count(), rows.len() + 1);
        assert!(store.indexer().is_indexed(1));
        // And the journal is a log: both packs' extents are in it, in order.
        let acked = read_journal(&SafeWriter::journal_path(store.blobs_path())).unwrap();
        assert_eq!(
            acked.len(),
            2,
            "the reopen truncated the journal: {acked:?}"
        );
        assert_eq!(acked[1], tx.extent.unwrap());
    }

    /// **What the derive-on-open diff costs at a realistic pack count.**
    ///
    /// Not a guard — a measurement, with two assertions on it that keep it from
    /// measuring the wrong thing: every absorbed pack must come back `true`, and
    /// the one pack that has no rows must come back `false`.
    ///
    /// Both cases are timed because they are different algorithms in practice.
    /// The **clean reopen** — every pack absorbed — terminates as soon as each
    /// extent has been hit once, which happens after a few rows per pack because
    /// the tail is in oid order and an oid says nothing about which pack its
    /// object came from. The **crash** case has one pack with no rows at all, and
    /// proving an absence means reading the tail to the end.
    ///
    /// MEASURED on oden 2026-08-08, release, 1000 real pushed packs / 3686 object
    /// rows, 1-minute loadavg 2.77: **0.38 ms** clean and 0.39 ms with one pack
    /// un-absorbed (0.40 / 0.39 ms at loadavg 13.06, so the figure barely moves
    /// with the box). At this row count the two cases are the same, because a
    /// 3686-row tail is scanned to the end faster than the early exit saves
    /// anything; they diverge with the object count, and the table on
    /// [`ObjectReadStack::extents_with_rows`] carries that at a million rows
    /// (0.74 ms against 109 ms).
    ///
    /// The printed line carries the loadavg of its own run, the pack count and
    /// the row count, because none of the three is comparable across machines or
    /// across a busy box.
    #[test]
    fn the_derive_on_open_diff_costs_milliseconds_at_a_realistic_pack_count() {
        const PACKS: usize = 1000;
        let dir = tmpdir("diff-cost");
        let (real, rows) = real_pack();

        {
            let store = GitStore::open(&dir, "rickard").unwrap();
            store.put_pack(&real).unwrap();
            for i in 0..PACKS - 1 {
                let (p, _) = one_blob_pack(format!("pack number {i}").as_bytes());
                store.put_pack(&p).unwrap();
            }
            store.wait_indexed();
            assert_eq!(
                store.object_count(),
                rows.len() + PACKS - 1,
                "the fixture did not absorb"
            );
        }

        let acked = read_journal(&SafeWriter::journal_path(&dir.join("objects.pack"))).unwrap();
        assert_eq!(acked.len(), PACKS, "one journal row per pack");
        // The tail on its own, so what is timed is the diff rather than redb's
        // open and the projection build that a store's open also pays for.
        let tail = ObjectReadStack::<OneTableFourColumns>::open(
            &dir.join("objects.tail"),
            RebuildTriggers::default(),
            crate::arms::DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();

        let t = Instant::now();
        let hit = tail.extents_with_rows(&acked).unwrap();
        let clean = t.elapsed();
        assert!(
            hit.iter().all(|&h| h),
            "an absorbed pack was reported unabsorbed"
        );

        // One pack in flight when the machine died: durable bytes, no rows.
        let mut in_flight = acked.clone();
        let end = acked[PACKS - 1].0 + acked[PACKS - 1].1;
        in_flight.push((end, 4096));
        let t = Instant::now();
        let hit = tail.extents_with_rows(&in_flight).unwrap();
        let crashed = t.elapsed();
        assert_eq!(
            hit.iter().filter(|h| !**h).count(),
            1,
            "the pack with no rows is not the only one reported unabsorbed"
        );

        eprintln!(
            "load {}; {PACKS} packs / {} object rows: derive-on-open diff {:.2} ms clean \
             (early exit), {:.2} ms with one pack un-absorbed (full tail scan)",
            loadavg(),
            tail.len(),
            clean.as_secs_f64() * 1e3,
            crashed.as_secs_f64() * 1e3,
        );
    }

    /// **The bit is a bitset over dense ordinals, and one bit is one pack.**
    ///
    /// Ordinals that straddle word boundaries (63, 64, 65) are the whole point:
    /// a shift that ignores the word split, or a word index that does not, sets
    /// a neighbour's bit — and a neighbour's bit set means a durable pack is
    /// declared absorbed with no rows behind it.
    ///
    /// Seen RED by `>> (i % 64) & 1` → `>> (i % 63) & 1` in
    /// `PackState::is_absorbed`: "the bit for ordinal 64 did not go up". Restored.
    ///
    /// Seen RED a second time by `if self.extent[i].is_none()` → `if
    /// self.extent[i].is_none() || true` in `PackState::note`, i.e. letting a
    /// re-queue of a pack that is already recorded count as new work — which is
    /// the race the ack path really produces, since the drain can absorb before
    /// `queue` runs: "an absorbed pack was queued again — left: 6, right: 5".
    /// Restored.
    #[test]
    fn the_indexed_bit_is_a_bitset_over_dense_pack_ordinals() {
        let ids = [0u64, 1, 63, 64, 65, 4095];
        let mut s = PackState::default();
        for id in ids {
            s.note(id, (id * 4096 + 12, 4096));
        }
        assert_eq!(s.unabsorbed, ids.len());
        assert!(
            ids.iter().all(|&i| !s.is_absorbed(i)),
            "a bit is up already"
        );

        s.mark_absorbed(64, (64 * 4096 + 12, 4096));
        assert!(s.is_absorbed(64), "the bit for ordinal 64 did not go up");
        for other in [0u64, 1, 63, 65, 4095] {
            assert!(
                !s.is_absorbed(other),
                "setting ordinal 64 also set ordinal {other} — one bit spilled onto a neighbour"
            );
        }
        assert_eq!(s.unabsorbed, ids.len() - 1);
        assert_eq!(
            s.pending().iter().map(|p| p.pack_id).collect::<Vec<_>>(),
            vec![0, 1, 63, 65, 4095],
            "the pending list is not the clear bits"
        );
        assert_eq!(
            s.pending()[0],
            PendingPack {
                pack_id: 0,
                offset: 12,
                len: 4096
            },
            "a pending pack lost the extent it must be absorbed from"
        );

        // Both writers are idempotent, because both of them race.
        s.mark_absorbed(64, (64 * 4096 + 12, 4096));
        s.note(64, (64 * 4096 + 12, 4096));
        assert!(s.is_absorbed(64));
        assert_eq!(
            s.unabsorbed,
            ids.len() - 1,
            "an absorbed pack was queued again"
        );

        // One bit per pack, and that is the whole cost: 4096 ordinals is 64
        // words. A hash set of the same ordinals is an order of magnitude more,
        // and a bloom would be smaller and wrong.
        assert_eq!(s.words.len(), 64);
        assert_eq!(s.words.len() * 8, 512, "4096 packs is 512 bytes of bits");
    }

    /// **A read that arrives before the drain waits; it never answers absent.**
    ///
    /// The "before" is a *state*, not a race: the absorb gate is held by this
    /// thread, so the drain is provably parked at the top of its absorb and the
    /// index provably holds no row for the pack that `put_pack` just made
    /// durable.
    ///
    /// Two things are then asserted about a read issued into that state, and
    /// they are the two halves of §13.12. It must not **finish** — an answer
    /// while the index is empty could only have been "absent" — and when the
    /// gate is released it must answer **true**. After the drain, the same
    /// question is answered from the rows with nothing left to absorb.
    ///
    /// Seen RED **twice**, for the two halves.
    ///
    /// 1. The fallback: dropping `if self.unindexed_packs() > 0 {
    ///    self.absorb_pending()?; }` from `lookup_one` gave "a read that arrived
    ///    before the drain answered in 150 ms with an index holding 0 rows — the
    ///    only answer it can have given is `absent`". Restored.
    /// 2. The bit's ordering: publishing the pack rows in `index_worker`
    ///    *before* the absorb loop instead of after it gave "the pack's indexed
    ///    bit went up while the drain is still parked in front of its absorb —
    ///    the bit does not gate the object rows". Restored.
    ///
    /// **The second break is the reason the two assertions sit after the 150 ms
    /// sleep and not before it.** Asserted immediately after `put_pack` they were
    /// blind: the worker is still in its `pread`-and-SHA-1 fan-out at that
    /// instant, so the bit is legitimately down and the mutation stayed green. A
    /// guard for an ordering has to be read at a point where the wrong order has
    /// already had its chance.
    #[test]
    fn a_read_before_the_drain_waits_and_never_answers_absent() {
        let dir = tmpdir("drain-before");
        let store = Arc::new(GitStore::open(&dir, "rickard").unwrap());
        let (pack, rows) = real_pack();

        // The drain is parked here, before it can absorb anything.
        let gate = store.hold_absorb_gate();
        let tx = store.put_pack(&pack).unwrap();
        let pack_id = tx.pack_id.unwrap();

        assert_eq!(store.unindexed_packs(), 1, "the pack is not queued");
        assert_eq!(
            store.index().len(),
            0,
            "the index already holds rows — the 'before the drain' state is not the one under test"
        );
        assert!(
            !store.indexer().is_indexed(pack_id),
            "the bit is already up"
        );

        // A read into exactly that state.
        let oid = rows[0].oid.clone();
        let answered = Arc::new(AtomicBool::new(false));
        let (s, a) = (store.clone(), answered.clone());
        // `Job` is this constellation's one sanctioned background thread (LAW 3);
        // a bare `std::thread::spawn` here would trip `rayon_free_law`.
        let reader = Job::spawn(move || {
            let got = s.has(&oid).unwrap();
            a.store(true, AtomicOrdering::Release);
            got
        });
        std::thread::sleep(Duration::from_millis(150));
        assert!(
            !answered.load(AtomicOrdering::Acquire),
            "a read that arrived before the drain answered in 150 ms with an index holding {} \
             rows — the only answer it can have given is `absent`",
            store.index().len()
        );
        // The drain has had 150 ms and is parked *before* its absorb, which is
        // where the bit's ordering is observable: it must still be down, and the
        // index must still be empty.
        assert_eq!(
            store.index().len(),
            0,
            "the index gained rows while the absorb gate was held"
        );
        assert!(
            !store.indexer().is_indexed(pack_id),
            "the pack's indexed bit went up while the drain is still parked in front of its \
             absorb — the bit does not gate the object rows"
        );

        drop(gate);
        assert!(
            reader.join().unwrap(),
            "a read that waited for the drain still answered absent"
        );

        // And afterwards it is served from the rows.
        store.wait_indexed();
        assert_eq!(store.unindexed_packs(), 0);
        assert!(store.indexer().is_indexed(pack_id));
        assert_eq!(store.index().len(), rows.len());
        assert_eq!(store.absorb_pending().unwrap(), 0);
        for r in rows.iter().take(256) {
            assert!(
                store.index().lookup(&r.oid).is_some(),
                "{} is not in the rows after the drain",
                hex::encode(&r.oid)
            );
        }
    }

    /// **The bit is cleared after the rows go in, never before.**
    ///
    /// The ordering inside `absorb_gated` is the whole reason a read cannot be
    /// told "absent" while the drain is running, and the only way to see it is to
    /// read *during* one. Four gatling workers (LAW 3 — never rayon, and never a
    /// raw thread pool) hammer `has` over the pack's oids from the moment
    /// `put_pack` returns until the pack's bit is up.
    ///
    /// Asserted on applied output: not one of those reads may answer absent for
    /// an object whose bytes are already durable.
    ///
    /// Seen RED by clearing the bit *before* absorbing (a
    /// `s.pending.retain(|p| p.pack_id != job.pack_id)` added in `absorb_gated`
    /// above the `self.absorb_one(job)` call): "4 reader(s) were told a durable
    /// object was absent — **32 865 593** absent answers over 2687 objects".
    /// Restored. The count is that large because the readers spin until the bit
    /// goes up and, with the bit cleared first, every pass over the sample missed
    /// for the whole length of the absorb.
    ///
    /// The honest limit: this is a race, so a green run proves the window was not
    /// hit, not that no window exists. That is why the printed line carries how
    /// many of the reads actually landed while the pack was still un-indexed —
    /// **a run where that number is 0 proved nothing**, and it is reported rather
    /// than asserted because forcing it would mean pausing the drain, which is
    /// the one thing this test must not do. MEASURED 2026-08-08 on oden at
    /// loadavg 16.94: 336 of 336 reads landed early.
    #[test]
    fn no_read_is_told_absent_while_the_drain_is_running() {
        let dir = tmpdir("drain-during");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, rows) = real_pack();
        let oids: Vec<&[u8]> = rows.iter().map(|r| r.oid.as_slice()).collect();

        let tx = store.put_pack(&pack).unwrap();
        let pack_id = tx.pack_id.unwrap();

        let early = AtomicU64::new(0);
        let absent = AtomicU64::new(0);
        let reads = AtomicU64::new(0);
        let workers = 4usize;
        gatling_for_each(workers, workers, |w| loop {
            let still_pending = store.unindexed_packs() > 0;
            for oid in oids.iter().skip(w).step_by(workers * 8) {
                if still_pending {
                    early.fetch_add(1, AtomicOrdering::Relaxed);
                }
                reads.fetch_add(1, AtomicOrdering::Relaxed);
                if !store.has(oid).unwrap() {
                    absent.fetch_add(1, AtomicOrdering::Relaxed);
                }
            }
            if store.indexer().is_indexed(pack_id) {
                break;
            }
        });

        let absent = absent.load(AtomicOrdering::Acquire);
        assert_eq!(
            absent,
            0,
            "{workers} reader(s) were told a durable object was absent — {absent} absent answers \
             over {} objects",
            rows.len()
        );
        assert_eq!(store.object_count(), rows.len());
        eprintln!(
            "load {}; {} reads across {workers} gatling workers, {} of them while the pack was \
             still un-indexed",
            loadavg(),
            reads.load(AtomicOrdering::Acquire),
            early.load(AtomicOrdering::Acquire),
        );
    }

    /// **One pack is absorbed once, whichever of the two callers gets there
    /// first.**
    ///
    /// Both callers are made to attempt it, deterministically: with the gate
    /// held, the drain is parked inside `ObjectAbsorb::absorb` and a reader is
    /// parked inside `absorb_pending`, both past their own early exits. Releasing
    /// the gate lets exactly one do the work; the other must take the skip.
    ///
    /// The observable is the **commit graph**, not the object table:
    /// `ObjectReadStack::append` is keyed by oid and would swallow a second
    /// identical row, but the graph is a `Vec` and a second absorb pushes every
    /// commit into it again — which silently doubles the fold every reachability
    /// bitmap is built from.
    ///
    /// Seen RED by disabling the `s.absorbed.contains(&job.pack_id)` skip in
    /// `absorb_gated`: "the commit graph carries **1102** commits, the pack has
    /// **551**" — exactly double, both callers having done the whole fold.
    /// Restored.
    #[test]
    fn a_pack_is_absorbed_once_even_when_both_callers_race_for_it() {
        let dir = tmpdir("drain-once");
        let store = Arc::new(GitStore::open(&dir, "rickard").unwrap());
        let (pack, rows) = real_pack();

        let gate = store.hold_absorb_gate();
        store.put_pack(&pack).unwrap();

        let s = store.clone();
        let reader = Job::spawn(move || s.absorb_pending().unwrap());
        // Both are now blocked on the gate: the drain inside `absorb`, the reader
        // inside `absorb_pending`, each having already decided there is work.
        std::thread::sleep(Duration::from_millis(100));
        drop(gate);

        let by_read = reader.join().unwrap();
        store.wait_indexed();

        assert_eq!(
            store.commit_count(),
            commits_in(&rows),
            "the commit graph carries {} commits, the pack has {}",
            store.commit_count(),
            commits_in(&rows)
        );
        assert_eq!(store.object_count(), rows.len());
        assert_eq!(store.unindexed_packs(), 0);
        eprintln!(
            "load {}; the read absorbed {by_read} pack(s), the drain absorbed the rest",
            loadavg()
        );
    }

    // ── §14's exploded table ─────────────────────────────────────────────────

    /// **A push produces an exploded row for every object in the pack, and each
    /// one hashes back to its own oid.**
    ///
    /// Eager, decided 2026-08-08: no threshold, no size cap, no type skipped —
    /// so the row count is the *object* count and not some subset of it. Nothing
    /// is read between `put_pack` and `wait_indexed`, so the only thing that
    /// could have built the rows is the background drain, which is the half of
    /// §14 that says "built by the same background indexer".
    ///
    /// The assertion is applied output in its strongest available form: every
    /// object's content is read back out of the table, re-serialised canonically
    /// and **re-hashed**, and the hash has to be the oid it was filed under. A
    /// row count alone could be satisfied by 2687 rows of the wrong bytes; a
    /// re-hash cannot.
    ///
    /// Seen RED by deleting the `sink.explode(&oid, kind, &payload)?` call in
    /// `resolve::resolve_walked`: "the drain built no exploded rows: 0 of 2687".
    /// Restored.
    ///
    /// Seen RED a second time by making the sink skip blobs in
    /// `ExplodedTable::explode` (`if kind == GitObjectKind::Blob { return Ok(()) }`
    /// at the top — which is what a *lazy* or size-capped table would look like
    /// from here): "the drain built no exploded rows: 1420 of 2687". Restored,
    /// because §14's open question was closed EAGER and disk is explicitly not a
    /// consideration — and because the 1267 rows that edit removes are exactly
    /// the blobs a content read wants most.
    #[test]
    fn a_push_produces_an_exploded_row_for_every_object_in_the_pack() {
        let dir = tmpdir("exploded-every-object");
        let store = GitStore::open(&dir, "rickard").unwrap();
        let (pack, rows) = real_pack();

        let t = Instant::now();
        store.put_pack(&pack).unwrap();
        let ack = t.elapsed();
        store.wait_indexed();
        // **Drain, printed beside ack**, because the two answer different
        // questions and only one of them was ever in doubt. Ack is unchanged by
        // construction — nothing on it touches this table. Drain is where the
        // eager resolve and the sink both live, and it is the column redb's
        // 2026-08-08 measurement recorded as 316.2–329.8 ms against 166.7–168.6
        // ms with the sink stubbed out. Without it here, "did the medium buy
        // anything but disk" has no answer in this repository.
        let drain = t.elapsed();

        let distinct: HashSet<Vec<u8>> = rows.iter().map(|r| r.oid.clone()).collect();
        let stats = store.exploded_stats();
        assert_eq!(
            stats.rows,
            distinct.len() as u64,
            "the drain built no exploded rows: {} of {}",
            stats.rows,
            distinct.len()
        );
        assert_eq!(
            stats.written, stats.rows,
            "rows exist that this process never committed"
        );

        // Every one of them, re-hashed. Bytes that came back wrong cannot pass.
        let hash = store.hash_kind();
        let mut checked = 0usize;
        for oid in &distinct {
            let (kind, payload) = store
                .content(oid)
                .unwrap()
                .unwrap_or_else(|| panic!("{} has no exploded row", hex::encode(oid)));
            let rehashed = hash.oid_of(&crate::object::canonical(kind, &payload));
            assert_eq!(
                &rehashed,
                oid,
                "the exploded row filed under {} contains an object that hashes to {}",
                hex::encode(oid),
                hex::encode(&rehashed)
            );
            checked += 1;
        }
        assert_eq!(checked, distinct.len());
        // Nothing re-derived: the table answered all of them.
        assert_eq!(
            store.exploded_stats().rederived,
            0,
            "a content read re-resolved a pack even though the table was whole"
        );
        // **What it costs on disk, measured rather than assumed.** §14 says the
        // price is "both copies are held" and the decision was taken with "we
        // don't care if disk is tripled" — so the real multiple belongs in the
        // output where anybody can read it, not in a sentence. It is not 3x on a
        // real pack: the verbatim bytes are deflated *and* delta-encoded, the
        // table holds raw inflated content, and redb pages it.
        let inflated: u64 = rows.iter().map(|r| r.uncompressed_size).sum();
        let table = std::fs::metadata(store.exploded_path())
            .map(|m| m.len())
            .unwrap_or(0);
        eprintln!(
            "load {}; {} pack entries → {} exploded rows, all {} re-hashed to their own oid; \
             ack {:.1} ms, drain {:.1} ms, {:.0} rows/s; disk: pack {:.1} MiB verbatim, objects \
             inflate to {:.1} MiB, table file {:.1} MiB ({:.0}x the pack, {:.3}x the payload)",
            loadavg(),
            rows.len(),
            stats.rows,
            checked,
            ack.as_secs_f64() * 1e3,
            drain.as_secs_f64() * 1e3,
            stats.rows as f64 / drain.as_secs_f64(),
            pack.len() as f64 / (1 << 20) as f64,
            inflated as f64 / (1 << 20) as f64,
            table as f64 / (1 << 20) as f64,
            table as f64 / pack.len() as f64,
            table as f64 / inflated.max(1) as f64,
        );
    }

    /// **The amplification measurement, at whatever scale you point it at.**
    ///
    /// [`a_push_produces_an_exploded_row_for_every_object_in_the_pack`] prints
    /// the same ratio on a 5.4 MiB fixture and is a *guard*, so it has to stay
    /// fast. This is the same arithmetic with no assertion about size and no
    /// fixture of its own: give it a pack and it reports what the table cost.
    ///
    /// ```text
    /// ZNIPPY_EXPLODE_BENCH_PACK=/path/to/pack-….pack \
    ///   cargo test -p znippy-plugin-git --lib the_table_costs_what_it_holds \
    ///   -- --ignored --nocapture
    /// ```
    ///
    /// It exists because the 12× figure that killed redb — 16.8 GB of resolved
    /// `linux.git` content in a 204 GB file — was taken by hand off a bench run
    /// that no longer exists, and a number nobody can re-take is a number that
    /// rots. `#[ignore]`d and pack-less by default: with no pack named it
    /// returns, so it never fails a suite for the absence of a corpus nobody
    /// promised to keep.
    #[test]
    #[ignore = "needs a pack: ZNIPPY_EXPLODE_BENCH_PACK=<path> cargo test … -- --ignored --nocapture"]
    fn the_table_costs_what_it_holds() {
        let Ok(path) = std::env::var("ZNIPPY_EXPLODE_BENCH_PACK") else {
            eprintln!("no ZNIPPY_EXPLODE_BENCH_PACK named; nothing measured");
            return;
        };
        let pack = std::fs::read(&path).unwrap_or_else(|e| panic!("reading {path}: {e}"));
        let dir = tmpdir("exploded-cost");
        let store = GitStore::open(&dir, "rickard").unwrap();

        let t = Instant::now();
        store.put_pack(&pack).unwrap();
        let ack = t.elapsed();
        store.wait_indexed();
        let drain = t.elapsed();

        let stats = store.exploded_stats();
        let table = std::fs::metadata(store.exploded_path())
            .map(|m| m.len())
            .unwrap_or(0);
        // The payload the table actually holds, asked of the table rather than
        // of the pack — a pack entry's `uncompressed_size` is what git recorded,
        // and this is what we stored. Only the second one can be divided into
        // the file size and mean anything.
        let mut held = 0u64;
        for kind in [
            GitObjectKind::Commit,
            GitObjectKind::Tree,
            GitObjectKind::Blob,
            GitObjectKind::Tag,
        ] {
            held += store
                .exploded_of_kind(kind)
                .unwrap()
                .iter()
                .map(|(_, p)| p.len() as u64)
                .sum::<u64>();
        }
        let mib = |n: u64| n as f64 / (1 << 20) as f64;
        eprintln!(
            "load {}; pack {} — {:.1} MiB verbatim → {} rows holding {:.1} MiB, table file \
             {:.1} MiB = {:.3}x the payload and {:.1}x the pack; ack {:.0} ms, drain {:.0} ms",
            loadavg(),
            path,
            mib(pack.len() as u64),
            stats.rows,
            mib(held),
            mib(table),
            table as f64 / held.max(1) as f64,
            table as f64 / pack.len() as f64,
            ack.as_secs_f64() * 1e3,
            drain.as_secs_f64() * 1e3,
        );
    }

    /// Where the child of the clean-shutdown guard finds its fixture and leaves
    /// its markers.
    const CLEAN_DIR: &str = "GUNNAR_CLEAN_SHUTDOWN_DIR";

    /// **The other half of
    /// [`a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable`]: the process
    /// that exits normally.**
    ///
    /// A real process, a real push, a real drain, and then a real `drop` — every
    /// worker joined, every redb handle closed, no signal, exit status 0. That is
    /// what makes the parent's reopen a *clean* reopen rather than a recovery,
    /// and it is the case the crash guard cannot reach: after a clean shutdown
    /// every pack's bit is legitimately set and **nothing is re-queued**, so the
    /// derived tables have to have been on disk or they are gone.
    ///
    /// `#[ignore]`d because it is a fixture, not a guard — run on its own it has
    /// no directory to work in and returns.
    #[test]
    #[ignore = "spawned by a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable"]
    fn the_child_that_pushes_and_shuts_down_cleanly() {
        let Ok(dir) = std::env::var(CLEAN_DIR) else {
            return;
        };
        let dir = PathBuf::from(dir);
        let pack = std::fs::read(dir.join("fixture.pack")).expect("the parent's fixture");

        let store = GitStore::open(&dir, "rickard").expect("open");
        store.put_pack(&pack).expect("the ack path");
        store.wait_indexed();
        assert_eq!(store.unindexed_packs(), 0, "the child shut down mid-index");

        let tip = store
            .graph_snapshot()
            .into_iter()
            .max_by_key(|c| c.generation)
            .expect("the child built no graph at all");
        let tip_raw = hex::decode(&tip.oid).unwrap();
        let closure = store.reachable(&[&tip_raw], &[]).unwrap();
        assert!(!closure.is_empty(), "the child's own reachable() was empty");

        std::fs::write(dir.join("tip"), &tip.oid).expect("marker");
        std::fs::write(dir.join("closure"), closure.len().to_string()).expect("marker");
        std::fs::write(dir.join("commits"), store.commit_count().to_string()).expect("marker");
        std::fs::write(dir.join("objects"), store.object_count().to_string()).expect("marker");

        // **The clean shutdown.** Not a signal and not an abort: the store drops,
        // which closes the channel, joins the account indexer's worker and closes
        // redb. The process then returns normally with status 0.
        drop(store);
        std::fs::write(dir.join("clean"), b"closed").expect("marker");
    }

    /// **`commit_count()` and `reachable()` are correct after a CLEAN shutdown
    /// and reopen.** This is the guard the exploded table exists to make pass.
    ///
    /// It was the live bug: the commit graph was folded from commit and tree
    /// *payloads* which were stored nowhere, accumulated in RAM as packs were
    /// absorbed. A store killed mid-absorb got them back only because the
    /// interrupted pack was re-queued and re-resolved
    /// ([`a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack`]); a store
    /// that shut down **cleanly** re-queued nothing, and MEASURED on oden
    /// 2026-08-08 came back with 2687 rows and **0 of 551 commits** —
    /// `reachable()` returning a commit alone instead of its closure, and `gc`
    /// seeing an empty live set. Both are quiet wrong answers.
    ///
    /// The shape is the crash guard's: a **real** child process, which here exits
    /// normally, and the parent asserts against what is on disk. Two assertions
    /// carry it and they have to be in this order:
    ///
    /// 1. **`commit_count()` immediately after `open`**, before anything is
    ///    waited on — so the graph can only have come from the fold over the
    ///    exploded table that `open` performs, not from a re-absorb;
    /// 2. **`is_indexed(0)` is false after `wait_indexed()`** — the account
    ///    indexer in *this* process publishes a pack only for a job somebody
    ///    submitted, so a clear bit proves open re-queued nothing and the graph
    ///    above is not a recovery artefact.
    ///
    /// `reachable()` is then asserted against the closure the child computed on
    /// the same tip, so it is the same answer and not merely a non-empty one.
    ///
    /// Seen RED by deleting `store.refold()?` from the end of
    /// `GitStore::open_with_arms`, so nothing folds the table on open: "the
    /// commit graph did not survive a clean shutdown: 0 of 551 commits — the
    /// exploded table is a warm cache, not a durable table", left 0 right 551.
    /// That is the recorded bug's own number, reproduced. Restored.
    ///
    /// Seen RED a second time by deleting `self.exploded.flush()?` from
    /// `Absorber::absorb_one`, which is the **durability** edit: the rows then
    /// live only in the un-flushed buffer, which answers perfectly inside the
    /// child and is gone when the process ends. It fires assertion (2) rather
    /// than (1) — "the reopen re-queued the pack, so the graph above is a
    /// recovery artefact rather than the table's" — and that is worth writing
    /// down, because assertion (1) *passed*: the 64 MiB flush threshold had
    /// tripped once mid-pack, leaving 1798 of 2687 rows on disk which happened
    /// to include all 551 commits. A guard with only assertion (1) would have
    /// called that edit green. Restored.
    #[test]
    fn a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable() {
        let dir = tmpdir("clean-shutdown-graph");
        let (pack, rows) = real_pack();
        std::fs::write(dir.join("fixture.pack"), &pack).unwrap();

        let status = std::process::Command::new(std::env::current_exe().unwrap())
            .args([
                "--exact",
                "--ignored",
                "--nocapture",
                "git_ops::tests::the_child_that_pushes_and_shuts_down_cleanly",
            ])
            .env(CLEAN_DIR, &dir)
            .status()
            .expect("spawn the child that shuts down cleanly");
        assert!(
            status.success(),
            "the child did not exit cleanly — it {status:?}, so this is a crash guard and not a \
             clean-shutdown one"
        );
        assert!(
            dir.join("clean").exists(),
            "the child never reached its clean shutdown"
        );

        let tip_hex = std::fs::read_to_string(dir.join("tip")).unwrap();
        let closure_before: usize = std::fs::read_to_string(dir.join("closure"))
            .unwrap()
            .parse()
            .unwrap();
        let commits_before: usize = std::fs::read_to_string(dir.join("commits"))
            .unwrap()
            .parse()
            .unwrap();
        assert_eq!(
            commits_before,
            commits_in(&rows),
            "the child's own graph was already wrong, so the reopen proves nothing"
        );

        // ── the reopen, and (1): the graph before anything is waited on ──────
        let store = GitStore::open(&dir, "rickard").unwrap();
        assert_eq!(
            store.commit_count(),
            commits_in(&rows),
            "the commit graph did not survive a clean shutdown: {} of {} commits — the exploded \
             table is a warm cache, not a durable table",
            store.commit_count(),
            commits_in(&rows)
        );
        let tip_raw = hex::decode(&tip_hex).unwrap();
        let closure = store.reachable(&[&tip_raw], &[]).unwrap();
        assert_eq!(
            closure.len(),
            closure_before,
            "reachable() answers differently after a clean reopen: {} objects against the {} the \
             same tip closed over before the shutdown",
            closure.len(),
            closure_before
        );
        assert!(closure.contains(&tip_raw));

        // ── (2): and none of it came from a re-absorb ────────────────────────
        store.wait_indexed();
        assert!(
            !store.indexer().is_indexed(0),
            "the reopen re-queued the pack, so the graph above is a recovery artefact rather \
             than the table's"
        );
        assert_eq!(store.unindexed_packs(), 0);
        assert_eq!(
            store.absorb_pending().unwrap(),
            0,
            "the reopen left index work"
        );
        assert_eq!(store.object_count(), rows.len());
        assert_eq!(
            store.exploded_stats().written,
            0,
            "this process wrote exploded rows, so the table was rebuilt rather than read"
        );

        eprintln!(
            "load {}; clean shutdown → reopen: {} rows, {} commits, tip closure {} objects \
             (rebuilt nothing)",
            loadavg(),
            store.object_count(),
            store.commit_count(),
            closure.len(),
        );
    }

    /// **A content read is served from the table, not re-derived** — and the only
    /// thing that can say so is a counter.
    ///
    /// Both paths return the **identical** bytes: the table hands back what the
    /// resolver produced, and the fallback re-resolves the same pack and produces
    /// it again. A byte comparison therefore proves nothing at all about which
    /// one ran — this is the identity-value trap in its exact form, and it fired
    /// on this codebase once already. So the assertion is on
    /// [`ExplodedStats::served`] against [`ExplodedStats::rederived`], both of
    /// which are bumped where the work happens.
    ///
    /// The fallback is not asserted away: it is exercised in the same test, on a
    /// store whose table has been dropped, and the counters move the other way.
    /// A guard that only saw the fast path could not tell a working counter from
    /// one that is never incremented.
    ///
    /// Seen RED by deleting the `if let Some(hit) = self.exploded.content(oid)?`
    /// early return in `Absorber::resolved`, so every read re-derives: "512 content
    /// reads re-resolved a whole pack instead of hitting the table — served 0,
    /// rederived 512". The bytes were still correct on every one of them, which
    /// is the point. Restored.
    ///
    /// Seen RED a second time by bumping `served` unconditionally at the top of
    /// `ExplodedTable::content`, before the `pending`/redb probe — the counter
    /// then reports a hit whether or not the table had the object. The first half
    /// fires on the double count, "512 content reads re-resolved a whole pack
    /// instead of hitting the table — served 1024, rederived 0", and
    /// `exploded::tests::a_row_round_trips_and_the_kind_index_finds_it` fires with
    /// it ("a content read was not counted", left 5 right 2). Restored — a
    /// counter that cannot be wrong about a miss is the only kind worth
    /// asserting on.
    #[test]
    fn a_content_read_is_served_from_the_table_and_not_re_derived() {
        const READS: usize = 512;
        let dir = tmpdir("exploded-served");
        let (pack, rows) = real_pack();
        let oids: Vec<Vec<u8>> = rows
            .iter()
            .map(|r| r.oid.clone())
            .collect::<HashSet<Vec<u8>>>()
            .into_iter()
            .take(READS)
            .collect();

        {
            let store = GitStore::open(&dir, "rickard").unwrap();
            store.put_pack(&pack).unwrap();
            store.wait_indexed();

            let before = store.exploded_stats();
            let t = Instant::now();
            for oid in &oids {
                assert!(
                    store.content(oid).unwrap().is_some(),
                    "the table lost an object"
                );
            }
            let served_in = t.elapsed();
            let after = store.exploded_stats();
            assert_eq!(
                (
                    after.served - before.served,
                    after.rederived - before.rederived
                ),
                (oids.len() as u64, 0),
                "{} content reads re-resolved a whole pack instead of hitting the table — \
                 served {}, rederived {}",
                oids.len(),
                after.served - before.served,
                after.rederived - before.rederived,
            );
            eprintln!(
                "load {}; {} content reads from §14's table in {:.1} ms ({:.0} ns/read)",
                loadavg(),
                oids.len(),
                served_in.as_secs_f64() * 1e3,
                served_in.as_secs_f64() * 1e9 / oids.len() as f64,
            );
        }

        // ── and the fallback, so the counter is not a constant ───────────────
        //
        // The table is dropped and the *drain* is what would rebuild it, so the
        // reads have to happen before it gets there. `absorb_pending` is called
        // by hand first so the fallback under test is the content path's and not
        // the index's.
        std::fs::remove_file(dir.join("objects.exploded")).unwrap();
        let store = GitStore::open(&dir, "rickard").unwrap();
        let gate = store.hold_absorb_gate();
        let before = store.exploded_stats();
        let mut rederived = 0u64;
        for oid in oids.iter().take(8) {
            // Straight at the absorber: the store's `content` would absorb the
            // re-queued pack first and rebuild the very table this is testing
            // the absence of.
            assert!(
                store.absorber.resolved(oid).unwrap().is_some(),
                "the verbatim truth could not re-derive {}",
                hex::encode(oid)
            );
            rederived += 1;
        }
        let after = store.exploded_stats();
        drop(gate);
        assert_eq!(
            after.served - before.served,
            0,
            "a dropped table still reported {} reads served",
            after.served - before.served
        );
        assert_eq!(
            after.rederived - before.rederived,
            rederived,
            "the fallback ran but was not counted"
        );
    }

    /// **Dropping the table and reopening still answers correctly — fall back,
    /// then rebuild.**
    ///
    /// §14 calls the resolved table *droppable*: derived, verifiable against the
    /// verbatim truth, and deletable at any time without consulting a client.
    /// This is that sentence as an operation — `rm objects.exploded` between two
    /// opens — and §13.12's rule applied to it unchanged: **absent means fall
    /// back, never means wrong.**
    ///
    /// Three things are asserted, in this order:
    ///
    /// 1. the file really is gone and the reopened store really does start with
    ///    an empty table — otherwise the rest is testing a warm one;
    /// 2. **`reachable()` is already correct**, asked before any drain has been
    ///    waited on: the fall-back absorb runs inline and rebuilds what the
    ///    answer needs;
    /// 3. the table is **rebuilt** — one row per object again, the graph back,
    ///    and the rows written by *this* process, which is what says a rebuild
    ///    happened rather than a survival.
    ///
    /// Seen RED by `let exploded_is_whole = self.exploded.rows()? >= self.objects.len()
    /// as u64;` → `= true;` in `Absorber::adopt_journal`, so a dropped table is
    /// never noticed and nothing is re-queued. Assertion (2) fires first:
    /// **"reachable() after dropping the table: 1 objects, against 2683
    /// before"** — one object, the tip alone, which is the recorded bug's exact
    /// symptom ("`reachable` on a commit returns that commit alone rather than
    /// its closure") arrived at from the other direction. Restored.
    ///
    /// Seen RED a second time by `>=` → `>` in the same expression, which makes a
    /// *whole* table look short. This guard stays green — the rebuild is
    /// correct — and two others fire instead:
    /// [`a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable`]'s "the
    /// reopen re-queued the pack, so the graph above is a recovery artefact
    /// rather than the table's", and the pre-existing
    /// [`a_clean_reopen_requeues_nothing_and_the_next_push_gets_a_fresh_ordinal`]'s
    /// "a fully absorbed pack was re-queued on open — the diff is not exact".
    /// That pair is the point: one guard catches the table not being rebuilt when
    /// it should be, the others catch it being rebuilt when it should not.
    /// Restored.
    #[test]
    fn dropping_the_exploded_table_and_reopening_still_answers() {
        let dir = tmpdir("exploded-dropped");
        let (pack, rows) = real_pack();
        let tip_hex;
        let closure_before;
        {
            let store = GitStore::open(&dir, "rickard").unwrap();
            store.put_pack(&pack).unwrap();
            store.wait_indexed();
            let tip = store
                .graph_snapshot()
                .into_iter()
                .max_by_key(|c| c.generation)
                .expect("a graph");
            let tip_raw = hex::decode(&tip.oid).unwrap();
            closure_before = store.reachable(&[&tip_raw], &[]).unwrap().len();
            tip_hex = tip.oid;
        }

        // (1) — the drop.
        let table = dir.join("objects.exploded");
        assert!(table.exists(), "the store never built an exploded table");
        std::fs::remove_file(&table).unwrap();
        {
            let fresh = crate::exploded_arrow::ExplodedArchive::open(&table).unwrap();
            assert_eq!(fresh.rows().unwrap(), 0, "the drop did not drop anything");
        }
        // Opening an absent table does not create one — so this only has
        // anything to remove if the assertion above was inspecting a live file.
        let _ = std::fs::remove_file(&table);

        // (2) — a correct answer with the table gone.
        let store = GitStore::open(&dir, "rickard").unwrap();
        let tip_raw = hex::decode(&tip_hex).unwrap();
        let closure = store.reachable(&[&tip_raw], &[]).unwrap();
        assert_eq!(
            closure.len(),
            closure_before,
            "reachable() after dropping the table: {} objects, against {} before",
            closure.len(),
            closure_before
        );
        assert!(store.has(&rows[0].oid).unwrap());

        // (3) — and it is back.
        store.wait_indexed();
        let stats = store.exploded_stats();
        let distinct: HashSet<Vec<u8>> = rows.iter().map(|r| r.oid.clone()).collect();
        assert_eq!(
            stats.rows,
            distinct.len() as u64,
            "the dropped table was never rebuilt: {} of {} rows",
            stats.rows,
            distinct.len()
        );
        assert!(
            stats.written > 0,
            "the table came back without this process writing a row — it was never dropped"
        );
        assert_eq!(
            store.commit_count(),
            commits_in(&rows),
            "the graph did not come back with the table: {} of {}",
            store.commit_count(),
            commits_in(&rows)
        );
        assert_eq!(
            store.object_count(),
            rows.len(),
            "the rebuild lost index rows"
        );
        eprintln!(
            "load {}; dropped and rebuilt: {} rows written, {} commits, tip closure {} objects",
            loadavg(),
            stats.written,
            store.commit_count(),
            closure.len(),
        );
    }
}