znippy-plugin-git 0.1.4

//! **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::fs::{self, File};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};

use anyhow::{Context, Result, anyhow, bail};
use redb::{Database, ReadableTable, ReadableTableMetadata, TableDefinition};
use znippy_common::arrow::buffer::Buffer;

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";
/// A durable, monotonically increasing marker of the OBJECTS table's **content
/// state**: bumped by one, inside the very same redb write transaction, whenever
/// a commit changes the set of oids — an [`ObjectReadStack::append`] that inserts
/// at least one new row, or a [`ObjectReadStack::retain`] that drops at least one.
///
/// It was introduced as the validity key of the `objects.ordinal` memo (S-082),
/// which is gone since 2026-09-04 — the persisted index's oid column is the
/// ordinal space now, current by construction. The epoch stays: it is the one
/// cheap, sound, crash-consistent "has the oid set changed" signal this store
/// has. `arrival_seq` alone will not do — it counts inserts and is blind to
/// `retain`'s removals — and a blake3 over the oid set is a full scan. Because
/// the bump shares the mutation's transaction, `epoch` names the OBJECTS content
/// uniquely across this store's life: equal epochs ⟹ an unchanged set. See
/// [`ObjectReadStack::tail_epoch_and_len`].
const META_TAIL_EPOCH: &str = "tail_epoch";

/// **The layout [`encode_row`] wrote these rows with.**
///
/// [`decode_row`] refuses a row whose width is not [`TAIL_ROW_BYTES`] — on
/// purpose, because reinterpreting an unknown layout is how a wrong offset gets
/// into an index and is then queried and believed. But a refusal with nothing
/// behind it is a store that has stopped answering, and until this key existed
/// that is exactly what a layout change produced: every read `bail`ing, with no
/// way back short of deleting the repository.
///
/// There is a way back, and it costs nothing to take, because **the tail is a
/// cache** (see [`decode_row`]) and the store already knows how to rebuild one:
/// `Absorber::adopt_journal` derives what is unabsorbed as *"its extent is in
/// the journal and its rows are not in the index"* and requeues it. Empty the
/// OBJECTS table and every pack is unabsorbed by that definition, so the drain
/// re-indexes the archive and reads fall back to the pack meanwhile — **slower,
/// never wrong**, which is the same path a crash mid-absorb already takes.
///
/// So a version mismatch is a **drop and rebuild**, not a failure. Absent is
/// treated as [`TAIL_FORMAT`] rather than as "old": every tail in existence when
/// this key was introduced was written by the layout it names, so stamping them
/// silently is correct and costs no rebuild. The next change to the row bumps
/// the constant and those stores rebuild once.
const META_TAIL_FORMAT: &str = "tail_format";

/// The current value of [`META_TAIL_FORMAT`]. **Bump this in the same commit as
/// any change to [`TAIL_ROW_BYTES`] or to what [`encode_row`] puts in it.**
///
/// `1` is the layout in use since the `delta_base` field landed (PLAN §13) — the
/// 41-byte row this constant was introduced beside, not a new one.
const TAIL_FORMAT: u64 = 1;

/// `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 {
    is_tail_ordinal(row.ordinal)
}

/// The same question of a bare ordinal.
#[inline]
pub fn is_tail_ordinal(ordinal: u32) -> bool {
    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. On a persisting stack this
    /// is the **floor** of the volume threshold; see `fold_ratio`.
    pub tail_bytes: u64,
    /// **On a persisting stack, the volume threshold scales with the index**:
    /// a fold fires once the object bytes appended since the last fold reach
    /// `fold_ratio ×` the persisted index's own size (never below `tail_bytes`).
    /// `0.0` leaves the fixed floor alone. See
    /// [`DEFAULT_FOLD_RATIO`](Self::DEFAULT_FOLD_RATIO) for why the value is 1.
    /// Ignored on a stack that does not persist — there a rebuild writes nothing
    /// and the fixed 64 MiB is the right cadence.
    pub fold_ratio: f64,
}

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;

    /// **1.0 — fold when the bytes appended since the last fold equal the
    /// index's own size.**
    ///
    /// Since 2026-09-04 a rebuild on a persisting stack WRITES the projection
    /// (`objects.index`), so the cadence above stopped being free. MEASURED on
    /// the kernel mirror with the fixed 64 MiB floor alone: an 8.72 GB import
    /// folded ≈136 times, each rewriting the whole growing file — ≈51 GB of
    /// sequential writes for a 747 MB index (INFERRED from the arithmetic; ≥18
    /// distinct file sizes OBSERVED at 10 s sampling). The ack did not slow on
    /// oden's NVMe; on the appliance's disks it would.
    ///
    /// A ratio of 1 gives a bound that is easy to say and holds by construction:
    /// **the fold's total writes never exceed the push's own bytes.** Each fold
    /// writes an index roughly the size of the bytes that triggered it, so the
    /// sum of index writes over an import is ≈ the bytes imported — ~8.7 GB for
    /// the kernel instead of ~51, and the number of folds falls from ~136 to
    /// O(log n). The price is the redb tail's high-water mark: with the index at
    /// 53.5 B/object and pack bytes at ~620 B/object, "appended bytes = index
    /// bytes" is ≈ 8.6 % of the folded rows in redb before a fold, i.e. ~1.2 M
    /// rows ≈ 250 MB of redb rows on the kernel mirror (redb allocates in
    /// doublings, so a ~512 MiB file) against 270 MB with the floor alone.
    /// Re-measure on the rig after changing either number; both are in
    /// `.nornir/storage-levers-2026-09-03.md`.
    pub const DEFAULT_FOLD_RATIO: f64 = 1.0;

    /// 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,
            fold_ratio: 0.0,
        }
    }
}

impl Default for RebuildTriggers {
    fn default() -> Self {
        Self {
            tail_hits_per_row: Self::DEFAULT_TAIL_HITS_PER_ROW,
            fold_ratio: Self::DEFAULT_FOLD_RATIO,
            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,
}

/// What [`ObjectReadStack::oids_raw`] hands back: the ordinal space and its
/// provenance.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OrdinalSpace {
    /// Every oid, fixed width, back to back, in oid order.
    pub raw: Vec<u8>,
    /// Oids that came straight off the persisted projection's column.
    pub folded_rows: u64,
    /// Oids that came off a scan of the redb tail — the rows since the last
    /// fold, or every row on a stack that does not persist.
    pub tail_rows: u64,
}

/// 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,
    /// Folds that persisted the projection and emptied redb of what it holds.
    /// `0` on an in-memory stack and on an arm that cannot be persisted.
    pub folds: u64,
    /// `1` when this stack's open mapped `objects.index` and skipped the redb
    /// scan-and-build; `0` when it had to rebuild. The applied-output proof
    /// that a warm open is warm.
    pub persisted_loads: 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,
    /// **Where the projection is persisted** — `objects.index` beside the redb
    /// tail — or `None` for an in-memory stack and for an arm whose
    /// [`ObjectIndex::PERSISTS`] is false. When `Some`, every rebuild is a
    /// **fold**: the projection is written here, then the rows it holds are
    /// removed from redb, so redb is the tail in fact and not only in name.
    persisted: Option<PathBuf>,
    /// Serialises the fold against `append` and `retain`. Held for the whole
    /// of a fold, and by `append` around its transaction, so the counters a
    /// fold resets cannot be bumped underneath it by a row it did not see —
    /// which would leave `unabsorbed_rows` at 0 over a row the projection
    /// lacks, and `projection_is_complete` would then answer `absent` for an
    /// object the repository holds.
    fold: Mutex<()>,
    /// Folds that wrote `persisted`, over this stack's lifetime.
    folds: AtomicU64,
    /// Opens that took the projection off `persisted` instead of rebuilding
    /// it from a redb scan. `1` after a warm open, `0` after a cold one.
    persisted_loads: AtomicU64,
}

impl<S: ObjectIndex> ObjectReadStack<S> {
    /// True when this stack folds: it has somewhere to persist to and an arm
    /// that can be persisted.
    #[inline]
    fn persists(&self) -> bool {
        self.persisted.is_some()
    }

