znippy_plugin_git/

read_stack.rs

//! **One read stack, not three components.**
//!
//! A git object lookup in this crate goes through a single stack with three
//! surfaces, and the whole point of putting them in one type is that they cannot
//! be reasoned about — or drift — separately:
//!
//! 1. **The Ragnar `stree` is the fast path.** It is not a cache in front of an
//!    index; it *is* the index's key surface, [`crate::oid_index`]'s 8-byte-prefix
//!    static B-tree, reached through whichever [`ObjectIndex`] arm the stack was
//!    parameterised with. No third oid index is written here (LAW 5).
//! 2. **The Arrow tables are what it points into.** The stree resolves an oid to
//!    an ordinal; the ordinal addresses the five facts in the Arrow IPC columns
//!    of [`FourTables`](crate::index_layout::FourTables) or
//!    [`OneTableFourColumns`]. Ordinal and column live in the same object and are
//!    replaced together.
//! 3. **redb is what it falls through to, and redb always answers.** Every object
//!    ever appended is in redb. The stree+Arrow surface is a *projection* of a
//!    prefix of it, rebuilt on a trigger. So a stree miss is never an answer — it
//!    is a routing decision, and the row is fetched from redb. The single
//!    exception is not an exception to the rule but a proof of it: when the
//!    projection covers **every** row, "the stree missed" and "the repository
//!    does not have it" are the same statement, and the trip is skipped. See
//!    [`ObjectReadStack::projection_is_complete`].
//!
//! This is [`skade`'s arrangement](../../../../skade/src/static_index.rs) copied
//! rather than reinvented: `StaticIndex` (STree64 over snapshot ids) in front,
//! the redb commit log behind, `resolve_many` doing one pipelined tree pass and
//! then one redb transaction for the slots the tree did not fill. The names
//! differ because the keys differ (oids, not snapshot ids); the shape does not.
//!
//! ## Scope: one stack per REPOSITORY
//!
//! Not per account. A git negotiation — `have`/`want`, the push connectivity
//! check — is scoped to one repository, and an oid means nothing outside the
//! repository that stores it. [`ObjectReadStack::open`] therefore takes the path
//! of one repository's tail database.
//!
//! ## Why the projection may be incomplete and can still be trusted
//!
//! **The archive is append-only.** An object's five facts are written once and
//! never rewritten — [`ObjectReadStack::append`] *refuses* a rewrite that would
//! change them rather than accepting it (that refusal is what makes the rest of
//! this true, and it is asserted in
//! [`tests::an_append_only_violation_is_refused`]). Given that:
//!
//! * a **hit** in the stree is always valid — the row it points at cannot have
//!   become stale, because nothing may change it;
//! * a **miss** carries no information at all — it means "not in this
//!   projection", which is not the same as "not in this repository";
//! * so the projection is allowed to lag, and the only cost of lag is the redb
//!   fall-through on the rows it has not absorbed yet.
//!
//! That asymmetry is the entire reason a *trigger* is sound where a
//! rebuild-per-write would be ruinous.
//!
//! ## The rebuild trigger
//!
//! Two triggers, both in [`RebuildTriggers`], either one fires:
//!
//! * **misses** — lookups the stree missed **and redb answered**, against a
//!   threshold relative to the size of the projection. Genuine absences are
//!   counted separately in `absent` and deliberately do NOT count: a `have`
//!   negotiation is mostly misses, and if those drove the trigger, a busy
//!   read-only repository would rebuild for ever without a single new object
//!   having arrived.
//! * **volume** — stored object bytes appended since the last rebuild.
//!
//! Defaults, and the measurements they were derived from, are on
//! [`RebuildTriggers::DEFAULT_TAIL_HITS_PER_ROW`],
//! [`RebuildTriggers::DEFAULT_MIN_TAIL_HITS`] and
//! [`RebuildTriggers::DEFAULT_TAIL_BYTES`].
//!
//! ## Threads
//!
//! There are none here. A rebuild runs inline on the [`append`](ObjectReadStack::append)
//! that trips the trigger, or on an explicit [`rebuild`](ObjectReadStack::rebuild).
//! skade rebuilds on a detached thread; the sanctioned home for a detached thread
//! in this constellation is `gatling::background::Job` (LAW 3 — rayon is banned
//! and so is a hand-rolled pool), and `Job` is join-required with no completion
//! probe, so "in the background" would mean picking an arbitrary later call to
//! block on. Inline on a trigger is deterministic and, with the defaults below,
//! rare. An owner who wants the overlap can drive [`rebuild`](ObjectReadStack::rebuild)
//! from a `Job` itself — it takes `&self` and swaps under an `RwLock`.

use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};

use anyhow::{Context, Result, anyhow, bail};
use redb::{Database, ReadableTable, ReadableTableMetadata, TableDefinition};

use crate::index_layout::{IndexEntry, IndexRow, ObjType, ObjectIndex, OneTableFourColumns};

// ── the durable tail ──────────────────────────────────────────────────────────

/// `oid → packed row`. The key is the raw oid (20 or 32 bytes), so redb's own
/// B-tree order is oid-lexicographic order — the same order the stree and the
/// Arrow columns are in, which makes a rebuild a straight ordered scan.
const OBJECTS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("objects");

/// Small counters. `arrival_seq` is the only one that must survive a reopen: it
/// is what gives a tail row an ordinal that no other tail row shares.
const META: TableDefinition<&str, u64> = TableDefinition::new("meta");
const META_ARRIVAL_SEQ: &str = "arrival_seq";

/// `seq | offset | len | type | uncompressed_size | delta_base`, little-endian,
/// fixed width.
///
/// ```text
///   0  u64  seq (arrival, not a fact about the object)
///   8  u64  offset
///  16  u64  len
///  24  u8   object type code
///  25  u64  uncompressed_size
///  33  u64  delta_base — archive offset of the base entry, 0 for none
///                                                                     ══
///                                                                     41
/// ```
///
/// **This was 33 bytes before `delta_base` (PLAN §13).** A tail written by the
/// older encoding is refused by [`decode_row`] on its length rather than
/// misread: the tail is a cache the whole of §14 permits dropping and
/// rebuilding, so a hard refusal is the right failure and a silent
/// reinterpretation is not.
const TAIL_ROW_BYTES: usize = 41;

/// Set on the ordinal of every row that came from the tail rather than from the
/// Arrow projection.
///
/// An ordinal is a **row address within one generation**, not an identity — the
/// oid is the identity. A rebuild re-derives every ordinal as the new
/// oid-lexicographic rank, so an ordinal held across a rebuild is meaningless.
/// The high bit makes the dangerous version of that mistake impossible instead
/// of merely documented: a tail ordinal used to index an Arrow column is ≥ 2^31
/// and therefore out of bounds — a panic — rather than a silently wrong row.
pub const TAIL_ORDINAL_BIT: u32 = 0x8000_0000;

/// True when `row` was answered by the redb tail rather than by the projection.
pub fn is_tail_row(row: &IndexRow) -> bool {
    row.ordinal & TAIL_ORDINAL_BIT != 0
}

fn tail_ordinal(seq: u64) -> u32 {
    TAIL_ORDINAL_BIT | (seq as u32 & !TAIL_ORDINAL_BIT)
}

fn encode_row(seq: u64, e: &IndexEntry) -> [u8; TAIL_ROW_BYTES] {
    let mut b = [0u8; TAIL_ROW_BYTES];
    b[0..8].copy_from_slice(&seq.to_le_bytes());
    b[8..16].copy_from_slice(&e.offset.to_le_bytes());
    b[16..24].copy_from_slice(&e.len.to_le_bytes());
    b[24] = e.obj_type.code();
    b[25..33].copy_from_slice(&e.uncompressed_size.to_le_bytes());
    b[33..41].copy_from_slice(&e.delta_base.to_le_bytes());
    b
}

/// One decoded tail row. `seq` is the arrival sequence, not a fact about the
/// object.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct TailRow {
    seq: u64,
    offset: u64,
    len: u64,
    obj_type: ObjType,
    uncompressed_size: u64,
    /// Archive offset of the base entry, `0` for none. An offset and not an
    /// ordinal precisely because of this table: a tail ordinal is
    /// [`TAIL_ORDINAL_BIT`]-tagged and a rebuild re-derives every ordinal, so an
    /// ordinal stored here would be meaningless one rebuild later.
    delta_base: u64,
}

fn decode_row(b: &[u8]) -> Result<TailRow> {
    if b.len() != TAIL_ROW_BYTES {
        bail!("tail row is {} bytes, expected {TAIL_ROW_BYTES}", b.len());
    }
    let u64_at = |i: usize| {
        let mut w = [0u8; 8];
        w.copy_from_slice(&b[i..i + 8]);
        u64::from_le_bytes(w)
    };
    Ok(TailRow {
        seq: u64_at(0),
        offset: u64_at(8),
        len: u64_at(16),
        // A code the writer cannot produce must not be invented on the way out:
        // a wrong type in an index is worse than a failure, because it is
        // queried and believed.
        obj_type: ObjType::from_code(b[24])
            .ok_or_else(|| anyhow!("tail row carries object type code {}", b[24]))?,
        uncompressed_size: u64_at(25),
        delta_base: u64_at(33),
    })
}

impl TailRow {
    fn as_index_row(&self) -> IndexRow {
        IndexRow {
            ordinal: tail_ordinal(self.seq),
            offset: self.offset,
            len: self.len,
            obj_type: self.obj_type,
            uncompressed_size: self.uncompressed_size,
            delta_base: self.delta_base,
        }
    }

    /// The same facts as an [`IndexEntry`], for the rebuild scan.
    fn as_entry(&self, oid: &[u8]) -> IndexEntry {
        IndexEntry {
            oid: oid.to_vec(),
            offset: self.offset,
            len: self.len,
            obj_type: self.obj_type,
            uncompressed_size: self.uncompressed_size,
            delta_base: self.delta_base,
        }
    }
}