    /// 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()))?;
        // `objects.tail` → `objects.index`: the persisted projection sits beside
        // the redb tail it is folded from. Only for an arm that can be written
        // down; the others keep every row in redb, as before 2026-09-04.
        let persisted = S::persists().then(|| tail_path.with_extension("index"));
        Self::from_db(db, triggers, persisted, Some(tail_path.to_path_buf()))
    }

    /// Where this stack persists its projection, if it does.
    pub fn persisted_path(&self) -> Option<&Path> {
        self.persisted.as_deref()
    }

    /// **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, None, None)
    }

    fn from_db(
        db: Database,
        triggers: RebuildTriggers,
        persisted: Option<PathBuf>,
        tail_path: Option<PathBuf>,
    ) -> Result<Self> {
        // Materialise both tables so a read on a brand-new database does not
        // trip `TableDoesNotExist`.
        let w = db.begin_write()?;
        let dropped = {
            let _ = w.open_table(OBJECTS)?;
            let mut meta = w.open_table(META)?;
            let found = meta.get(META_TAIL_FORMAT)?.map(|v| v.value());
            match found {
                // Written by this layout, or by the layout in use when the key
                // was introduced (see `META_TAIL_FORMAT`): nothing to do.
                Some(TAIL_FORMAT) => 0,
                None => {
                    meta.insert(META_TAIL_FORMAT, TAIL_FORMAT)?;
                    0
                }
                // A DIFFERENT layout. Every row in OBJECTS is unreadable by
                // `decode_row`, so the tail is emptied here — in the same
                // transaction that stamps the new version, so a crash between
                // them cannot leave rows of one layout under the mark of
                // another. `adopt_journal` then sees every pack as unabsorbed
                // and requeues it.
                Some(other) => {
                    let mut objects = w.open_table(OBJECTS)?;
                    let n = objects.len()?;
                    objects.retain(|_, _| false)?;
                    meta.insert(META_TAIL_FORMAT, TAIL_FORMAT)?;
                    // The oid set changed, so the content epoch moves with it.
                    let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
                    meta.insert(META_TAIL_EPOCH, epoch + 1)?;
                    tracing::warn!(
                        target: "znippy.git",
                        found = other,
                        expected = TAIL_FORMAT,
                        rows = n,
                        "the object tail was written in a different layout; dropping it and \
                         re-indexing from the journal. Reads fall back to the pack until the \
                         drain catches up — slower, never wrong"
                    );
                    n
                }
            }
        };
        w.commit()?;
        let _ = dropped;
        let mut db = db;

        // ── the warm open ────────────────────────────────────────────────────
        //
        // If a persisted projection exists, it IS the projection: mapped, not
        // rebuilt. redb then holds only the rows that arrived after the last
        // fold, and those are served by the fall-through exactly as an
        // un-absorbed row always was. The scan below is over that remainder —
        // small by construction — not over the repository.
        //
        // The one thing to reconcile is the crash window a fold leaves: the
        // file is renamed into place BEFORE redb forgets the rows it holds, so
        // a crash between the two leaves the same row in both. Rows redb has
        // that the file also has are removed here, in one transaction, so the
        // two layers are disjoint again and `len` / `oids_in_order` count each
        // object once. The other direction — file missing rows redb dropped —
        // cannot happen, because the delete only ever follows a completed
        // rename.
        let folded: Option<(Arc<S>, u128, u128)> = match persisted.as_deref() {
            Some(path) => {
                // A `.tmp` left by a fold that died before its rename is not a
                // projection and must not become one by accident.
                let _ = fs::remove_file(tmp_path(path));
                load_persisted::<S>(path)
            }
            None => None,
        };
        let t_scan = std::time::Instant::now();
        let mut entries = scan(&db)?;
        let mut migrated = 0u64;
        let (projection, unfolded, loaded) = match folded {
            Some((folded, map_ms, build_ms)) => {
                let dup: Vec<Vec<u8>> = entries
                    .iter()
                    .filter(|e| folded.lookup(&e.oid).is_some())
                    .map(|e| e.oid.clone())
                    .collect();
                if !dup.is_empty() {
                    let w = db.begin_write()?;
                    {
                        let mut objects = w.open_table(OBJECTS)?;
                        for oid in &dup {
                            objects.remove(oid.as_slice())?;
                        }
                    }
                    w.commit()?;
                    tracing::warn!(
                        target: "znippy.git",
                        rows = dup.len(),
                        "the object tail held rows the persisted index already has — a fold \
                         was interrupted between writing the index and emptying the tail; \
                         the duplicates are dropped from the tail, nothing is lost"
                    );
                    entries.retain(|e| folded.lookup(&e.oid).is_none());
                }
                // The three costs of a warm open, separately, so the next lever
                // (persisting the stree section) is judged on the number it
                // would actually remove: `build_ms` is the O(n) stree build over
                // the mapped oid column; `map_ms` is the O(1) open+mmap;
                // `tail_ms` is the redb scan of the rows since the last fold
                // plus the reconcile. `stree_bytes` is what that build produces
                // and what persisting it would have to write.
                tracing::info!(
                    target: "znippy.git",
                    folded = folded.len(),
                    tail = entries.len(),
                    map_ms,
                    build_ms,
                    tail_ms = t_scan.elapsed().as_millis(),
                    stree_bytes = folded.resident_bytes().saturating_sub(folded.ipc_bytes()),
                    "mapped the persisted object index; the redb tail holds only what \
                     arrived after the last fold"
                );
                (folded, entries, 1u64)
            }
            None => {
                let projection = Arc::new(S::build(&entries)?);
                // ── the cold open of a store that predates persistence ──────
                //
                // No file, and redb holds rows: a store written before
                // 2026-09-04, or one whose index was lost. The projection was
                // just built from the full scan the old design paid on every
                // open; fold it now, so this is the LAST time this store pays
                // it. Nothing else holds `db` yet, so every row it has is a row
                // the projection was just built from: drop the table rather
                // than remove 14 M keys one by one. The kernel mirror pays this
                // once — 14 M rows out of redb, ~850 MB written — and never
                // again.
                if let Some(path) = persisted.as_deref()
                    && !entries.is_empty()
                {
                    write_atomic(path, projection.persisted_bytes().expect("PERSISTS arm"))?;
                    let w = db.begin_write()?;
                    w.delete_table(OBJECTS)?;
                    let _ = w.open_table(OBJECTS)?;
                    w.commit()?;
                    migrated = 1;
                    tracing::info!(
                        target: "znippy.git",
                        rows = entries.len(),
                        path = %path.display(),
                        "folded the object tail into a persisted index on first open; the \
                         redb tail now holds only rows that arrive after this"
                    );
                }
                (projection, Vec::new(), 0u64)
            }
        };

        // ── give redb its file back ──────────────────────────────────────────
        //
        // redb reuses freed pages but never shrinks its file. A store that just
        // migrated has a redb at its old high-water mark with nothing in it —
        // 4 GiB allocated, 2.9 GB of blocks, for the kernel mirror — and a store
        // that has been folding for a while sits at one fold window of dead
        // pages (270 MB on the same repository, over 54 k live rows). This is the
        // one moment the `Database` is owned and no transaction is open, so
        // compaction is free of coordination here and nowhere else.
        compact_if_fragmented(&mut db, tail_path.as_deref());

        // What the projection does not cover, counted the way `append` would
        // have counted it, so the column-scan aggregates and the completeness
        // fast path are exact from the first read.
        let total = projection.len() as u64 + unfolded.len() as u64;
        let mut unabsorbed_size = 0u64;
        let mut unabsorbed_types = [0u64; 8];
        let mut tail_bytes = 0u64;
        for e in &unfolded {
            unabsorbed_size += e.uncompressed_size;
            unabsorbed_types[e.obj_type.code() as usize] += 1;
            tail_bytes += e.len;
        }
        let db = Arc::new(db);
        let stack = 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(tail_bytes),
            rebuilds: AtomicU64::new(1),
            generation: AtomicU64::new(1),
            unabsorbed_size: AtomicU64::new(unabsorbed_size),
            unabsorbed_types: unabsorbed_types.map(AtomicU64::new),
            unabsorbed_rows: AtomicU64::new(unfolded.len() as u64),
            tail_txns: AtomicU64::new(0),
            persisted,
            fold: Mutex::new(()),
            folds: AtomicU64::new(migrated),
            persisted_loads: AtomicU64::new(loaded),
        };

        Ok(stack)
    }

    /// **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)");
        }

        // Under the fold lock for the transaction AND the counter updates, so a
        // fold cannot reset `unabsorbed_rows` between this batch landing in
        // redb and its count landing in the atomics. Released before the
        // trigger check below, because `rebuild` takes the same lock.
        let guard = self.fold.lock().map_err(|_| anyhow!("the fold lock is poisoned"))?;
        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);
            // The rows already known: the projection first, then redb. On a
            // persisting stack the projection holds every folded row and redb
            // does NOT, so a re-pushed or re-indexed object that was folded
            // would otherwise be inserted into redb as new and exist in both
            // layers — `len()` doubles, `oids_in_order` repeats, and the next
            // fold's `validate` refuses the duplicate oid. MEASURED 2026-09-04
            // on `dropping_the_exploded_table_and_reopening_still_answers`:
            // 2704 rows for a 1352-object repository, exactly ×2. The stree
            // lookup is also the cheaper check by an order of magnitude, so the
            // common case of a re-push now never opens redb for it.
            let projection = self.projection.read().expect("projection lock").clone();
            for e in entries {
                if e.oid.len() != width {
                    bail!(
                        "mixed oid widths in one append: {width} and {}",
                        e.oid.len()
                    );
                }
                let known: Option<TailRow> = match projection.lookup(&e.oid) {
                    Some(row) => Some(TailRow {
                        seq: 0,
                        offset: row.offset,
                        len: row.len,
                        obj_type: row.obj_type,
                        uncompressed_size: row.uncompressed_size,
                        delta_base: row.delta_base,
                    }),
                    None => match objects.get(e.oid.as_slice())? {
                        Some(existing) => Some(decode_row(existing.value())?),
                        None => None,
                    },
                };
                if let Some(old) = known {
                    // ── 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)?;
            // Only a genuine insertion changes the oid set, so the content epoch
            // holds still across a re-pushed pack that adds nothing (added == 0).
            // The bump rides this same transaction, so the epoch and the rows it
            // describes commit or roll back together.
            if added > 0 {
                let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
                meta.insert(META_TAIL_EPOCH, epoch + 1)?;
            }
        }
        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;
        drop(guard);
        if let Some(threshold) = self.volume_threshold()
            && total_bytes >= threshold
        {
            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 let Some(threshold) = self.volume_threshold()
            && bytes >= threshold
        {
            return Some(RebuildReason::TailBytes(bytes));
        }
        None
    }

    /// The volume trigger's threshold right now, or `None` when it is off.
    ///
    /// The fixed floor everywhere; on a persisting stack, the larger of the
    /// floor and `fold_ratio ×` the persisted index's current size — so a fold
    /// on a big repository is paid for by a proportionate amount of new data,
    /// and the sum of index writes over an import stays ≈ the bytes imported.
    /// See [`RebuildTriggers::DEFAULT_FOLD_RATIO`].
    fn volume_threshold(&self) -> Option<u64> {
        let floor = self.triggers.tail_bytes;
        if floor == 0 {
            return None;
        }
        if !self.persists() || self.triggers.fold_ratio <= 0.0 {
            return Some(floor);
        }
        let index_bytes = self.projection.read().expect("projection lock").ipc_bytes() as f64;
        Some(floor.max((self.triggers.fold_ratio * index_bytes) as u64))
    }

    /// 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 _g = self.fold.lock().map_err(|_| anyhow!("the fold lock is poisoned"))?;
        self.fold_locked(None).map(|_| ())
    }

    /// **The fold** — a rebuild that, on a persisting stack, also writes the
    /// projection down and empties redb of what it wrote. Caller holds
    /// [`fold`](Self::fold).
    ///
    /// The merge is of two disjoint sorted runs: the current projection (which,
    /// on a persisting stack, IS the file's content — the invariant every fold
    /// re-establishes) and the rows redb holds right now. On a non-persisting
    /// stack redb holds every row and the projection is not consulted, which is
    /// the pre-2026-09-04 rebuild unchanged.
    ///
    /// `live`, when given, is applied to the **projection's** rows only: those
    /// are all older than the reachability set `retain`'s caller computed, so
    /// the predicate has an answer for each of them. The redb rows are never
    /// filtered here — `retain` has already removed the dead ones under this
    /// same lock, and any row that landed after the caller's scan is one the
    /// predicate does not know and must not be asked about (`gc`'s closure
    /// answers `false` for an oid it never saw).
    ///
    /// Durability order, and why it is that order: (1) the new index goes to
    /// `objects.index.tmp`, is fsynced, and is renamed over `objects.index`;
    /// (2) the rows it holds are removed from redb. A crash after (1) and
    /// before (2) leaves them in both — reconciled at the next open. A crash
    /// during (1) leaves a `.tmp` nobody reads and a redb that still has every
    /// row. Nothing is ever lost, because a row leaves redb only after the
    /// rename that made it durable elsewhere has returned.
    ///
    /// Returns how many projection rows `live` rejected.
    fn fold_locked(&self, live: Option<&dyn Fn(&[u8]) -> bool>) -> Result<u64> {
        let unfolded = scan(&self.tail)?;
        let mut dropped = 0u64;
        let mut merged: Vec<IndexEntry>;
        if self.persists() {
            let cur = self.projection.read().expect("projection lock").clone();
            merged = Vec::with_capacity(cur.len() + unfolded.len());
            for o in 0..cur.len() as u32 {
                let e = cur.entry_at(o);
                if live.is_none_or(|f| f(&e.oid)) {
                    merged.push(e);
                } else {
                    dropped += 1;
                }
            }
        } else {
            merged = Vec::with_capacity(unfolded.len());
        }
        let forget: Vec<Vec<u8>> = if self.persists() {
            unfolded.iter().map(|e| e.oid.clone()).collect()
        } else {
            Vec::new()
        };
        merged.extend(unfolded);
        let total = merged.len() as u64;
        let fresh = Arc::new(S::build(&merged)?);
        drop(merged);

        if let Some(path) = self.persisted.as_deref() {
            write_atomic(path, fresh.persisted_bytes().expect("PERSISTS arm"))?;
            if !forget.is_empty() {
                let w = self.tail.begin_write()?;
                {
                    let mut objects = w.open_table(OBJECTS)?;
                    for oid in &forget {
                        objects.remove(oid.as_slice())?;
                    }
                }
                w.commit()?;
            }
            self.folds.fetch_add(1, Ordering::AcqRel);
        }

        *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(dropped)
    }

    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),
            folds: self.folds.load(Ordering::Acquire),
            persisted_loads: self.persisted_loads.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 tail = Vec::with_capacity(objects.len()? as usize);
        for row in objects.iter()? {
            let (k, _) = row?;
            tail.push(k.value().to_vec());
        }
        if !self.persists() {
            // redb holds every row, so its key order is the whole answer.
            return Ok(tail);
        }
        // Two sorted runs — the persisted projection and the redb tail — merged
        // once. Disjoint by the fold's invariant; the equal-key arm below exists
        // for the window inside a fold where the file is renamed and redb has
        // not yet forgotten, so a caller in that window counts each oid once.
        let cur = self.projection.read().expect("projection lock").clone();
        let n = cur.len();
        let mut out = Vec::with_capacity(n + tail.len());
        let (mut i, mut j) = (0usize, 0usize);
        while i < n || j < tail.len() {
            let a = (i < n).then(|| cur.oid_at(i as u32));
            let b = tail.get(j).map(Vec::as_slice);
            match (a, b) {
                (Some(a), Some(b)) => match a.as_slice().cmp(b) {
                    std::cmp::Ordering::Less => {
                        out.push(a);
                        i += 1;
                    }
                    std::cmp::Ordering::Greater => {
                        out.push(b.to_vec());
                        j += 1;
                    }
                    std::cmp::Ordering::Equal => {
                        out.push(a);
                        i += 1;
                        j += 1;
                    }
                },
                (Some(a), None) => {
                    out.push(a);
                    i += 1;
                }
                (None, Some(b)) => {
                    out.push(b.to_vec());
                    j += 1;
                }
                (None, None) => break,
            }
        }
        Ok(out)
    }

    /// **The ordinal space** — every oid in the repository, back to back, in oid
    /// (= ordinal) order — and where each part came from.
    ///
    /// This replaces the `objects.ordinal` memo (S-082). That file existed to
    /// spare a cold open the full redb scan it took to rebuild the sorted oid
    /// list; the scan no longer exists to be spared. On a persisting stack the
    /// folded oids ARE the projection's oid column — one contiguous copy out of
    /// the mapped buffer — and only the redb tail (the rows since the last fold)
    /// is scanned and merged in. `folded_rows` / `tail_rows` say which is which,
    /// so a test can assert the space came off the index and not off a scan
    /// without comparing two byte strings that are equal either way.
    ///
    /// Same overlap note as [`oids_in_order`](Self::oids_in_order): inside a
    /// fold's rename-then-forget window an oid may sit in both layers for a
    /// moment; the merge emits it once.
    pub fn oids_raw(&self) -> Result<OrdinalSpace> {
        let read = self.tail.begin_read()?;
        let objects = read.open_table(OBJECTS)?;
        let mut tail: Vec<Vec<u8>> = Vec::with_capacity(objects.len()? as usize);
        for row in objects.iter()? {
            let (k, _) = row?;
            tail.push(k.value().to_vec());
        }
        let tail_rows = tail.len() as u64;
        if !self.persists() {
            let mut raw = Vec::with_capacity(tail.iter().map(Vec::len).sum());
            for oid in &tail {
                raw.extend_from_slice(oid);
            }
            return Ok(OrdinalSpace { raw, folded_rows: 0, tail_rows });
        }
        let cur = self.projection.read().expect("projection lock").clone();
        let folded = cur.oids_raw();
        let folded_rows = cur.len() as u64;
        if tail.is_empty() {
            return Ok(OrdinalSpace { raw: folded, folded_rows, tail_rows: 0 });
        }
        let w = match tail.first() {
            Some(o) => o.len(),
            None => unreachable!("tail is non-empty"),
        };
        let mut raw = Vec::with_capacity(folded.len() + tail.len() * w);
        let (mut i, mut j) = (0usize, 0usize);
        let n = folded.len() / w;
        while i < n || j < tail.len() {
            let a = (i < n).then(|| &folded[i * w..(i + 1) * w]);
            let b = tail.get(j).map(Vec::as_slice);
            match (a, b) {
                (Some(a), Some(b)) => match a.cmp(b) {
                    std::cmp::Ordering::Less => {
                        raw.extend_from_slice(a);
                        i += 1;
                    }
                    std::cmp::Ordering::Greater => {
                        raw.extend_from_slice(b);
                        j += 1;
                    }
                    std::cmp::Ordering::Equal => {
                        raw.extend_from_slice(a);
                        i += 1;
                        j += 1;
                    }
                },
                (Some(a), None) => {
                    raw.extend_from_slice(a);
                    i += 1;
                }
                (None, Some(b)) => {
                    raw.extend_from_slice(b);
                    j += 1;
                }
                (None, None) => break,
            }
        }
        Ok(OrdinalSpace { raw, folded_rows, tail_rows })
    }

    /// `(tail_epoch, object_count)` in one read transaction — the two cheap facts
    /// that say whether the oid set has moved, **without** the ordered scan
    /// [`Self::oids_in_order`] pays. (They were the `objects.ordinal` memo's
    /// validity key until that file went, 2026-09-04.) `epoch` is [`META_TAIL_EPOCH`] (0 when a
    /// store predates it, e.g. was populated by an older binary); the count is
    /// redb's own O(1) table length. Both are metadata reads — a point lookup and
    /// a stored length — so this is microseconds against the scan's hundreds of
    /// milliseconds, which is the whole reason the memo can exist.
    pub fn tail_epoch_and_len(&self) -> Result<(u64, u64)> {
        let read = self.tail.begin_read()?;
        let meta = read.open_table(META)?;
        let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
        let objects = read.open_table(OBJECTS)?;
        // On a persisting stack the repository is the persisted projection plus
        // the redb tail, disjoint. Inside a fold's rename-then-forget window the
        // two overlap for a moment and this over-counts; the memo it validates
        // then reads as stale and re-scans, and the next call is exact again —
        // a transient re-scan, never a wrong answer, and not worth taking the
        // fold lock on the read path for.
        let folded = if self.persists() { self.projection_len() as u64 } else { 0 };
        Ok((epoch, objects.len()? + folded))
    }

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

        // One row's offset, placed against the sorted extents. `true` when every
        // extent has now been hit and the scan can stop.
        let mut place = |offset: u64| -> bool {
            // The last extent that starts at or before this row.
            let p = starts.partition_point(|&s| s <= offset);
            if p == 0 {
                return false;
            }
            let i = order[p - 1];
            let (start, len) = extents[i];
            if offset < start + len && !hit[i] {
                hit[i] = true;
                wanted -= 1;
            }
            wanted == 0
        };

        // The persisted projection first: on a persisting stack it holds every
        // row older than the last fold, which is nearly all of them, and its
        // offset column is one contiguous `u64` run — the cheapest scan there
        // is. Then the redb tail for the rows that arrived since.
        if self.persists() {
            let cur = self.projection.read().expect("projection lock").clone();
            for o in 0..cur.len() as u32 {
                if place(cur.extent_at(o).0) {
                    return Ok(hit);
                }
            }
        }
        let read = self.tail.begin_read()?;
        let objects = read.open_table(OBJECTS)?;
        for row in objects.iter()? {
            let (_, v) = row?;
            if place(decode_row(v.value())?.offset) {
                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> {
        // The whole of `retain` — the redb pass, the fold that rewrites the
        // persisted index without the dead rows, and the epoch bump — under one
        // hold of the fold lock, so no fold can interleave and re-persist a row
        // this call is removing.
        let _g = self.fold.lock().map_err(|_| anyhow!("the fold lock is poisoned"))?;
        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;
            }
            // A removal changes the oid set exactly as an insert does, so the
            // content epoch must record it: bump it in this same transaction
            // when anything actually went. `arrival_seq` is deliberately not
            // touched — it is an *arrival* ordinal, not a live count.
            if dropped > 0 {
                let mut meta = w.open_table(META)?;
                let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
                meta.insert(META_TAIL_EPOCH, epoch + 1)?;
            }
        }
        w.commit()?;
        // The persisted projection's dead rows go in the fold: it rewrites the
        // index from `live` rows only and swaps it in, so a stree that answered
        // for a deleted oid a moment ago answers `None` now. On a persisting
        // stack this is where most of what dies actually goes — those rows were
        // never in redb — so the epoch is bumped again for them. (`len` in the
        // memo's `(epoch, len)` key would catch it too; the bump keeps the
        // contract that a changed oid set is a changed epoch.)
        let folded_dead = self.fold_locked(Some(live))?;
        if folded_dead > 0 {
            let w = self.tail.begin_write()?;
            {
                let mut meta = w.open_table(META)?;
                let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
                meta.insert(META_TAIL_EPOCH, epoch + 1)?;
            }
            w.commit()?;
        }
        Ok(dropped + folded_dead)
    }

    /// 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)
}

// ── the persisted projection on disk ─────────────────────────────────────────

/// `objects.index` → `objects.index.tmp`: where a fold writes before its rename.
fn tmp_path(path: &Path) -> PathBuf {
    let mut name = path.file_name().map(|n| n.to_os_string()).unwrap_or_default();
    name.push(".tmp");
    path.with_file_name(name)
}

/// Write `bytes` so that `path` either holds all of them or is untouched:
/// sibling temp file, fsync, rename, fsync the directory. The same idiom
/// every durable writer in this crate uses, spelled once here for the index.
fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
    let tmp = tmp_path(path);
    {
        let mut f = File::create(&tmp)
            .with_context(|| format!("creating {}", tmp.display()))?;
        f.write_all(bytes)
            .with_context(|| format!("writing {} bytes to {}", bytes.len(), tmp.display()))?;
        f.sync_all()
            .with_context(|| format!("syncing {}", tmp.display()))?;
    }
    fs::rename(&tmp, path)
        .with_context(|| format!("renaming {} over {}", tmp.display(), path.display()))?;
    if let Some(dir) = path.parent().filter(|d| !d.as_os_str().is_empty()) {
        File::open(dir)
            .and_then(|d| d.sync_all())
            .with_context(|| format!("syncing the directory of {}", path.display()))?;
    }
    Ok(())
}