// ── the trigger ───────────────────────────────────────────────────────────────

/// When the Arrow/stree projection is rebuilt from the redb tail. Either
/// trigger fires; both are per repository and per generation.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct RebuildTriggers {
    /// The miss trigger, as a multiple of the rows already in the projection.
    /// See [`DEFAULT_TAIL_HITS_PER_ROW`](Self::DEFAULT_TAIL_HITS_PER_ROW) for why
    /// it is relative and not a constant.
    pub tail_hits_per_row: f64,
    /// Floor and off-switch for the miss trigger: the threshold is never below
    /// this, and **`0` disables the miss trigger entirely** whatever
    /// `tail_hits_per_row` says.
    pub min_tail_hits: u64,
    /// Rebuild once this many stored object bytes have been appended since the
    /// last rebuild. `0` disables the volume trigger.
    pub tail_bytes: u64,
}

impl RebuildTriggers {
    /// **1.0 — one tail-served lookup per row in the projection.**
    ///
    /// The miss trigger is relative because the two costs it balances scale
    /// differently, and a single constant is therefore wrong at one end or the
    /// other. Both costs MEASURED on oden 2026-08-07, 1-min loadavg 2.1–2.2,
    /// `examples/read_stack_bench.rs`, sha1 oids, batch 1000, 3 runs per cell,
    /// worst run-to-run spread 24.9%:
    ///
    /// | | 100 000 objects | 1 000 000 objects |
    /// |---|---:|---:|
    /// | lookup the projection answers | 80 ns | 221 ns |
    /// | lookup the tail answers | 463 ns | 937 ns |
    /// | **the fall-through costs** | **383 ns** | **716 ns** |
    /// | rebuild (ordered tail scan + projection build) | 40.9 ms | 547.9 ms |
    /// | **per row in the repository** | **409 ns** | **548 ns** |
    /// | break-even, tail-served lookups per row | **1.07** | **0.77** |
    ///
    /// The rebuild has paid for itself once the fall-through has carried roughly
    /// one lookup per row, at both sizes and an order of magnitude apart — which
    /// is why the ratio is the right shape for this trigger and **1.0** is the
    /// right value in it. A fixed 4096 would rebuild a million-object repository
    /// (0.55 s) to save 4096 × 716 ns ≈ 2.9 ms — 190× the wrong way.
    ///
    /// Re-measure if the tail engine or the projection build changes: this is a
    /// ratio of two measured costs and nothing else.
    pub const DEFAULT_TAIL_HITS_PER_ROW: f64 = 1.0;

    /// **4096 — the floor under the relative threshold.**
    ///
    /// On an empty or nearly-empty projection the ratio above is ~0 and would
    /// rebuild on the first miss, over and over, during exactly the period when
    /// objects are still arriving. 4096 misses is ~1.6 ms of fall-through at the
    /// measured 100 000-object price — cheap enough to be worth waiting for on
    /// any repository, and enough that a burst of small pushes coalesces into
    /// one rebuild.
    pub const DEFAULT_MIN_TAIL_HITS: u64 = 4096;

    /// **64 MiB of appended object bytes.**
    ///
    /// A git push is a packfile, and the pack is the unit that lands in the tail.
    /// 64 MiB is comfortably more than one ordinary push and less than a big
    /// one, so a busy repository rebuilds on the order of once per large push
    /// rather than once per push — while a repository taking a 2 GiB initial
    /// import rebuilds ~32 times over that import instead of once at the end,
    /// which is what keeps the fall-through from carrying the whole import.
    /// It is the trigger that carries the normal case; the miss trigger is the
    /// backstop for a repository that is read hard and written rarely.
    pub const DEFAULT_TAIL_BYTES: u64 = 64 * 1024 * 1024;

    /// The miss threshold for a projection of `projection_rows` rows, or `None`
    /// when the miss trigger is off.
    pub fn miss_threshold(&self, projection_rows: u64) -> Option<u64> {
        if self.min_tail_hits == 0 {
            return None;
        }
        let scaled = (self.tail_hits_per_row * projection_rows as f64) as u64;
        Some(scaled.max(self.min_tail_hits))
    }

    /// Both triggers off: the projection is rebuilt only when asked.
    pub fn manual() -> Self {
        Self {
            tail_hits_per_row: 0.0,
            min_tail_hits: 0,
            tail_bytes: 0,
        }
    }
}

impl Default for RebuildTriggers {
    fn default() -> Self {
        Self {
            tail_hits_per_row: Self::DEFAULT_TAIL_HITS_PER_ROW,
            min_tail_hits: Self::DEFAULT_MIN_TAIL_HITS,
            tail_bytes: Self::DEFAULT_TAIL_BYTES,
        }
    }
}

/// Which threshold fired, with the value that fired it. Returned rather than
/// logged, so a caller can record *why* a rebuild happened.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RebuildReason {
    /// `n` lookups had to be answered by the tail.
    TailHits(u64),
    /// `n` bytes of objects were appended since the last rebuild.
    TailBytes(u64),
    /// [`ObjectReadStack::rebuild`] was called directly.
    Explicit,
}

/// A snapshot of the stack's counters. Everything here is applied output — rows
/// that exist, lookups that happened — not configuration echoed back.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct StackStats {
    /// Objects in the Arrow/stree projection.
    pub sealed_rows: u64,
    /// Objects in the repository. Always ≥ `sealed_rows`; the difference is the
    /// un-absorbed tail.
    pub total_rows: u64,
    /// Lookups the stree missed and redb answered, since the last rebuild.
    pub tail_hits: u64,
    /// Lookups nothing answered — the object is not in this repository. Never
    /// drives the trigger.
    pub absent: u64,
    /// Rows the projection does not cover yet. **Zero means the projection *is*
    /// the truth**, which is what lets a miss be answered without a redb round
    /// trip at all.
    pub unabsorbed_rows: u64,
    /// redb read transactions opened by the read path since the last rebuild.
    /// The number that shows whether the completeness fast path is actually
    /// being taken; there is no other way to see it from outside.
    pub tail_txns: u64,
    /// Object bytes appended since the last rebuild.
    pub tail_bytes: u64,
    /// Projections built over this stack's lifetime, including the one at open.
    pub rebuilds: u64,
    /// Bumped by every rebuild. An [`IndexRow::ordinal`] is only meaningful
    /// within one generation.
    pub generation: u64,
}

// ── the stack ─────────────────────────────────────────────────────────────────

/// The read stack: Ragnar stree → Arrow columns → redb, in one object.
///
/// Parameterised by the Arrow layout so it drops into
/// `examples/index_layout_bench.rs` against
/// [`FourTables`](crate::index_layout::FourTables) and [`OneTableFourColumns`]
/// unchanged — it implements the same [`ObjectIndex`] trait, and
/// [`ObjectIndex::build`] produces a stack whose projection already covers every
/// entry, so on that benchmark it is those arms plus one redb round trip per
/// genuine miss.
pub struct ObjectReadStack<S: ObjectIndex = OneTableFourColumns> {
    /// The durable truth. Every object ever appended is here.
    tail: Arc<Database>,
    /// The projection. Replaced wholesale by a rebuild, never mutated in place —
    /// which is what lets a reader hold it briefly without coordinating.
    projection: RwLock<Arc<S>>,
    triggers: RebuildTriggers,
    total_rows: AtomicU64,
    tail_hits: AtomicU64,
    absent: AtomicU64,
    tail_bytes: AtomicU64,
    rebuilds: AtomicU64,
    generation: AtomicU64,
    /// Uncompressed bytes of the rows the projection has **not** absorbed.
    ///
    /// This is what makes [`ObjectIndex::sum_uncompressed`] exact without a scan.
    /// The projection is, by definition, every row that existed at the last
    /// rebuild; append-only means none of them can have changed since; so the
    /// repository's total is the projection's column scan plus the rows appended
    /// after it, and those are counted here as they arrive. Reset by a rebuild,
    /// which is the moment they stop being un-absorbed.
    unabsorbed_size: AtomicU64,
    /// The same trick for [`ObjectIndex::count_type`], indexed by
    /// [`ObjType::code`] (1–4, 6, 7; slots 0 and 5 stay zero).
    unabsorbed_types: [AtomicU64; 8],
    /// Rows the projection does not cover. **Zero is the interesting value**:
    /// see [`ObjectReadStack::projection_is_complete`].
    unabsorbed_rows: AtomicU64,
    /// redb read transactions the read path has opened since the last rebuild.
    tail_txns: AtomicU64,
}

impl<S: ObjectIndex> ObjectReadStack<S> {
    /// Open (or create) one repository's stack at `tail_path`, warm-starting the
    /// projection from everything already in the tail.
    /// `cache_bytes` is redb's page-cache ceiling for **this** database. A
    /// parameter and not a `getenv` down here, so the whole store costs one
    /// environment read at construction and none below it; see
    /// [`crate::arms::redb_cache_bytes`], and note that the stock
    /// `Database::create` would take redb's 1 GiB default instead.
    pub fn open(tail_path: &Path, triggers: RebuildTriggers, cache_bytes: usize) -> Result<Self> {
        let db = Database::builder()
            .set_cache_size(cache_bytes)
            .create(tail_path)
            .with_context(|| format!("opening object tail at {}", tail_path.display()))?;
        Self::from_db(db, triggers)
    }