/// Dead redb pages worth compacting for: below this an open leaves the file
/// alone, because a small repository's redb is a few pages either way and a
/// rewrite of it on every open would be churn for nothing. Compaction costs
/// what is LIVE, not what is dead — after a fold that is little — so the floor
/// is set by the file lying about the store, not by the work: 16 MiB is a redb
/// that has doubled itself four times past a few thousand live rows.
const COMPACT_FRAGMENTED_FLOOR: u64 = 16 << 20;

/// Shrink redb's file when it holds at least [`COMPACT_FRAGMENTED_FLOOR`] of dead
/// pages. Needs the `Database` owned and no transaction open, which is true at
/// exactly one place — the open — and nowhere else; that is why it lives here
/// rather than behind a maintain verb. Any failure is logged and swallowed: the
/// file being large is not a fault, and a store must open over it regardless.
fn compact_if_fragmented(db: &mut Database, tail_path: Option<&Path>) {
    let Some(tail_path) = tail_path else { return }; // in-memory: nothing on disk to shrink
    let file_len = match fs::metadata(tail_path) {
        Ok(m) => m.len(),
        Err(_) => return,
    };
    // `stats()` lives on the write transaction in redb 2; this one is opened to
    // be read and aborted, never committed. The dead-page figure is taken from
    // the FILE, not from `fragmented_bytes()`: right after a commit that freed
    // pages (the migration fold's `delete_table`) redb still has them PENDING
    // — two-phase free — and reports them as neither stored nor fragmented.
    // MEASURED: the same 34 MB file read `fragmented = 12.8 MiB` inside the
    // open and `33 MB` from a fresh handle a moment later. The file length
    // minus what is stored is what the store is actually lying by, and
    // `compact()` begins with the two commits that turn pending into free.
    let (stored, metadata) = {
        let w = match db.begin_write() {
            Ok(w) => w,
            Err(e) => {
                tracing::debug!(target: "znippy.git", error = %e, "redb stats unavailable; not compacting");
                return;
            }
        };
        let s = w.stats().map(|s| (s.stored_bytes(), s.metadata_bytes()));
        let _ = w.abort();
        match s {
            Ok(s) => s,
            Err(e) => {
                tracing::debug!(target: "znippy.git", error = %e, "redb stats unavailable; not compacting");
                return;
            }
        }
    };
    let dead = file_len.saturating_sub(stored + metadata);
    if dead < COMPACT_FRAGMENTED_FLOOR {
        return;
    }
    let t = std::time::Instant::now();
    match db.compact() {
        Ok(did) => tracing::info!(
            target: "znippy.git",
            file_before = file_len,
            file_after = fs::metadata(tail_path).map(|m| m.len()).unwrap_or(0),
            stored_bytes = stored,
            compacted = did,
            ms = t.elapsed().as_millis(),
            tail = %tail_path.display(),
            "compacted the redb object tail: it was carrying dead pages from before a fold"
        ),
        Err(e) => tracing::warn!(
            target: "znippy.git",
            file_len,
            stored_bytes = stored,
            error = %e,
            "could not compact the redb object tail; it keeps its size, nothing else changes"
        ),
    }
}

/// The persisted index as an Arrow [`Buffer`] — `None` when there is none.
///
/// Memory-mapped where the platform allows it, so the columns
/// [`ObjectIndex::from_persisted`] decodes are views into the page cache: the
/// index costs resident memory only for the pages a lookup touches, and none
/// of it is a copy. Elsewhere the file is read into an aligned buffer, which
/// keeps every other property (no rebuild, no redb scan) and gives up only
/// the sharing.
fn map_persisted(path: &Path) -> Result<Option<Buffer>> {
    let f = match File::open(path) {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(e).with_context(|| format!("opening {}", path.display())),
    };
    let len = f
        .metadata()
        .with_context(|| format!("stat {}", path.display()))?
        .len();
    if len == 0 {
        // A persisted projection is never zero bytes — even an empty index
        // carries its schema — so this is a truncated write, not an index.
        return Ok(None);
    }
    #[cfg(target_os = "linux")]
    {
        // SAFETY: the file is replaced only by `write_atomic`'s rename, which
        // swaps the directory entry and never writes into the inode this
        // mapping is over; the old inode lives until the last mapping drops.
        let map = unsafe { memmap2::Mmap::map(&f) }
            .with_context(|| format!("mapping {}", path.display()))?;
        let map = Arc::new(map);
        let ptr = std::ptr::NonNull::new(map.as_ptr() as *mut u8)
            .ok_or_else(|| anyhow!("mmap of {} returned a null pointer", path.display()))?;
        // SAFETY: `ptr` and `len` describe exactly the mapping `map` owns, and
        // `map` is kept alive inside the buffer as its allocation owner.
        let buf = unsafe { Buffer::from_custom_allocation(ptr, map.len(), map) };
        Ok(Some(buf))
    }
    #[cfg(not(target_os = "linux"))]
    {
        use std::io::Read;
        let mut f = f;
        let mut mb = znippy_common::arrow::buffer::MutableBuffer::with_capacity(len as usize);
        let mut v = Vec::with_capacity(len as usize);
        f.read_to_end(&mut v)
            .with_context(|| format!("reading {}", path.display()))?;
        mb.extend_from_slice(&v);
        Ok(Some(mb.into()))
    }
}