    /// **The name the built projection reports about itself** — the applied
    /// output that says which [`crate::arms::IndexArm`] actually got built.
    ///
    /// Deliberately not [`ObjectIndex::name`] on the stack, which answers
    /// `"ObjectReadStack"` for every arm because that is what the stack is. `S`
    /// is the layout, and only `S` can say which one it is. It is the value
    /// [`crate::arms::IndexArm::projection_name`] is written to be compared
    /// against, so a caller that selected an arm can prove the selection took
    /// rather than echo the selector back to itself.
    pub fn projection_name(&self) -> &'static str {
        self.projection.read().expect("projection lock").name()
    }

    /// A stack whose tail has no file behind it. Used by [`ObjectIndex::build`]
    /// and by benchmarks; the redb code path is identical, only the backend
    /// differs, so a test of this stack is a test of the durable one.
    pub fn in_memory(triggers: RebuildTriggers) -> Result<Self> {
        let db = Database::builder()
            .create_with_backend(redb::backends::InMemoryBackend::new())
            .context("creating an in-memory object tail")?;
        Self::from_db(db, triggers)
    }

    fn from_db(db: Database, triggers: RebuildTriggers) -> Result<Self> {
        // Materialise both tables so a read on a brand-new database does not
        // trip `TableDoesNotExist`.
        let w = db.begin_write()?;
        {
            let _ = w.open_table(OBJECTS)?;
            let _ = w.open_table(META)?;
        }
        w.commit()?;

        let db = Arc::new(db);
        let entries = scan(&db)?;
        let total = entries.len() as u64;
        let projection = Arc::new(S::build(&entries)?);
        Ok(Self {
            tail: db,
            projection: RwLock::new(projection),
            triggers,
            total_rows: AtomicU64::new(total),
            tail_hits: AtomicU64::new(0),
            absent: AtomicU64::new(0),
            tail_bytes: AtomicU64::new(0),
            rebuilds: AtomicU64::new(1),
            generation: AtomicU64::new(1),
            unabsorbed_size: AtomicU64::new(0),
            unabsorbed_types: Default::default(),
            unabsorbed_rows: AtomicU64::new(0),
            tail_txns: AtomicU64::new(0),
        })
    }

    /// **The projection covers every row in the repository**, so a miss in the
    /// stree *is* an absence and the tail need not be asked.
    ///
    /// Sound by construction, not by hope: the projection is built from a scan
    /// of the tail, so its rows are always a subset of the tail's; the archive
    /// is append-only, so no row ever leaves; therefore equal counts mean equal
    /// sets. A concurrent append between this check and the lookup only means
    /// the lookup answers as of an instant before that append, which is what a
    /// lookup issued a microsecond earlier would have done anyway.
    ///
    /// This is worth the paragraph because of what it buys. MEASURED on oden
    /// 2026-08-07 (`read_stack_bench`, 100 000 objects, batch 1000): a `have`
    /// negotiation — 10% hit, so 90% of the oids are in no repository at all —
    /// cost **544 ns** per oid when every one of those absences took a redb
    /// round trip, against **77 ns** for the bare Arrow arm. With this fast path
    /// the same workload costs **68 ns**, i.e. the Arrow arm's own cost inside
    /// the noise band. A `have` negotiation is *mostly* misses, so without this
    /// the stack would be the dominant cost of the most common operation a git
    /// server performs.
    #[inline]
    pub fn projection_is_complete(&self) -> bool {
        self.unabsorbed_rows.load(Ordering::Acquire) == 0
    }

    /// Append objects to the tail.
    ///
    /// **Append-only is enforced here, not assumed.** An oid already present with
    /// identical facts is a no-op (a re-pushed pack repeats objects, and that is
    /// normal); an oid already present with *different* facts is an error and
    /// nothing in the batch is written. Without that refusal a stree hit could
    /// return a superseded row and the projection would be not merely incomplete
    /// but wrong, which is the one thing this design must not allow.
    ///
    /// Runs a rebuild inline if the batch trips a threshold; the reason is
    /// returned so the caller can see that it happened.
    pub fn append(&self, entries: &[IndexEntry]) -> Result<Option<RebuildReason>> {
        if entries.is_empty() {
            return Ok(None);
        }
        let width = entries[0].oid.len();
        if width != 20 && width != 32 {
            bail!("oid width {width} is neither sha1 (20) nor sha256 (32)");
        }

        let w = self.tail.begin_write()?;
        let mut added = 0u64;
        let mut bytes = 0u64;
        // Accumulated locally and applied only after the commit: a `bail!`
        // half-way through the batch rolls redb back, and counters bumped inside
        // the loop would survive a rollback and put the column-scan aggregates
        // permanently out of step with the tail.
        let mut added_size = 0u64;
        let mut added_types = [0u64; 8];
        {
            let mut objects = w.open_table(OBJECTS)?;
            let mut meta = w.open_table(META)?;
            let mut seq = meta.get(META_ARRIVAL_SEQ)?.map(|v| v.value()).unwrap_or(0);
            for e in entries {
                if e.oid.len() != width {
                    bail!(
                        "mixed oid widths in one append: {width} and {}",
                        e.oid.len()
                    );
                }
                if let Some(existing) = objects.get(e.oid.as_slice())? {
                    let old = decode_row(existing.value())?;
                    // ── IDENTITY, not PLACEMENT ─────────────────────────
                    //
                    // `offset`, `len` and `delta_base` say WHERE this copy of
                    // the object landed. They are not what it IS. A re-pushed
                    // pack legitimately places the same oid somewhere else —
                    // that is the ordinary case, not a violation — so comparing
                    // them refused a normal push.
                    //
                    // MEASURED 2026-08-14: pushing an object the repository
                    // already held aborted the ingest and dropped the
                    // connection with `client_told=false`, so the client saw
                    // "send-pack: unexpected disconnect" and no reason at all.
                    // It bit `bare_lifecycle` and `endurance` on all four
                    // znippy columns of the bench, and it bites any real
                    // `--force` push that resends a known object.
                    //
                    // The guard's own doc three lines up already said what it
                    // meant to do — "An oid already present with identical
                    // facts is a no-op (a re-pushed pack repeats objects, and
                    // that is normal)" — so the code contradicted its contract.
                    //
                    // What the original comparison was protecting is still
                    // protected, and by a stronger argument: we KEEP THE OLD
                    // ROW. The archive is append-only and `gc` truncates
                    // nothing, so the offset and delta base already recorded
                    // still address bytes that are still there. The projection
                    // therefore never carries a base the archive did not write
                    // — the first writer's placement stands, forever.
                    //
                    // What CANNOT be waved through is a disagreement about the
                    // object itself. Same oid, different type or different
                    // inflated size, means either a hash collision or a
                    // corrupted row, and no push may quietly overwrite that.
                    // …and `obj_type` is a REPRESENTATION, not the object's
                    // kind. `ofs-delta` and `ref-delta` say how these bytes are
                    // ENCODED; what the object IS lives at the end of the delta
                    // chain. The same oid legitimately arrives whole in one
                    // pack and delta-encoded in another, so comparing those two
                    // is comparing encodings and calling them different
                    // objects. MEASURED here, `gunnar.import_export`, znippy:
                    //
                    //   6da81e5e… is already stored as (…, ofs-delta, size 2084)
                    //   and cannot be redeclared as (…, tree, size 2084)
                    //
                    // — the inflated size agrees to the byte, which is the
                    // identity; only the encoding moved. The refusal poisoned
                    // the repository (`it stays un-indexed; reads will keep
                    // falling back`) and every later read answered `the
                    // repository is unavailable; try again`.
                    //
                    // So a type disagreement is a violation only when BOTH
                    // rows name a real object kind. The inflated size is
                    // compared unconditionally, and it is the half that
                    // actually catches a collision or a corrupt row.
                    let encoded = |t: ObjType| matches!(t, ObjType::OfsDelta | ObjType::RefDelta);
                    let kind_disagrees = !encoded(old.obj_type)
                        && !encoded(e.obj_type)
                        && old.obj_type != e.obj_type;
                    if kind_disagrees || old.uncompressed_size != e.uncompressed_size {
                        bail!(
                            "identity violation: {} is already stored as \
                             (offset {}, len {}, {}, size {}, delta_base {}) and cannot be \
                             redeclared as (offset {}, len {}, {}, size {}, delta_base {}) — \
                             the TYPE or the inflated SIZE differs, so the same oid is \
                             describing a different object. Placement (offset, len, \
                             delta_base) may differ freely: a re-pushed pack lands its \
                             copies elsewhere and the first writer's row stands.",
                            hex::encode(&e.oid),
                            old.offset,
                            old.len,
                            old.obj_type.as_str(),
                            old.uncompressed_size,
                            old.delta_base,
                            e.offset,
                            e.len,
                            e.obj_type.as_str(),
                            e.uncompressed_size,
                            e.delta_base,
                        );
                    }
                    continue;
                }
                objects.insert(e.oid.as_slice(), encode_row(seq, e).as_slice())?;
                seq += 1;
                added += 1;
                bytes += e.len;
                added_size += e.uncompressed_size;
                added_types[e.obj_type.code() as usize] += 1;
            }
            meta.insert(META_ARRIVAL_SEQ, seq)?;
        }
        w.commit()?;

        self.total_rows.fetch_add(added, Ordering::AcqRel);
        self.unabsorbed_rows.fetch_add(added, Ordering::AcqRel);
        self.unabsorbed_size.fetch_add(added_size, Ordering::AcqRel);
        for (slot, n) in self.unabsorbed_types.iter().zip(added_types) {
            slot.fetch_add(n, Ordering::AcqRel);
        }
        let total_bytes = self.tail_bytes.fetch_add(bytes, Ordering::AcqRel) + bytes;
        if self.triggers.tail_bytes != 0 && total_bytes >= self.triggers.tail_bytes {
            self.rebuild()?;
            return Ok(Some(RebuildReason::TailBytes(total_bytes)));
        }
        self.maybe_rebuild()
    }

    /// Which threshold, if any, is currently tripped. Pure read; a caller on a
    /// read-only repository can poll this from its own maintenance tick.
    pub fn rebuild_due(&self) -> Option<RebuildReason> {
        let hits = self.tail_hits.load(Ordering::Acquire);
        if let Some(threshold) = self.triggers.miss_threshold(self.projection_len() as u64)
            && hits >= threshold
        {
            return Some(RebuildReason::TailHits(hits));
        }
        let bytes = self.tail_bytes.load(Ordering::Acquire);
        if self.triggers.tail_bytes != 0 && bytes >= self.triggers.tail_bytes {
            return Some(RebuildReason::TailBytes(bytes));
        }
        None
    }

    /// Rebuild if [`rebuild_due`](Self::rebuild_due) says so.
    pub fn maybe_rebuild(&self) -> Result<Option<RebuildReason>> {
        match self.rebuild_due() {
            Some(reason) => {
                self.rebuild()?;
                Ok(Some(reason))
            }
            None => Ok(None),
        }
    }

    /// Rebuild the Arrow/stree projection from a full ordered scan of the tail
    /// and swap it in, resetting the counters and bumping the generation.
    ///
    /// The old projection stays live for every reader until the swap; the write
    /// lock is held only for the pointer store, not for the build.
    pub fn rebuild(&self) -> Result<()> {
        let entries = scan(&self.tail)?;
        let total = entries.len() as u64;
        let fresh = Arc::new(S::build(&entries)?);
        *self
            .projection
            .write()
            .map_err(|_| anyhow!("the projection lock is poisoned"))? = fresh;
        self.total_rows.store(total, Ordering::Release);
        self.tail_hits.store(0, Ordering::Release);
        self.tail_bytes.store(0, Ordering::Release);
        self.unabsorbed_size.store(0, Ordering::Release);
        self.unabsorbed_rows.store(0, Ordering::Release);
        self.tail_txns.store(0, Ordering::Release);
        for slot in &self.unabsorbed_types {
            slot.store(0, Ordering::Release);
        }
        self.rebuilds.fetch_add(1, Ordering::AcqRel);
        self.generation.fetch_add(1, Ordering::AcqRel);
        Ok(())
    }

    pub fn stats(&self) -> StackStats {
        StackStats {
            sealed_rows: self.projection_len() as u64,
            total_rows: self.total_rows.load(Ordering::Acquire),
            tail_hits: self.tail_hits.load(Ordering::Acquire),
            absent: self.absent.load(Ordering::Acquire),
            tail_bytes: self.tail_bytes.load(Ordering::Acquire),
            rebuilds: self.rebuilds.load(Ordering::Acquire),
            generation: self.generation.load(Ordering::Acquire),
            unabsorbed_rows: self.unabsorbed_rows.load(Ordering::Acquire),
            tail_txns: self.tail_txns.load(Ordering::Acquire),
        }
    }

    /// Rows in the projection right now. `stats().total_rows - this` is the
    /// un-absorbed tail.
    pub fn projection_len(&self) -> usize {
        self.projection.read().expect("projection lock").len()
    }

    /// Every oid in the repository, in oid order — which is ordinal order, since
    /// redb keys the tail by the raw oid.
    ///
    /// A full scan, and it is here for the one caller that genuinely needs the
    /// whole set at once: a GC, which has to name what is *not* live.
    pub fn oids_in_order(&self) -> Result<Vec<Vec<u8>>> {
        let read = self.tail.begin_read()?;
        let objects = read.open_table(OBJECTS)?;
        let mut out = Vec::with_capacity(objects.len()? as usize);
        for row in objects.iter()? {
            let (k, _) = row?;
            out.push(k.value().to_vec());
        }
        Ok(out)
    }

    /// **Which of these archive extents already have rows here** — the
    /// crash-recovery diff, and the reason §13.12's `indexed` bit is derived
    /// rather than stored.
    ///
    /// A pack is unabsorbed **iff its extent is in the journal and its rows are
    /// not in the index**. Both of those are already durable — the journal is
    /// fsynced on the ack path, the tail is a redb commit — so the bit is a diff
    /// of two durable facts and there is nothing for it to drift from. A *stored*
    /// bit would be a third fact that can disagree with the two it describes,
    /// which is precisely how a pack ends up marked absorbed with no rows behind
    /// it: fast and wrong, in the one direction (`absent`) that loses a client's
    /// objects during negotiation.
    ///
    /// `out[i]` answers `extents[i]`. **The tail and not the projection**, because
    /// the projection is a snapshot of a prefix of the tail and the question is
    /// about what survived the crash.
    ///
    /// # Cost
    ///
    /// One ordered scan of the tail that **stops the moment every extent has been
    /// hit**. Rows arrive in oid order, which is uncorrelated with the pack an
    /// object came from, so a store whose packs are all absorbed answers after a
    /// few rows per pack rather than after a full scan — the coupon-collector
    /// case, and the one every clean reopen takes. The scan only runs to the end
    /// when some pack genuinely has **no** rows, which is exactly the case where
    /// the caller is about to re-resolve that whole pack anyway.
    ///
    /// MEASURED on oden 2026-08-08, release, file-backed redb tail, 1-minute
    /// loadavg 7.1–7.5 (another tenant's work — read these as ratios):
    ///
    /// | packs | rows | every pack absorbed | one pack un-absorbed |
    /// |---:|---:|---:|---:|
    /// | 1 000 | 1 000 000 | **0.77 ms** | 120.6 ms |
    /// | 10 000 | 1 000 000 | **11.7 ms** | 141.8 ms |
    /// | 1 000 | 3 686 | **0.40 ms** | 0.39 ms |
    ///
    /// The first column is what a clean reopen pays and it is bounded by the
    /// **pack** count, not the object count — 1000 packs over a million objects
    /// costs the same order as 1000 packs over four thousand. The second column
    /// is the crash path, where the scan runs to the end: 120 ms over a million
    /// rows, against the ~160 ms a *single* 2687-object pack takes to re-absorb
    /// (`a_push_ends_in_object_rows_and_no_read_asked_for_them`). The diff is
    /// therefore never the expensive half of a recovery.
    /// The third row is the shape
    /// [`crate::git_ops::tests::the_derive_on_open_diff_costs_milliseconds_at_a_realistic_pack_count`]
    /// asserts on every run, over real pushed packs rather than synthetic rows.
    ///
    /// A zero-length extent has no rows by construction and is answered `false`
    /// without looking.
    pub fn extents_with_rows(&self, extents: &[(u64, u64)]) -> Result<Vec<bool>> {
        let mut hit = vec![false; extents.len()];
        // Sorted by start, so one row is placed with a binary search instead of a
        // pass over every extent: the scan below is the hot loop.
        let mut order: Vec<usize> = (0..extents.len()).filter(|&i| extents[i].1 > 0).collect();
        order.sort_unstable_by_key(|&i| extents[i].0);
        let starts: Vec<u64> = order.iter().map(|&i| extents[i].0).collect();
        let mut wanted = order.len();
        if wanted == 0 {
            return Ok(hit);
        }

        let read = self.tail.begin_read()?;
        let objects = read.open_table(OBJECTS)?;
        for row in objects.iter()? {
            let (_, v) = row?;
            let offset = decode_row(v.value())?.offset;
            // The last extent that starts at or before this row.
            let p = starts.partition_point(|&s| s <= offset);
            if p == 0 {
                continue;
            }
            let i = order[p - 1];
            let (start, len) = extents[i];
            if offset < start + len && !hit[i] {
                hit[i] = true;
                wanted -= 1;
                if wanted == 0 {
                    break;
                }
            }
        }
        Ok(hit)
    }

    /// **Drop every row `live` rejects.** Returns how many rows went.
    ///
    /// This is the one operation that is not append-only, and it exists for
    /// exactly one caller: `GitOps::gc`, which computes reachability and then has
    /// to remove what is unreachable *before* base znippy compacts the payload —
    /// so that base's notion of "live" already means what git means (§13.20).
    ///
    /// The projection is rebuilt inside the same call. It has to be: it is a
    /// snapshot of a tail that no longer says the same thing, and a stree that
    /// still answers for a deleted oid would be **wrong** rather than merely
    /// incomplete, which is the one failure this design does not tolerate.
    pub fn retain(&self, live: &dyn Fn(&[u8]) -> bool) -> Result<u64> {
        let w = self.tail.begin_write()?;
        let mut dropped = 0u64;
        {
            let mut objects = w.open_table(OBJECTS)?;
            let dead: Vec<Vec<u8>> = objects
                .iter()?
                .filter_map(|row| row.ok())
                .filter(|(k, _)| !live(k.value()))
                .map(|(k, _)| k.value().to_vec())
                .collect();
            for oid in &dead {
                objects.remove(oid.as_slice())?;
                dropped += 1;
            }
        }
        w.commit()?;
        self.rebuild()?;
        Ok(dropped)
    }

    /// One tail read transaction serving `oids`, filling only the slots that are
    /// still `None`. This is the fall-through, and it is a single transaction for
    /// the whole batch on purpose — skade's `resolve_many` phase 2.
    ///
    /// **Not called at all when the projection is complete** — see
    /// [`projection_is_complete`](Self::projection_is_complete). The unfilled
    /// slots are then genuine absences and are counted as such by
    /// [`absences_only`](Self::absences_only).
    fn fill_from_tail(&self, oids: &[&[u8]], out: &mut [Option<IndexRow>]) -> Result<()> {
        self.tail_txns.fetch_add(1, Ordering::AcqRel);
        let read = self.tail.begin_read()?;
        let objects = read.open_table(OBJECTS)?;
        let mut hits = 0u64;
        let mut absent = 0u64;
        for (slot, oid) in out.iter_mut().zip(oids) {
            if slot.is_some() {
                continue;
            }
            match objects.get(*oid)? {
                Some(v) => {
                    *slot = Some(decode_row(v.value())?.as_index_row());
                    hits += 1;
                }
                None => absent += 1,
            }
        }
        self.tail_hits.fetch_add(hits, Ordering::AcqRel);
        self.absent.fetch_add(absent, Ordering::AcqRel);
        Ok(())
    }

    /// Count the unfilled slots as absences without asking the tail. Only ever
    /// called when [`projection_is_complete`](Self::projection_is_complete),
    /// where "the projection does not have it" and "the repository does not have
    /// it" are the same statement.
    fn absences_only(&self, out: &[Option<IndexRow>]) {
        let n = out.iter().filter(|s| s.is_none()).count() as u64;
        self.absent.fetch_add(n, Ordering::AcqRel);
    }
}