/// The projection at `path`, or `None` if there is none — or if what is there
/// cannot be read as one. The second case is a warning and not an error on
/// purpose: the index is a cache the journal can rebuild, and a store that
/// refuses to open over a damaged cache would be a store that has stopped
/// answering over something it knows how to recover from. The rows redb still
/// holds are served; the rest are re-indexed from the journal by
/// `adopt_journal`, which sees them as "extent journaled, rows absent".
/// Returns the projection and two wall-clock figures in milliseconds: the map
/// (open + mmap, which is O(1)) and the decode-plus-stree build (which is O(n)
/// in the oid count and is the whole of a warm open's real cost). They are
/// logged by the open, so the number the next lever has to beat is in the
/// server log rather than in a bench nobody reruns.
fn load_persisted<S: ObjectIndex>(path: &Path) -> Option<(Arc<S>, u128, u128)> {
    let t0 = std::time::Instant::now();
    match map_persisted(path) {
        Ok(None) => None,
        Ok(Some(buf)) => {
            let map_ms = t0.elapsed().as_millis();
            let t1 = std::time::Instant::now();
            match S::from_persisted(buf) {
                Ok(s) => Some((Arc::new(s), map_ms, t1.elapsed().as_millis())),
                Err(e) => {
                tracing::warn!(
                    target: "znippy.git",
                    path = %path.display(),
                    error = %format!("{e:#}"),
                    "the persisted object index could not be read; opening without it and \
                     re-indexing from the journal — slower until the drain catches up, \
                     never wrong"
                );
                    None
                }
            }
        }
        Err(e) => {
            tracing::warn!(
                target: "znippy.git",
                path = %path.display(),
                error = %format!("{e:#}"),
                "the persisted object index could not be opened; opening without it"
            );
            None
        }
    }
}

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"
    }

    /// Positional rows are the **projection's** rows. A tail ordinal carries
    /// [`TAIL_ORDINAL_BIT`] and is a row address in no table — asking for it
    /// positionally is the same programming error as indexing past `len()`,
    /// and fails the same way, out loud.
    fn oid_at(&self, ordinal: u32) -> Vec<u8> {
        assert!(!is_tail_ordinal(ordinal), "ordinal {ordinal:#x} is tail-served, not positional");
        self.projection.read().expect("projection lock").oid_at(ordinal)
    }

    fn entry_at(&self, ordinal: u32) -> IndexEntry {
        assert!(!is_tail_ordinal(ordinal), "ordinal {ordinal:#x} is tail-served, not positional");
        self.projection.read().expect("projection lock").entry_at(ordinal)
    }

    fn extent_at(&self, ordinal: u32) -> (u64, u64) {
        assert!(!is_tail_ordinal(ordinal), "ordinal {ordinal:#x} is tail-served, not positional");
        self.projection.read().expect("projection lock").extent_at(ordinal)
    }

    /// 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,
            fold_ratio: 0.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();
        // **What "warm" means since 2026-09-04.** This used to assert
        // `projection_len() == 90`: the old open rebuilt the projection from a
        // full redb scan, so everything was in it. Now the open MAPS the folded
        // index (the 40 rows the rebuild persisted) and leaves the 50 that
        // arrived after it in redb, where the fall-through serves them — the
        // same routing an un-absorbed row always had. The repository is still
        // 90 rows and every one of them answers; what changed is that nothing
        // was rebuilt to make that so, and `persisted_loads` is the applied
        // output that says so.
        assert_eq!(stack.len(), 90, "the repository is {} of 90 rows", stack.len());
        assert_eq!(stack.projection_len(), 40, "the mapped index holds what the fold wrote");
        assert_eq!(stack.stats().persisted_loads, 1, "the open must have mapped, not rebuilt");
        assert_eq!(stack.stats().unabsorbed_rows, 50, "the 50 later rows are redb's to serve");
        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 fold ─────────────────────────────────────────────────────────────

    /// A file-backed stack over `all[..40]` folded, `all[40..]` appended after.
    /// Returns the tail path and the index path.
    fn folded_then_appended(
        dir: &Path,
        all: &[IndexEntry],
        split: usize,
    ) -> (PathBuf, PathBuf, ObjectReadStack<OneTableFourColumns>) {
        let tail = dir.join("objects.tail");
        let stack = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        stack.append(&all[..split]).unwrap();
        stack.rebuild().unwrap();
        stack.append(&all[split..]).unwrap();
        let index = stack.persisted_path().expect("the default arm persists").to_path_buf();
        (tail, index, stack)
    }

    /// How many rows redb holds right now — the thing the fold exists to keep
    /// small, read straight off the table so the test cannot be fooled by the
    /// stack's own bookkeeping.
    fn redb_rows(stack: &ObjectReadStack<OneTableFourColumns>) -> u64 {
        let r = stack.tail.begin_read().unwrap();
        r.open_table(OBJECTS).unwrap().len().unwrap()
    }

    /// **A rebuild on a file-backed stack is a fold**: the projection lands in
    /// `objects.index`, and redb forgets the rows it holds. Asserted on the
    /// file and on redb's own row count, not on the stack's counters.
    ///
    /// Seen RED by skipping the `objects.remove` loop in `fold_locked`: redb
    /// kept 40 rows after the fold ("redb must hold only what arrived after the
    /// fold / left: 40 / right: 0"). Seen RED by skipping `write_atomic`: the
    /// index file was absent. Restored.
    #[test]
    fn a_rebuild_on_a_file_backed_stack_is_a_fold() {
        let dir = tempfile::tempdir().unwrap();
        let all = synthetic_entries(90, 20, 43);
        let (_tail, index, stack) = folded_then_appended(dir.path(), &all, 40);

        assert!(index.is_file(), "the fold must persist the projection at {}", index.display());
        assert!(
            fs::metadata(&index).unwrap().len() > 40 * 61,
            "the persisted index must carry at least the rows' own bytes"
        );
        assert_eq!(stack.stats().folds, 1);
        assert_eq!(redb_rows(&stack), 50, "redb must hold only what arrived after the fold");
        assert_eq!(stack.projection_len(), 40);
        assert_eq!(stack.len(), 90);
        assert!(!stack.projection_is_complete());
        for e in &all {
            let row = stack.lookup(&e.oid).unwrap_or_else(|| panic!("lost {}", hex::encode(&e.oid)));
            assert_eq!((row.offset, row.len, row.delta_base), (e.offset, e.len, e.delta_base));
        }
        // The tail rows are the tail's: served with the tail bit, as before.
        assert!(is_tail_row(&stack.lookup(&all[80].oid).unwrap()));
        assert!(!is_tail_row(&stack.lookup(&all[10].oid).unwrap()));

        // A second fold takes the 50 with it and redb is empty.
        stack.rebuild().unwrap();
        assert_eq!(redb_rows(&stack), 0);
        assert_eq!(stack.projection_len(), 90);
        assert!(stack.projection_is_complete());
        assert_eq!(stack.stats().folds, 2);
    }

    /// **Re-appending a folded row creates nothing.** `append` dedupes against
    /// the projection AND redb, because on a persisting stack the folded rows
    /// are in the projection only — a check against redb alone would insert
    /// every re-pushed or re-indexed object a second time, into redb, and the
    /// two layers would each hold it.
    ///
    /// Seen RED before the projection check existed, on the real store:
    /// `git_ops::tests::dropping_the_exploded_table_and_reopening_still_answers`
    /// re-indexed a pack from the journal after a reopen and the row count
    /// read **2704 for 1352 objects** — exactly twice. Seen RED here by
    /// consulting redb only: `redb_rows` read 40 where 0 was expected and
    /// `len()` read 80. Restored.
    #[test]
    fn re_appending_a_folded_row_creates_no_second_row() {
        let dir = tempfile::tempdir().unwrap();
        let all = synthetic_entries(40, 20, 50);
        let tail = dir.path().join("objects.tail");
        let stack = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        stack.append(&all).unwrap();
        stack.rebuild().unwrap();
        assert_eq!(redb_rows(&stack), 0, "precondition: everything folded");

        // The same objects again — a re-push, or the journal re-indexing a
        // pack after a reopen. Placement may differ; identity must agree.
        let mut again = all.clone();
        for e in &mut again {
            e.offset += 1_000_000;
        }
        stack.append(&again).unwrap();
        assert_eq!(redb_rows(&stack), 0, "a folded row re-appended must not land in redb");
        assert_eq!(stack.len(), 40);
        assert_eq!(stack.oids_in_order().unwrap().len(), 40);
        // And the first writer's placement stands, as `append`'s contract says.
        assert_eq!(stack.lookup(&all[3].oid).unwrap().offset, all[3].offset);

        // The identity check still bites through the projection path: a
        // different inflated size for a folded oid is refused, not inserted.
        let mut lie = all[7].clone();
        lie.uncompressed_size += 1;
        let err = stack.append(&[lie]).unwrap_err().to_string();
        assert!(err.contains("identity violation"), "{err}");
        assert_eq!(redb_rows(&stack), 0);
    }

    /// **A warm open maps the index and rebuilds nothing.** The proof is
    /// `persisted_loads == 1` plus a projection whose size is exactly what the
    /// fold wrote — a rebuilt projection would hold all 90.
    ///
    /// Seen RED by making `load_persisted` return `None` unconditionally: the
    /// open fell back to the scan-and-build, `persisted_loads` read 0, and the
    /// projection held 90. Restored.
    #[test]
    fn a_warm_open_maps_the_folded_index_and_does_not_rebuild() {
        let dir = tempfile::tempdir().unwrap();
        let all = synthetic_entries(90, 20, 44);
        let (tail, _index, stack) = folded_then_appended(dir.path(), &all, 60);
        drop(stack);

        let again = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        let s = again.stats();
        assert_eq!(s.persisted_loads, 1, "the open must have taken the index off disk");
        assert_eq!(s.folds, 0, "a warm open folds nothing — there is nothing to migrate");
        assert_eq!(again.projection_len(), 60, "the mapped projection is what the fold wrote");
        assert_eq!(again.len(), 90);
        assert_eq!(s.unabsorbed_rows, 30);
        assert_eq!(redb_rows(&again), 30);
        // The column-scan aggregates are exact from the first read — computed
        // over the redb remainder at open, not left at zero.
        let expect_size: u64 = all.iter().map(|e| e.uncompressed_size).sum();
        assert_eq!(again.sum_uncompressed(), expect_size);
        for t in [ObjType::Commit, ObjType::Tree, ObjType::Blob, ObjType::OfsDelta] {
            assert_eq!(
                again.count_type(t),
                all.iter().filter(|e| e.obj_type == t).count(),
                "count_type({t:?}) over both layers"
            );
        }
        for e in &all {
            let row = again.lookup(&e.oid).unwrap_or_else(|| panic!("lost {}", hex::encode(&e.oid)));
            assert_eq!(row.uncompressed_size, e.uncompressed_size);
        }
    }

    /// **The crash window a fold leaves is reconciled at open.** The fold
    /// renames the index into place and THEN forgets the rows in redb; a crash
    /// between the two leaves every folded row in both. Staged here by putting
    /// the folded rows back into redb by hand after a fold, exactly as that
    /// crash would leave them.
    ///
    /// Seen RED by removing the `dup` pass in `from_db`: `oids_in_order` came
    /// back with 90 entries for a 60-object repository and `len()` read 90.
    /// Restored.
    #[test]
    fn an_interrupted_fold_is_reconciled_at_the_next_open() {
        let dir = tempfile::tempdir().unwrap();
        let all = synthetic_entries(60, 20, 45);
        let (tail, _index, stack) = folded_then_appended(dir.path(), &all, 30);
        // Put the 30 folded rows back — the state after "renamed, then died".
        {
            let w = stack.tail.begin_write().unwrap();
            {
                let mut objects = w.open_table(OBJECTS).unwrap();
                for (i, e) in all[..30].iter().enumerate() {
                    objects.insert(e.oid.as_slice(), encode_row(1000 + i as u64, e).as_slice()).unwrap();
                }
            }
            w.commit().unwrap();
        }
        assert_eq!(redb_rows(&stack), 60, "precondition: every row is now in both layers");
        drop(stack);

        let again = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(redb_rows(&again), 30, "the duplicates must leave redb at open");
        assert_eq!(again.len(), 60);
        assert_eq!(again.tail_epoch_and_len().unwrap().1, 60);
        let oids = again.oids_in_order().unwrap();
        assert_eq!(oids.len(), 60, "each object once");
        assert!(oids.windows(2).all(|w| w[0] < w[1]), "oid order, no repeats");
    }

    /// **The tail-wide answers cover both layers.** `oids_in_order` is the
    /// merge of the persisted projection and redb; `tail_epoch_and_len` counts
    /// both; `extents_with_rows` finds a row whichever layer holds it.
    ///
    /// Seen RED by returning only redb's keys from `oids_in_order` on a
    /// persisting stack: 50 of 90. Seen RED by scanning only redb in
    /// `extents_with_rows`: the folded extent read `false`. Restored.
    #[test]
    fn the_tail_wide_answers_cover_both_layers() {
        let dir = tempfile::tempdir().unwrap();
        let all = synthetic_entries(90, 20, 46);
        let (_tail, _index, stack) = folded_then_appended(dir.path(), &all, 40);

        let mut expect: Vec<Vec<u8>> = all.iter().map(|e| e.oid.clone()).collect();
        expect.sort();
        assert_eq!(stack.oids_in_order().unwrap(), expect, "the merge of both layers, in oid order");
        assert_eq!(stack.tail_epoch_and_len().unwrap().1, 90);

        // One extent that only the persisted half can answer, one only redb can,
        // and one nobody can.
        let folded = &all[3];
        let tailed = &all[77];
        let nowhere = (u64::MAX / 2, 7u64);
        let hit = stack
            .extents_with_rows(&[(folded.offset, folded.len), (tailed.offset, tailed.len), nowhere])
            .unwrap();
        assert_eq!(hit, vec![true, true, false], "folded / tail / absent");
    }

    /// **The ordinal space is the merge of both layers, and says which is which.**
    /// After a fold with nothing appended, every oid comes off the index and
    /// none off redb; after more appends the tail rows are counted as such; in
    /// every state the bytes equal `oids_in_order` concatenated.
    ///
    /// Seen RED by returning the redb tail alone on a persisting stack:
    /// `folded_rows` read 0 where 40 was expected and the space held 50 of 90
    /// oids. Restored.
    #[test]
    fn the_ordinal_space_is_both_layers_with_provenance() {
        let dir = tempfile::tempdir().unwrap();
        let all = synthetic_entries(90, 20, 51);
        let (_tail, _index, stack) = folded_then_appended(dir.path(), &all, 40);

        let expect: Vec<u8> = stack.oids_in_order().unwrap().concat();
        let space = stack.oids_raw().unwrap();
        assert_eq!(space.raw, expect, "the space must be oids_in_order, concatenated");
        assert_eq!((space.folded_rows, space.tail_rows), (40, 50));
        assert_eq!(space.raw.len(), 90 * 20);

        stack.rebuild().unwrap();
        let space = stack.oids_raw().unwrap();
        assert_eq!(space.raw, expect, "a fold does not change the space");
        assert_eq!((space.folded_rows, space.tail_rows), (90, 0), "everything off the index now");
    }

    /// **On a persisting stack the volume trigger scales with the index.** With
    /// a 1-byte floor and `fold_ratio = 1`, appending fewer object bytes than the
    /// persisted index is long does not fold; appending past it does. On an
    /// in-memory stack the same triggers fold at the floor, because there a
    /// rebuild writes nothing and the fixed cadence is right.
    ///
    /// Seen RED by making `volume_threshold` return the floor unconditionally:
    /// the small append folded ("a small append must not fold a big index /
    /// left: 2 / right: 1"). Restored.
    #[test]
    fn the_volume_trigger_scales_with_the_persisted_index() {
        let dir = tempfile::tempdir().unwrap();
        let mut all = synthetic_entries(400, 20, 52);
        for e in &mut all {
            e.len = 100;
        }
        let triggers = RebuildTriggers {
            tail_bytes: 1,
            fold_ratio: 1.0,
            ..RebuildTriggers::manual()
        };
        let stack = ObjectReadStack::<OneTableFourColumns>::open(
            &dir.path().join("objects.tail"),
            triggers,
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        // The first append trips the 1-byte floor at once: there is no index yet
        // to scale against. After it the index is 300 rows ≈ 16 kB of IPC.
        stack.append(&all[..300]).unwrap();
        let folds = stack.stats().folds;
        assert_eq!(folds, 1, "the first batch folds at the floor");
        let index_bytes = stack.ipc_bytes() as u64;
        assert!(index_bytes > 10_000, "precondition: a real index, got {index_bytes} B");

        // 50 rows × 100 B = 5 000 B of object bytes: well under the index size.
        stack.append(&all[300..350]).unwrap();
        assert_eq!(stack.stats().folds, folds, "a small append must not fold a big index");
        assert_eq!(stack.stats().unabsorbed_rows, 50, "the 50 rows wait in redb");

        // Enough more to pass the index size: 50 rows × 100 B on top → 10 000 B
        // is still under 16 kB, so lift the lens for the last batch instead.
        let mut big: Vec<IndexEntry> = all[350..400].to_vec();
        for e in &mut big {
            e.len = index_bytes; // one row is enough to cross it
        }
        stack.append(&big[..1]).unwrap();
        assert_eq!(stack.stats().folds, folds + 1, "crossing the index size folds");
        assert!(stack.projection_is_complete());

        // The same triggers on an in-memory stack: the floor alone decides.
        let mem = ObjectReadStack::<OneTableFourColumns>::in_memory(triggers).unwrap();
        mem.append(&all[..300]).unwrap();
        let g = mem.stats().generation;
        mem.append(&all[300..350]).unwrap();
        assert!(
            mem.stats().generation > g,
            "without persistence the fixed floor still rebuilds on every batch"
        );
        assert_eq!(mem.stats().folds, 0, "and nothing is written");
    }

    /// **`retain` rewrites the persisted index without the dead rows**, whichever
    /// layer they were in, and the count it returns is both halves.
    ///
    /// Seen RED by passing `None` instead of `Some(live)` to `fold_locked` from
    /// `retain`: the dead folded oid was still answered after the call, and the
    /// count was 1 where 2 rows had died. Restored.
    #[test]
    fn retain_rewrites_the_persisted_index_without_the_dead_rows() {
        let dir = tempfile::tempdir().unwrap();
        let all = synthetic_entries(90, 20, 47);
        let (tail, index, stack) = folded_then_appended(dir.path(), &all, 40);
        let dead_folded = all[5].oid.clone();
        let dead_tailed = all[70].oid.clone();
        let before = fs::metadata(&index).unwrap().len();
        let epoch_before = stack.tail_epoch_and_len().unwrap().0;

        let dropped = stack
            .retain(&|oid: &[u8]| oid != dead_folded.as_slice() && oid != dead_tailed.as_slice())
            .unwrap();
        assert_eq!(dropped, 2, "one from each layer");
        assert!(stack.lookup(&dead_folded).is_none(), "the folded dead row must be gone");
        assert!(stack.lookup(&dead_tailed).is_none(), "the tail dead row must be gone");
        assert_eq!(stack.len(), 88);
        assert_eq!(stack.projection_len(), 88, "retain folds: everything live is in the index");
        assert_eq!(redb_rows(&stack), 0);
        assert!(
            fs::metadata(&index).unwrap().len() > before,
            "the index was rewritten (88 rows now, was 40)"
        );
        assert!(
            stack.tail_epoch_and_len().unwrap().0 > epoch_before,
            "a changed oid set is a changed epoch"
        );
        for e in all.iter().filter(|e| e.oid != dead_folded && e.oid != dead_tailed) {
            assert!(stack.lookup(&e.oid).is_some(), "live row {} lost", hex::encode(&e.oid));
        }

        // And it survives a reopen off the rewritten file.
        drop(stack);
        let again = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(again.stats().persisted_loads, 1);
        assert_eq!(again.len(), 88);
        assert!(again.lookup(&dead_folded).is_none());
    }

    /// **A damaged index is ignored, loudly, and the store still opens.** The
    /// rows redb has are served; the rest are the journal's to re-index. What
    /// this asserts is the shape of the failure: open succeeds, the projection
    /// is empty, redb's rows answer, and nothing panicked on a half-written file.
    ///
    /// Seen RED by making `load_persisted` propagate the decode error: the open
    /// failed. Restored.
    #[test]
    fn a_damaged_persisted_index_is_ignored_and_redb_still_answers() {
        let dir = tempfile::tempdir().unwrap();
        let all = synthetic_entries(60, 20, 48);
        let (tail, index, stack) = folded_then_appended(dir.path(), &all, 30);
        drop(stack);
        // Truncate the index mid-stream: an IPC prefix, not an index.
        let bytes = fs::read(&index).unwrap();
        fs::write(&index, &bytes[..bytes.len() / 2]).unwrap();

        let again = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(again.stats().persisted_loads, 0, "a damaged index must not count as loaded");
        assert_eq!(again.len(), 30, "the 30 redb rows are what the store knows");
        for e in &all[30..] {
            assert!(again.lookup(&e.oid).is_some(), "redb row {} lost", hex::encode(&e.oid));
        }
        for e in &all[..30] {
            assert!(
                again.lookup(&e.oid).is_none(),
                "a row only the damaged index held must read absent, so the journal re-indexes it"
            );
        }
        // And the store keeps working: the next fold overwrites the damage.
        again.append(&all[..30]).unwrap();
        again.rebuild().unwrap();
        assert_eq!(again.len(), 60);
        drop(again);
        let third = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(third.stats().persisted_loads, 1);
        assert_eq!(third.len(), 60);
    }

    /// **A store from before persistence folds once, at its first open, and
    /// never pays the scan again.** Staged by building a redb-only tail (the
    /// old shape: every row in redb, no index file) and opening it.
    ///
    /// Seen RED by removing the cold-open fold in `from_db`: `folds` read 0,
    /// the index file was absent, and a second open rebuilt from redb again
    /// (`persisted_loads == 0`). Restored.
    #[test]
    fn a_cold_open_of_a_pre_fold_store_folds_it_once() {
        let dir = tempfile::tempdir().unwrap();
        let tail = dir.path().join("objects.tail");
        let all = synthetic_entries(70, 20, 49);
        // The pre-2026-09-04 shape, written by hand: rows in redb, stamped with
        // this layout, no index anywhere.
        {
            let db = Database::create(&tail).unwrap();
            let w = db.begin_write().unwrap();
            {
                let mut objects = w.open_table(OBJECTS).unwrap();
                for (i, e) in all.iter().enumerate() {
                    objects.insert(e.oid.as_slice(), encode_row(i as u64, e).as_slice()).unwrap();
                }
                let mut meta = w.open_table(META).unwrap();
                meta.insert(META_TAIL_FORMAT, TAIL_FORMAT).unwrap();
                meta.insert(META_ARRIVAL_SEQ, all.len() as u64).unwrap();
            }
            w.commit().unwrap();
        }
        let index = tail.with_extension("index");
        assert!(!index.exists(), "precondition: no persisted index");

        let stack = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(stack.stats().persisted_loads, 0, "a cold open has nothing to map");
        assert_eq!(stack.stats().folds, 1, "and folds exactly once");
        assert!(index.is_file(), "the fold must have written {}", index.display());
        assert_eq!(redb_rows(&stack), 0, "redb is emptied by the migration fold");
        assert_eq!(stack.len(), 70);
        assert!(stack.projection_is_complete());
        for e in &all {
            assert!(stack.lookup(&e.oid).is_some(), "migrated row {} lost", hex::encode(&e.oid));
        }
        drop(stack);

        let again = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(again.stats().persisted_loads, 1, "the second open is warm");
        assert_eq!(again.stats().folds, 0);
        assert_eq!(again.len(), 70);
    }

    /// **A migrated store gets its redb file back.** A pre-fold store big enough
    /// to matter (here 200 k rows, ~34 MB of redb) is opened once: the migration
    /// fold empties redb, and the open compacts the file because it now carries
    /// more than [`COMPACT_FRAGMENTED_FLOOR`] of dead pages. Asserted on the
    /// file's size on disk, not on a flag. A small store (the 70-row fixture next
    /// door) is deliberately left alone — that case is covered by the floor, and
    /// this test's precondition asserts the fixture is above it.
    ///
    /// Seen RED by making `compact_if_fragmented` return before `db.compact()`:
    /// "the migrated redb must shrink" with equal before/after sizes. Restored.
    #[test]
    fn a_migrated_store_gets_its_redb_file_back() {
        let dir = tempfile::tempdir().unwrap();
        let tail = dir.path().join("objects.tail");
        let n = 200_000usize;
        let all = synthetic_entries(n, 20, 53);
        {
            let db = Database::create(&tail).unwrap();
            let w = db.begin_write().unwrap();
            {
                let mut objects = w.open_table(OBJECTS).unwrap();
                for (i, e) in all.iter().enumerate() {
                    objects.insert(e.oid.as_slice(), encode_row(i as u64, e).as_slice()).unwrap();
                }
                let mut meta = w.open_table(META).unwrap();
                meta.insert(META_TAIL_FORMAT, TAIL_FORMAT).unwrap();
                meta.insert(META_ARRIVAL_SEQ, n as u64).unwrap();
            }
            w.commit().unwrap();
        }
        let before = fs::metadata(&tail).unwrap().len();
        assert!(
            before >= COMPACT_FRAGMENTED_FLOOR,
            "precondition: the fixture must be above the compaction floor, is {before} B"
        );

        let stack = ObjectReadStack::<OneTableFourColumns>::open(
            &tail,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(stack.stats().folds, 1, "the migration fold ran");
        assert_eq!(redb_rows(&stack), 0);
        let after = fs::metadata(&tail).unwrap().len();
        assert!(
            after < before / 4,
            "the migrated redb must shrink: {after} >= {before} / 4"
        );
        // And it still works, in both directions.
        assert!(stack.lookup(&all[123].oid).is_some());
        stack.append(&synthetic_entries(10, 20, 54)).unwrap();
        assert_eq!(redb_rows(&stack), 10);
    }

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

    /// **A tail written in a different layout is DROPPED and rebuilt, not
    /// refused for ever.**
    ///
    /// `decode_row` refuses an unknown row width on purpose — reinterpreting one
    /// puts a wrong offset into an index that is then queried and believed. But
    /// until `META_TAIL_FORMAT` existed the refusal had nothing behind it: every
    /// read `bail`ed and the only way back was deleting the repository.
    ///
    /// The tail is a cache, so the way back is to empty it and let
    /// `Absorber::adopt_journal` requeue every pack — the same path a crash
    /// mid-absorb already takes.
    ///
    /// RED-FIRST: with the version check removed, the reopened stack keeps the
    /// 90 foreign rows and the first `get` through them fails on the row width
    /// instead. The assertion is on the APPLIED state — the row count and a
    /// working `append`/`get` afterwards — not on a returned flag.
    #[test]
    fn a_tail_in_a_foreign_layout_is_dropped_and_the_store_keeps_working() {
        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).unwrap();
            assert_eq!(stack.tail_epoch_and_len().unwrap().1, 90);
        }

        // Stamp a version this build does not know, exactly as a future layout
        // change would leave behind.
        {
            let db = Database::create(&path).unwrap();
            let w = db.begin_write().unwrap();
            {
                let mut meta = w.open_table(META).unwrap();
                meta.insert(META_TAIL_FORMAT, TAIL_FORMAT + 7).unwrap();
            }
            w.commit().unwrap();
        }

        let stack = ObjectReadStack::<OneTableFourColumns>::open(
            &path,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(
            stack.tail_epoch_and_len().unwrap().1,
            0,
            "a foreign-layout tail was kept; the next read would bail on the row width"
        );
        assert_eq!(stack.projection_len(), 0);

        // And the store is USABLE, which is the half that says "dropped" and not
        // "broken": the same oids re-appended come back.
        stack.append(&all).unwrap();
        assert_eq!(stack.tail_epoch_and_len().unwrap().1, 90);
        let got = stack.lookup(&all[7].oid).expect("re-appended row");
        assert_eq!(got.offset, all[7].offset);

        // The stamp is now this build's, so a second reopen drops nothing.
        drop(stack);
        let again = ObjectReadStack::<OneTableFourColumns>::open(
            &path,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(
            again.tail_epoch_and_len().unwrap().1,
            90,
            "the rewritten version did not stick, so every open would drop the tail"
        );
    }

    /// A tail written BEFORE the version key existed carries no stamp, and must
    /// be adopted rather than dropped: it holds rows of exactly this layout, and
    /// rebuilding every deployed store for a key that was merely absent would be
    /// a self-inflicted re-index of the whole fleet.
    #[test]
    fn a_tail_with_no_version_stamp_is_adopted_and_stamped_not_dropped() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("objects.tail.redb");
        let all = synthetic_entries(50, 20, 41);
        {
            let stack = ObjectReadStack::<OneTableFourColumns>::open(
                &path,
                RebuildTriggers::manual(),
                DEFAULT_REDB_CACHE_BYTES,
            )
            .unwrap();
            stack.append(&all).unwrap();
        }
        // Remove the stamp: this is what every tail on disk looked like before
        // the key existed.
        {
            let db = Database::create(&path).unwrap();
            let w = db.begin_write().unwrap();
            {
                let mut meta = w.open_table(META).unwrap();
                meta.remove(META_TAIL_FORMAT).unwrap();
            }
            w.commit().unwrap();
        }
        let stack = ObjectReadStack::<OneTableFourColumns>::open(
            &path,
            RebuildTriggers::manual(),
            DEFAULT_REDB_CACHE_BYTES,
        )
        .unwrap();
        assert_eq!(stack.tail_epoch_and_len().unwrap().1, 50, "an unstamped tail was dropped");
        let r = stack.tail.begin_read().unwrap();
        let meta = r.open_table(META).unwrap();
        assert_eq!(
            meta.get(META_TAIL_FORMAT).unwrap().map(|v| v.value()),
            Some(TAIL_FORMAT),
            "the tail was adopted but not stamped, so it is re-examined for ever"
        );
    }
}