/// Every row in the tail, in oid order, as index entries. redb's B-tree is
/// already ordered by the raw oid key, so this hands `S::build` its rows in the
/// order it wants them and the sort inside is a no-op scan.
fn scan(db: &Database) -> Result<Vec<IndexEntry>> {
    let read = db.begin_read()?;
    let objects = read.open_table(OBJECTS)?;
    let mut out = Vec::with_capacity(objects.len()? as usize);
    for row in objects.iter()? {
        let (k, v) = row?;
        out.push(decode_row(v.value())?.as_entry(k.value()));
    }
    Ok(out)
}

impl<S: ObjectIndex> ObjectIndex for ObjectReadStack<S> {
    /// Build a stack over `entries` with an in-memory tail. Every entry lands in
    /// redb and the projection is built from it, so the stack starts fully
    /// absorbed — which is what makes it directly comparable with the two Arrow
    /// arms on the layout benchmark.
    fn build(entries: &[IndexEntry]) -> Result<Self> {
        let stack = Self::in_memory(RebuildTriggers::default())?;
        stack.append(entries)?;
        stack.rebuild()?;
        Ok(stack)
    }

    /// Projection first; **a miss falls through to redb**, which always knows.
    /// `None` here means the tail said no, never that the stree said no.
    fn lookup(&self, oid: &[u8]) -> Option<IndexRow> {
        if let Some(row) = self.projection.read().expect("projection lock").lookup(oid) {
            return Some(row);
        }
        if self.projection_is_complete() {
            self.absent.fetch_add(1, Ordering::AcqRel);
            return None;
        }
        let mut out = [None];
        // A tail read that errors is reported as absent to this signature, which
        // cannot carry an error. Callers that need the distinction use
        // `stats().absent` against their own miss count, or `fill_from_tail`
        // through `lookup_batch`'s caller. A corrupt tail is a `rebuild()`
        // failure long before it is a wrong lookup.
        let _ = self.fill_from_tail(&[oid], &mut out);
        out[0]
    }

    /// Two phases, and the second is not optional: one pipelined stree pass, then
    /// **one** redb transaction for every slot the projection left empty.
    fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>> {
        let mut out = self
            .projection
            .read()
            .expect("projection lock")
            .lookup_batch(oids);
        if out.iter().any(Option::is_none) {
            if self.projection_is_complete() {
                self.absences_only(&out);
            } else {
                let _ = self.fill_from_tail(oids, &mut out);
            }
        }
        out
    }

    /// The floor path — oid → ordinal, no payload column touched. The tail is
    /// still consulted for the misses, because a miss is still not an answer;
    /// a tail-served ordinal carries [`TAIL_ORDINAL_BIT`].
    fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
        let mut out = self
            .projection
            .read()
            .expect("projection lock")
            .ordinals_batch(oids);
        if out.iter().any(Option::is_none) {
            if self.projection_is_complete() {
                let n = out.iter().filter(|s| s.is_none()).count() as u64;
                self.absent.fetch_add(n, Ordering::AcqRel);
                return out;
            }
            let mut rows: Vec<Option<IndexRow>> = out
                .iter()
                .map(|o| {
                    o.map(|ordinal| IndexRow {
                        ordinal,
                        offset: 0,
                        len: 0,
                        obj_type: ObjType::Blob,
                        uncompressed_size: 0,
                        delta_base: 0,
                    })
                })
                .collect();
            let _ = self.fill_from_tail(oids, &mut rows);
            for (slot, row) in out.iter_mut().zip(&rows) {
                *slot = row.map(|r| r.ordinal);
            }
        }
        out
    }

    /// The partial-row path — byte extent only.
    fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>> {
        let mut out = self
            .projection
            .read()
            .expect("projection lock")
            .extents_batch(oids);
        if out.iter().any(Option::is_none) {
            if self.projection_is_complete() {
                let n = out.iter().filter(|s| s.is_none()).count() as u64;
                self.absent.fetch_add(n, Ordering::AcqRel);
                return out;
            }
            let mut rows: Vec<Option<IndexRow>> = out
                .iter()
                .map(|e| {
                    e.map(|(offset, len)| IndexRow {
                        ordinal: 0,
                        offset,
                        len,
                        obj_type: ObjType::Blob,
                        uncompressed_size: 0,
                        delta_base: 0,
                    })
                })
                .collect();
            let _ = self.fill_from_tail(oids, &mut rows);
            for (slot, row) in out.iter_mut().zip(&rows) {
                *slot = row.map(|r| (r.offset, r.len));
            }
        }
        out
    }

    /// The quota gate, over the whole **repository** and without a scan of the
    /// tail: the projection's column scan plus the rows appended after it.
    /// Exact because the archive is append-only — a row the projection already
    /// holds can never gain or lose bytes, so the two terms cannot overlap.
    fn sum_uncompressed(&self) -> u64 {
        self.projection
            .read()
            .expect("projection lock")
            .sum_uncompressed()
            + self.unabsorbed_size.load(Ordering::Acquire)
    }

    /// [`sum_uncompressed`](Self::sum_uncompressed)'s argument, per type.
    fn count_type(&self, t: ObjType) -> usize {
        self.projection
            .read()
            .expect("projection lock")
            .count_type(t)
            + self.unabsorbed_types[t.code() as usize].load(Ordering::Acquire) as usize
    }

    fn name(&self) -> &'static str {
        "ObjectReadStack"
    }

    /// Objects in the **repository**, not in the projection — the stack answers
    /// for the whole repository and this has to agree with what `lookup` will
    /// resolve. [`projection_len`](Self::projection_len) is the other number.
    fn len(&self) -> usize {
        self.total_rows.load(Ordering::Acquire) as usize
    }

    fn ipc_bytes(&self) -> usize {
        self.projection.read().expect("projection lock").ipc_bytes()
    }

    /// The projection's resident bytes. The tail is on disk (or in redb's own
    /// page cache) and is not counted here — counting it would make this
    /// incomparable with the two Arrow arms, which is the number's only use.
    fn resident_bytes(&self) -> usize {
        self.projection
            .read()
            .expect("projection lock")
            .resident_bytes()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::arms::DEFAULT_REDB_CACHE_BYTES;
    use crate::index_layout::{FourTables, synthetic_entries};

    type Stack = ObjectReadStack<OneTableFourColumns>;

    /// A stack with the thresholds off, so a test decides when a rebuild happens.
    fn manual() -> Stack {
        Stack::in_memory(RebuildTriggers::manual()).expect("in-memory stack")
    }

    /// Seal `sealed` into the projection, then append `tail` without absorbing
    /// it. The returned stack is deliberately in the state the whole design is
    /// about: a projection that is a strict subset of the truth.
    fn split_stack(sealed: &[IndexEntry], tail: &[IndexEntry]) -> Stack {
        let s = manual();
        s.append(sealed).expect("seal half");
        s.rebuild().expect("rebuild");
        s.append(tail).expect("append tail");
        s
    }

    /// **The property the whole stack rests on: incomplete, never wrong.**
    ///
    /// The projection covers 300 of 500 objects. Asserted on applied output —
    /// every one of the five facts of all 500 rows, through both the serial and
    /// the batch path — plus the two things that would make the guard hollow:
    /// that the projection really is short (its own `lookup` misses all 200 tail
    /// objects, so the fall-through is genuinely being exercised) and that every
    /// projection hit is byte-identical to what a complete index built over all
    /// 500 returns.
    ///
    /// Seen RED by making `lookup_batch`'s phase 2 conditional on
    /// `out.iter().all(Option::is_none)` instead of `any` — i.e. falling through
    /// only when the projection answered nothing: "the stack lost tail object
    /// e50298d4cb83164fcbbd2a0d4fd99ebb43111e93, batch path". Restored.
    ///
    /// Seen RED a second time by returning `None` from `lookup` on a projection
    /// miss with no fall-through: "the stack lost tail object
    /// e50298d4cb83164fcbbd2a0d4fd99ebb43111e93, serial path". Restored.
    ///
    /// Seen RED a third time, for `delta_base`, by having [`encode_row`] write
    /// a literal `0` into bytes 33..41 — the shape of a column that exists in
    /// the struct and is never persisted: "the delta base of
    /// b2af964cd8a34795ce5ad114d4e2a2144cf2a6b1 did not survive the projection
    /// path / left: 0 / right: 6186". **Six** of this module's tests went red
    /// on that one edit, on both sides of the fall-through, which is what says
    /// the tail encoding is load-bearing rather than incidental. Restored.
    #[test]
    fn the_projection_is_incomplete_but_never_wrong() {
        let all = synthetic_entries(500, 20, 0xC0FFEE);
        let (sealed, tail) = all.split_at(300);
        let stack = split_stack(sealed, tail);

        assert_eq!(stack.projection_len(), 300, "the projection must be short");
        assert_eq!(stack.len(), 500, "the stack answers for the repository");
        // The tail half has to carry delta bases, or the `delta_base` assertion
        // below is an assertion about zero on both sides of the fall-through.
        assert!(
            tail.iter().filter(|e| e.delta_base != 0).count() >= 10,
            "the un-absorbed half carries no delta bases — the tail encoding would be untested"
        );

        // Two references. `sealed_only` is what the projection is *supposed* to
        // be — an index over exactly the 300 sealed objects, so its ordinals are
        // the ranks the projection must reproduce. `complete` is the index the
        // stack must be indistinguishable from on the five facts.
        let sealed_only = FourTables::build(sealed).expect("sealed-only index");
        let complete = FourTables::build(&all).expect("complete index");

        // The projection alone must MISS every tail object — otherwise the
        // fall-through below proves nothing.
        {
            let proj = stack.projection.read().unwrap();
            for e in tail {
                assert!(
                    proj.lookup(&e.oid).is_none(),
                    "the projection already holds {} — this test would be vacuous",
                    hex::encode(&e.oid)
                );
            }
            for e in sealed {
                assert!(
                    proj.lookup(&e.oid).is_some(),
                    "the projection lost sealed object {}",
                    hex::encode(&e.oid)
                );
            }
        }

        let refs: Vec<&[u8]> = all.iter().map(|e| e.oid.as_slice()).collect();
        let batched = stack.lookup_batch(&refs);
        for (i, e) in all.iter().enumerate() {
            let via_serial = stack.lookup(&e.oid).unwrap_or_else(|| {
                panic!(
                    "the stack lost {} object {}, serial path",
                    if i < 300 { "sealed" } else { "tail" },
                    hex::encode(&e.oid)
                )
            });
            let via_batch = batched[i].unwrap_or_else(|| {
                panic!(
                    "the stack lost {} object {}, batch path",
                    if i < 300 { "sealed" } else { "tail" },
                    hex::encode(&e.oid)
                )
            });
            assert_eq!(via_serial, via_batch, "serial and batch disagree");

            // The five facts, against the entry that was appended.
            assert_eq!(via_serial.offset, e.offset);
            assert_eq!(via_serial.len, e.len);
            assert_eq!(via_serial.obj_type, e.obj_type);
            assert_eq!(via_serial.uncompressed_size, e.uncompressed_size);
            assert_eq!(
                via_serial.delta_base,
                e.delta_base,
                "the delta base of {} did not survive the {} path",
                hex::encode(&e.oid),
                if i < 300 { "projection" } else { "tail" }
            );

            // A HIT in the projection is valid: the five facts are the complete
            // index's, and the ordinal is the rank within the generation the
            // projection covers — which is what an ordinal means.
            let truth = complete.lookup(&e.oid).expect("complete index has it");
            assert_eq!(via_serial.offset, truth.offset);
            assert_eq!(via_serial.len, truth.len);
            assert_eq!(via_serial.obj_type, truth.obj_type);
            assert_eq!(via_serial.uncompressed_size, truth.uncompressed_size);
            assert_eq!(via_serial.delta_base, truth.delta_base);
            if i < 300 {
                assert!(
                    !is_tail_row(&via_serial),
                    "sealed row wearing a tail ordinal"
                );
                assert_eq!(
                    via_serial.ordinal,
                    sealed_only.lookup(&e.oid).unwrap().ordinal,
                    "the projection's ordinal for {} is not its rank in the generation the \
                     projection covers",
                    hex::encode(&e.oid)
                );
            } else {
                assert!(
                    is_tail_row(&via_serial),
                    "tail row {} has no tail ordinal",
                    hex::encode(&e.oid)
                );
            }
        }
    }

    /// A miss in the tree is not an answer, and an absence is.
    ///
    /// Seen RED by having `fill_from_tail` skip its `objects.get` and count every
    /// unfilled slot as absent: the tail-served block came back empty —
    /// "assertion `left == right` failed / left: 0 / right: 80". Restored.
    #[test]
    fn only_a_tail_miss_is_absent_and_only_a_tail_hit_counts() {
        let all = synthetic_entries(200, 20, 7);
        let (sealed, tail) = all.split_at(120);
        let stack = split_stack(sealed, tail);
        let nowhere = synthetic_entries(64, 20, 0x00AB_5E47);

        let mut refs: Vec<&[u8]> = Vec::new();
        refs.extend(sealed.iter().map(|e| e.oid.as_slice()));
        refs.extend(tail.iter().map(|e| e.oid.as_slice()));
        refs.extend(nowhere.iter().map(|e| e.oid.as_slice()));
        let rows = stack.lookup_batch(&refs);

        assert_eq!(rows[..120].iter().filter(|r| r.is_some()).count(), 120);
        assert_eq!(rows[120..200].iter().filter(|r| r.is_some()).count(), 80);
        assert!(
            rows[200..].iter().all(Option::is_none),
            "an object in no repository resolved to a row"
        );

        let st = stack.stats();
        assert_eq!(st.tail_hits, 80, "tail-served lookups miscounted");
        assert_eq!(st.absent, 64, "genuine absences miscounted");
    }

    /// **Append-only, enforced.** The refusal is what makes a stree hit
    /// trustworthy; if a fact could be rewritten, a stale projection would return
    /// the old one.
    ///
    /// **Identity is refused; placement is not.** REWRITTEN 2026-08-14 — the
    /// old form compared `offset`, `len` and `delta_base` too, and those say
    /// where a copy landed rather than what the object is. A re-pushed pack
    /// places the same oid at a new offset, which is the ordinary case, so the
    /// guard refused normal pushes: the ingest aborted and the connection
    /// dropped with `client_told=false`, leaving the client with
    /// "send-pack: unexpected disconnect" and no reason. It failed
    /// `bare_lifecycle` and `endurance` on all four znippy bench columns and
    /// bit any `--force` push that resent a known object.
    ///
    /// The old comparison's stated fear — a projection carrying "a base offset
    /// the archive never wrote" — is answered better by keeping the OLD row:
    /// the archive is append-only and `gc` truncates nothing, so the first
    /// writer's offset and base still address bytes that are still there.
    ///
    /// Seen RED by comparing `offset` again: "a re-pushed pack at a new offset
    /// was refused" — which is precisely the production failure, reproduced.
    ///
    /// Seen RED the other way by dropping `old.obj_type != e.obj_type`: "an oid
    /// that changed type was accepted" — the same oid describing a different
    /// object, waved through.
    #[test]
    fn an_append_only_violation_is_refused() {
        let entries = synthetic_entries(16, 20, 11);
        let stack = manual();
        stack.append(&entries).expect("first append");

        // Re-appending the identical batch is a no-op, not an error: a re-pushed
        // pack repeats objects.
        stack
            .append(&entries)
            .expect("identical re-append is idempotent");
        assert_eq!(
            stack.len(),
            16,
            "an idempotent re-append changed the row count"
        );

        // ★ THE PRODUCTION CASE: the same oid, re-pushed, lands elsewhere.
        // This must be a NO-OP, and the first writer's placement must stand.
        let mut moved = entries[3].clone();
        moved.offset += 1;
        stack
            .append(std::slice::from_ref(&moved))
            .expect("a re-pushed pack at a new offset was refused");
        let row = stack.lookup(&entries[3].oid).expect("still there");
        assert_eq!(
            row.offset, entries[3].offset,
            "the second placement overwrote the first — the projection now names \
             an offset that is not where the first writer put the bytes"
        );
        assert_eq!(stack.len(), 16, "a re-placement added a row");

        // ★ IDENTITY still refuses: same oid, different object.
        let mut retyped = entries[4].clone();
        retyped.uncompressed_size += 1;
        let err = stack
            .append(std::slice::from_ref(&retyped))
            .expect_err("an oid that changed size was accepted");
        assert!(
            err.to_string().contains("identity violation"),
            "wrong error: {err}"
        );

        // ★ A DIFFERENT DELTA BASE IS PLACEMENT TOO, and the first row stands.
        //
        // The old form refused this, on the reasoning that a row naming a base
        // the archive never wrote would be left behind. Keeping the OLD row
        // answers that completely: the base the FIRST writer recorded is still
        // in the archive, because the blob is append-only and `gc` truncates
        // nothing. The second pack's base is simply not adopted.
        //
        // Refusing it instead is what broke real pushes — a re-pushed pack
        // re-deltas against whatever is in ITS window, so a differing base is
        // the ordinary case, not evidence of corruption.
        let based = entries
            .iter()
            .find(|e| e.delta_base != 0)
            .expect("the fixture must carry a delta");
        let mut rebased = based.clone();
        rebased.delta_base += 8;
        stack
            .append(std::slice::from_ref(&rebased))
            .expect("a re-pushed pack that re-deltaed was refused");
        assert_eq!(
            stack.lookup(&based.oid).unwrap().delta_base,
            based.delta_base,
            "the second pack's base was adopted — the projection now names a base \
             the FIRST writer never recorded, which is the thing the old guard \
             was right to fear"
        );
    }

    /// The volume trigger fires on appended bytes and the miss trigger on
    /// tail-served lookups, and neither fires on a genuine absence.
    ///
    /// Seen RED by counting `absent` into `tail_hits` in `fill_from_tail`:
    /// "assertion `left == right` failed: absences tripped the miss trigger /
    /// left: Some(TailHits(1000)) / right: None". Restored.
    #[test]
    fn each_trigger_fires_on_its_own_signal() {
        // ── volume ──
        let entries = synthetic_entries(64, 20, 21);
        let bytes: u64 = entries.iter().map(|e| e.len).sum();
        let stack = Stack::in_memory(RebuildTriggers {
            tail_bytes: bytes,
            ..RebuildTriggers::manual()
        })
        .unwrap();
        let g0 = stack.stats().generation;
        let reason = stack.append(&entries).unwrap();
        assert!(
            matches!(reason, Some(RebuildReason::TailBytes(_))),
            "the volume trigger did not fire at exactly its threshold: {reason:?}"
        );
        assert_eq!(stack.projection_len(), 64, "the rebuild absorbed nothing");
        assert_eq!(stack.stats().generation, g0 + 1);
        assert_eq!(
            stack.stats().tail_bytes,
            0,
            "the byte counter was not reset"
        );

        // ── misses ──
        let all = synthetic_entries(200, 20, 22);
        let (sealed, tail) = all.split_at(100);
        // A flat floor of 100 misses: `tail_hits_per_row` 0 means the threshold
        // is exactly the floor, which makes the arithmetic in this test the
        // trigger's and not the ratio's.
        let stack = Stack::in_memory(RebuildTriggers {
            tail_hits_per_row: 0.0,
            min_tail_hits: 100,
            tail_bytes: 0,
        })
        .unwrap();
        stack.append(sealed).unwrap();
        stack.rebuild().unwrap();
        stack.append(tail).unwrap();
        let g1 = stack.stats().generation;
        assert_eq!(stack.projection_len(), 100);

        // 1000 lookups of oids in NO repository: 1000 tail misses, zero
        // tail-served hits, and no rebuild may follow.
        let nowhere = synthetic_entries(1000, 20, 23);
        let refs: Vec<&[u8]> = nowhere.iter().map(|e| e.oid.as_slice()).collect();
        assert!(stack.lookup_batch(&refs).iter().all(Option::is_none));
        assert_eq!(stack.stats().absent, 1000);
        assert_eq!(
            stack.rebuild_due(),
            None,
            "absences tripped the miss trigger"
        );
        stack.maybe_rebuild().unwrap();
        assert_eq!(
            stack.stats().generation,
            g1,
            "1000 absences must not rebuild"
        );
        assert_eq!(stack.projection_len(), 100);

        // 100 lookups the tail DOES answer: the trigger fires.
        let refs: Vec<&[u8]> = tail.iter().map(|e| e.oid.as_slice()).collect();
        assert_eq!(stack.lookup_batch(&refs).iter().flatten().count(), 100);
        assert_eq!(
            stack.rebuild_due(),
            Some(RebuildReason::TailHits(100)),
            "100 tail-served lookups did not trip a threshold of 100"
        );
        assert!(stack.maybe_rebuild().unwrap().is_some());
        assert_eq!(
            stack.projection_len(),
            200,
            "the rebuild did not absorb the tail"
        );
        assert_eq!(stack.stats().generation, g1 + 1);
    }

    /// The two column scans answer for the **repository**, not for the
    /// projection — including the rows the projection has not absorbed, and
    /// without scanning the tail to find them.
    ///
    /// Seen RED by dropping the `unabsorbed_size` term from `sum_uncompressed`:
    /// "assertion `left == right` failed: sum_uncompressed answered for the
    /// projection, not the repository / left: 1179629 / right: 1915881".
    /// Restored.
    ///
    /// The rejected-append half of this test is what pins the *ordering* in
    /// `append`: the counters are accumulated locally and applied after the
    /// commit, because a `bail!` half-way through a batch rolls redb back and
    /// counters bumped inside the loop would not roll back with it. That one is
    /// fixed by construction rather than watched — there is one place the
    /// counters are written and it is after `commit()`.
    #[test]
    fn the_column_scans_answer_for_the_repository_not_the_projection() {
        let all = synthetic_entries(300, 20, 71);
        let (sealed, tail) = all.split_at(180);
        let stack = split_stack(sealed, tail);
        let complete = OneTableFourColumns::build(&all).unwrap();

        assert!(
            stack.projection.read().unwrap().sum_uncompressed() < complete.sum_uncompressed(),
            "the projection already sums to the whole repository — this test would be vacuous"
        );
        assert_eq!(
            stack.sum_uncompressed(),
            complete.sum_uncompressed(),
            "sum_uncompressed answered for the projection, not the repository"
        );
        for t in ObjType::ALL {
            assert_eq!(
                stack.count_type(t),
                complete.count_type(t),
                "count_type({}) answered for the projection",
                t.as_str()
            );
        }

        // A rejected append must not move either aggregate.
        let mut rewritten = all[0].clone();
        rewritten.uncompressed_size += 1_000_000;
        let batch = [all[7].clone(), rewritten];
        assert!(stack.append(&batch).is_err());
        assert_eq!(stack.sum_uncompressed(), complete.sum_uncompressed());

        // And a rebuild leaves them where they were.
        stack.rebuild().unwrap();
        assert_eq!(stack.sum_uncompressed(), complete.sum_uncompressed());
        for t in ObjType::ALL {
            assert_eq!(stack.count_type(t), complete.count_type(t));
        }
    }

    /// The two narrow batch paths fall through to the tail exactly as the
    /// full-row one does. A miss is not an answer on those either.
    ///
    /// Seen RED by returning the projection's `ordinals_batch` unchanged:
    /// "assertion `left == right` failed: the ordinal path lost 120 of 300
    /// objects / left: 180 / right: 300". Restored.
    #[test]
    fn the_narrow_batch_paths_also_fall_through() {
        let all = synthetic_entries(300, 20, 81);
        let (sealed, tail) = all.split_at(180);
        let stack = split_stack(sealed, tail);
        let refs: Vec<&[u8]> = all.iter().map(|e| e.oid.as_slice()).collect();

        let ordinals = stack.ordinals_batch(&refs);
        assert_eq!(
            ordinals.iter().filter(|o| o.is_some()).count(),
            300,
            "the ordinal path lost {} of 300 objects",
            300 - ordinals.iter().filter(|o| o.is_some()).count()
        );
        for (o, e) in ordinals[180..].iter().zip(tail) {
            assert!(
                o.unwrap() & TAIL_ORDINAL_BIT != 0,
                "tail object {} got a projection ordinal",
                hex::encode(&e.oid)
            );
        }

        let extents = stack.extents_batch(&refs);
        for (x, e) in extents.iter().zip(&all) {
            assert_eq!(
                *x,
                Some((e.offset, e.len)),
                "extent lost or wrong for {}",
                hex::encode(&e.oid)
            );
        }
    }

    /// A rebuild re-derives every ordinal as the new lexicographic rank, and the
    /// tail's high-bit ordinals disappear as their rows are absorbed. This is the
    /// generation caveat, asserted rather than only documented.
    ///
    /// Seen RED by having `rebuild` compute the fresh projection and never swap
    /// it in: "an absorbed row kept its tail ordinal". Restored.
    #[test]
    fn a_rebuild_re_derives_ordinals_and_clears_the_tail_bit() {
        let all = synthetic_entries(120, 32, 31);
        let (sealed, tail) = all.split_at(60);
        let stack = split_stack(sealed, tail);

        let before = stack.lookup(&tail[0].oid).unwrap();
        assert!(is_tail_row(&before));

        stack.rebuild().unwrap();
        let after = stack.lookup(&tail[0].oid).unwrap();
        assert!(
            !is_tail_row(&after),
            "an absorbed row kept its tail ordinal"
        );
        assert_eq!(after.offset, before.offset, "absorption changed a fact");
        assert_eq!(after.len, before.len);
        assert_eq!(after.obj_type, before.obj_type);
        assert_eq!(after.uncompressed_size, before.uncompressed_size);
        assert_eq!(after.delta_base, before.delta_base);

        // The delta base is an ARCHIVE OFFSET and absorption must not touch it.
        // An ordinal in this column would have had to be re-derived by the
        // rebuild — silently, and to a different number, because the rebuild
        // re-ranks every row. That is §13's argument, asserted on a row that
        // actually carries a base rather than on the 0 sentinel.
        let based = tail
            .iter()
            .find(|e| e.delta_base != 0)
            .expect("the tail half must carry at least one delta");
        let row = stack.lookup(&based.oid).unwrap();
        assert_eq!(
            row.delta_base,
            based.delta_base,
            "the rebuild moved {}'s delta base from {} to {}",
            hex::encode(&based.oid),
            based.delta_base,
            row.delta_base
        );
        assert_ne!(
            row.delta_base, row.ordinal as u64,
            "a delta base that equals a row ordinal is the mistake §13 forbids"
        );

        let complete = OneTableFourColumns::build(&all).unwrap();
        assert_eq!(
            after.ordinal,
            complete.lookup(&tail[0].oid).unwrap().ordinal,
            "the rebuilt ordinal is not the lexicographic rank"
        );
    }

    /// **The crash-recovery diff is exact at the pack boundary.**
    ///
    /// Three packs' worth of rows laid end to end, the middle one never appended:
    /// [`ObjectReadStack::extents_with_rows`] must answer `true, false, true`.
    /// The middle pack's extent is the interesting one — it is bounded on both
    /// sides by rows that *are* in the tail, so an off-by-one in either direction
    /// reports it absorbed, its objects are never re-queued after a restart, and a
    /// durable pack becomes unreadable for ever.
    ///
    /// Asserted per extent, not as a count.
    ///
    /// Seen RED by `let p = starts.partition_point(|&s| s <= offset)` →
    /// `partition_point(|&s| s < offset)`, i.e. placing a row against the
    /// *previous* extent when it lands on a pack's first byte: "left: [false] /
    /// right: [true]" on the single-row extent. Restored.
    ///
    /// Seen RED a second time by `offset < start + len` → `offset <= start + len`:
    /// "a row one byte past pack B's extent was counted as B's — left: [true,
    /// true], right: [true, false]". **That mutation stayed green against the
    /// four-pack question**, and the reason is worth writing down: when every
    /// extent is in the question and the packs are adjacent, the row at B's end
    /// offset is also C's first row, and the binary search assigns it to C before
    /// the bound is ever consulted. The bound only bites when a pack's successor
    /// is *not* in the question — a journal whose tail was torn, which is the
    /// state this whole diff exists to survive. The `[..2]` case below is
    /// therefore not an extra assertion, it is the one that tests the bound at
    /// all. Restored.
    #[test]
    fn the_crash_recovery_diff_is_exact_at_the_pack_boundary() {
        let all = synthetic_entries(400, 20, 0xB17_5E7);
        // Four packs, back to back, the way `SafeWriter` appends them.
        let bounds = [(0usize, 100usize), (100, 200), (200, 300), (300, 400)];
        let extents: Vec<(u64, u64)> = bounds
            .iter()
            .map(|&(a, b)| {
                let start = all[a].offset;
                let end = all[b - 1].offset + all[b - 1].len;
                (start, end - start)
            })
            .collect();
        assert_eq!(
            extents[1].0,
            extents[0].0 + extents[0].1,
            "the fixture's packs must be adjacent or the boundary is not under test"
        );

        let stack = manual();
        for (i, &(a, b)) in bounds.iter().enumerate() {
            if i != 1 {
                stack.append(&all[a..b]).unwrap();
            }
        }
        assert_eq!(stack.len(), 300, "the middle pack must be the missing one");

        let hit = stack.extents_with_rows(&extents).unwrap();
        assert_eq!(
            hit,
            vec![true, false, true, true],
            "pack B has no rows in the tail and was reported absorbed"
        );

        // **The containment bound, made observable.** Asking only about A and B
        // is the torn-journal case: pack C's rows are in the tail but its extent
        // is not in the question, and C's first row sits at exactly B's end
        // offset. That row belongs to C and must not answer for B.
        assert_eq!(
            stack.extents_with_rows(&extents[..2]).unwrap(),
            vec![true, false],
            "a row one byte past pack B's extent was counted as B's"
        );

        // A zero-length extent has no rows by construction, and an extent past
        // everything the tail knows about has none either.
        let past = all[399].offset + all[399].len;
        assert_eq!(
            stack.extents_with_rows(&[(0, 0), (past, 4096)]).unwrap(),
            vec![false, false]
        );
        // And an extent that covers exactly one row is that row's pack.
        assert_eq!(
            stack
                .extents_with_rows(&[(all[7].offset, all[7].len)])
                .unwrap(),
            vec![true]
        );
    }

    /// A file-backed tail survives being closed and reopened, and the reopened
    /// stack warm-starts a projection over everything — including rows that were
    /// only in the tail when the process went away.
    ///
    /// Seen RED by having `from_db` build the projection from an empty slice
    /// instead of `scan(&db)`: "reopened projection covers 0 of 90". Restored.
    #[test]
    fn a_file_backed_tail_reopens_warm() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("objects.tail.redb");
        let all = synthetic_entries(90, 20, 41);
        {
            let stack =
                ObjectReadStack::<OneTableFourColumns>::open(
                    &path,
                    RebuildTriggers::manual(),
                    DEFAULT_REDB_CACHE_BYTES,
                )
                    .unwrap();
            stack.append(&all[..40]).unwrap();
            stack.rebuild().unwrap();
            stack.append(&all[40..]).unwrap();
            assert_eq!(stack.projection_len(), 40);
        }
        let stack =
            ObjectReadStack::<OneTableFourColumns>::open(
                    &path,
                    RebuildTriggers::manual(),
                    DEFAULT_REDB_CACHE_BYTES,
                ).unwrap();
        assert_eq!(
            stack.projection_len(),
            90,
            "reopened projection covers {} of 90",
            stack.projection_len()
        );
        for e in &all {
            let row = stack
                .lookup(&e.oid)
                .unwrap_or_else(|| panic!("reopen lost {}", hex::encode(&e.oid)));
            assert_eq!(row.offset, e.offset);
            assert_eq!(row.uncompressed_size, e.uncompressed_size);
            assert_eq!(row.delta_base, e.delta_base, "reopen lost a delta base");
        }
        assert!(
            all.iter().filter(|e| e.delta_base != 0).count() >= 5,
            "the fixture must carry delta bases across the reopen"
        );
    }

    /// The packed tail row rejects the two things that would make it silently
    /// wrong: a short buffer and a type code git does not use.
    ///
    /// Seen RED by dropping the length check from `decode_row`: the short-buffer
    /// case panicked — "range end index 33 out of range for slice of length 32"
    /// — instead of returning an error. Restored.
    ///
    /// Seen RED for the `delta_base` field by decoding it from byte 25 (where
    /// `uncompressed_size` starts) instead of byte 33: "the delta base did not
    /// survive the tail row / left: 2448 / right: 4242" — it came back holding
    /// the *size*. Both are plausible u64s, so only a comparison against the
    /// entry that was encoded can tell them apart. Restored.
    ///
    /// A row written by the 33-byte encoding is a **length** error here rather
    /// than a misread: `decode_row` refuses anything that is not exactly
    /// `TAIL_ROW_BYTES`, and the assertion below is the same check the old width
    /// now trips.
    #[test]
    fn a_packed_tail_row_round_trips_and_refuses_nonsense() {
        let mut e = synthetic_entries(1, 20, 51)[0].clone();
        // A one-entry workload has nothing to delta against, so the base is set
        // here: a round trip that only ever saw 0 would not test the field.
        e.delta_base = 4242;
        let e = &e;
        let packed = encode_row(9, e);
        let back = decode_row(&packed).unwrap();
        assert_eq!(back.seq, 9);
        assert_eq!(back.offset, e.offset);
        assert_eq!(back.len, e.len);
        assert_eq!(back.obj_type, e.obj_type);
        assert_eq!(back.uncompressed_size, e.uncompressed_size);
        assert_eq!(
            back.delta_base, e.delta_base,
            "the delta base did not survive the tail row"
        );
        assert_ne!(
            back.delta_base, back.uncompressed_size,
            "the fixture must not let the two u64 fields alias"
        );

        assert!(decode_row(&packed[..TAIL_ROW_BYTES - 1]).is_err());
        // 33 bytes is exactly the pre-`delta_base` row width, and it is refused
        // rather than read as a row with the field missing.
        assert!(
            decode_row(&packed[..33]).is_err(),
            "a row in the old 33-byte encoding must be refused, not reinterpreted"
        );
        let mut bad = packed;
        bad[24] = 5; // git uses 1-4, 6, 7; 5 is unassigned
        assert!(
            decode_row(&bad).is_err(),
            "an unassigned object type code was accepted"
        );
    }

    /// The stack is the same index as the two Arrow arms once it is fully
    /// absorbed, which is what lets it be dropped into
    /// `examples/index_layout_bench.rs` unchanged.
    ///
    /// Seen RED by having `ObjectIndex::build` skip its final `rebuild()`, so
    /// every row was served from the tail instead of from the projection:
    /// "assertion `left == right` failed: build left rows in the tail / left: 0
    /// / right: 400". Restored.
    #[test]
    fn a_fully_absorbed_stack_agrees_with_both_arrow_arms() {
        let entries = synthetic_entries(400, 20, 61);
        let absent = synthetic_entries(100, 20, 62);
        let stack = Stack::build(&entries).unwrap();
        let a = FourTables::build(&entries).unwrap();
        let b = OneTableFourColumns::build(&entries).unwrap();

        assert_eq!(stack.len(), 400);
        assert_eq!(stack.projection_len(), 400, "build left rows in the tail");

        let mut refs: Vec<&[u8]> = entries.iter().map(|e| e.oid.as_slice()).collect();
        refs.extend(absent.iter().map(|e| e.oid.as_slice()));
        let rs = stack.lookup_batch(&refs);
        assert_eq!(
            rs,
            a.lookup_batch(&refs),
            "the stack disagrees with FourTables"
        );
        assert_eq!(
            rs,
            b.lookup_batch(&refs),
            "the stack disagrees with OneTableFourColumns"
        );
        assert_eq!(
            stack.stats().tail_hits,
            0,
            "a fully absorbed stack still went to the tail for a row"
        );
        assert_eq!(stack.stats().absent, 100);
    }

    /// **A complete projection answers an absence without opening a redb
    /// transaction at all** — and the moment one row is appended, it stops doing
    /// that.
    ///
    /// This is the fast path the bench found was worth having: a `have`
    /// negotiation is mostly misses, and paying a redb round trip for each of
    /// them made the stack six times the cost of the bare Arrow arm on the most
    /// common git operation. `tail_txns` is the applied output — an assertion on
    /// timing would be a flake, and an assertion on the returned rows cannot
    /// distinguish the two paths, because both return the same rows. That is the
    /// point: the fast path is only allowed to exist because it is
    /// indistinguishable in its answers.
    ///
    /// Seen RED by deleting the `projection_is_complete()` branch from
    /// `lookup_batch`: "assertion `left == right` failed: a complete projection
    /// opened 1 redb transactions to say 'no' / left: 1 / right: 0". Restored.
    ///
    /// Seen RED the other way — the dangerous way — by making
    /// `projection_is_complete` return `true` unconditionally: the
    /// `the_projection_is_incomplete_but_never_wrong` test failed with "the
    /// stack lost tail object …, batch path", because an incomplete projection
    /// then declared its misses absent. Restored.
    #[test]
    fn a_complete_projection_answers_absent_without_asking_the_tail() {
        let entries = synthetic_entries(200, 20, 91);
        let nowhere = synthetic_entries(500, 20, 92);
        let stack = Stack::build(&entries).unwrap();
        assert!(stack.projection_is_complete());

        let refs: Vec<&[u8]> = nowhere.iter().map(|e| e.oid.as_slice()).collect();
        assert!(stack.lookup_batch(&refs).iter().all(Option::is_none));
        assert!(stack.lookup(&nowhere[0].oid).is_none());
        assert!(stack.ordinals_batch(&refs).iter().all(Option::is_none));
        assert!(stack.extents_batch(&refs).iter().all(Option::is_none));

        let st = stack.stats();
        assert_eq!(st.unabsorbed_rows, 0);
        assert_eq!(
            st.tail_txns, 0,
            "a complete projection opened {} redb transactions to say 'no'",
            st.tail_txns
        );
        assert_eq!(st.absent, 500 + 1 + 500 + 500, "absences went uncounted");

        // One appended row and the fall-through is on again — the fast path is
        // conditional on completeness and not on having been complete once.
        let more = synthetic_entries(1, 20, 93);
        stack.append(&more).unwrap();
        assert!(!stack.projection_is_complete());
        assert!(stack.lookup_batch(&refs).iter().all(Option::is_none));
        assert!(
            stack.stats().tail_txns > 0,
            "an incomplete projection did not consult the tail"
        );
        assert!(
            stack.lookup(&more[0].oid).is_some(),
            "the appended row is not reachable"
        );
    }
}