znippy_plugin_git/

arms.rs

//! **Which implementation of each trait a store is built on** — one selector,
//! read once, never per operation.
//!
//! Three traits in this crate have more than one implementation, every one of
//! them written and benchmarked (PLAN §7, §8):
//!
//! | trait | arms | what varies |
//! |---|---|---|
//! | [`ArchiveWrite`] | [`WriterArm::Fast`] / [`Safe`](WriterArm::Safe) / [`Uring`](WriterArm::Uring) | **the durability contract**, and nothing else |
//! | [`ObjectIndex`](crate::index_layout::ObjectIndex) | [`IndexArm::OneTableFourColumns`] / [`FourTables`](IndexArm::FourTables) / [`PackedPayload`](IndexArm::PackedPayload) | the payload layout behind the same `stree` |
//! | [`Gc`] | [`GcArm::NewGeneration`] / [`CompactInPlace`](GcArm::CompactInPlace) | whether the old generation survives until the new one is proven |
//!
//! Until this module existed, `GitStore` named one of each by hand, so none of
//! them could be A/B'd through a server or a bench without editing the source.
//! [`StoreConfig`] is that choice made data.
//!
//! # The default is the shipping combination and it did not move
//!
//! [`StoreConfig::DEFAULT`] is `SafeWriter` + `ObjectReadStack<OneTableFourColumns>`
//! + `NewGeneration`, which is exactly what
//! [`GitStore::open`](crate::git_ops::GitStore::open) built before this module
//! and is exactly what it builds now.
//! **`open` and `open_with` do not read the environment at all** — an operator
//! who exports a variable cannot silently change the durability contract under a
//! caller that never asked for a selector. Choosing an arm is something a caller
//! does on purpose, through
//! [`open_with_arms`](crate::git_ops::GitStore::open_with_arms) or
//! [`open_from_env`](crate::git_ops::open_from_env).
//!
//! # Read once, at construction. Never per operation.
//!
//! Every `getenv` this crate performs goes through [`read_env`] and happens
//! either inside [`StoreConfig::from_env`] (which a store calls **once**, while
//! it is being opened — the three arms and, since 2026-08-21, the cache ceiling
//! with them) or behind a process-wide `OnceLock` for the knobs that never vary
//! between two stores in one process. A `std::env::var` in a copy loop would be
//! one syscall per object served, and that exact defect was found and fixed in
//! gunnar the day before this was written. It is not a style rule here, it is
//! asserted: [`env_reads_here`] counts every read this module makes on the
//! calling thread, and
//! `the_selector_is_read_once_at_construction_and_never_per_operation` pushes
//! and serves a whole pack across an unchanged counter.
//!
//! Five reads at construction — three arms, the cache ceiling and the explode
//! policy — and zero per operation. [`env_reads`] is the same count
//! process-wide; see [`env_reads_here`] for why the guard asserts on the
//! thread-local one. The complete roster of keys is [`ALL_ENV`].
//!
//! # The names, so an operator can type them
//!
//! ```text
//!   ZNIPPY_GIT_WRITER = fast | safe | uring     (default: safe)
//!   ZNIPPY_GIT_INDEX  = one-table | four-tables | packed   (default: one-table)
//!   ZNIPPY_GIT_GC     = new-generation | in-place          (default: new-generation)
//!
//!   ZNIPPY_GIT_REDB_CACHE_BYTES = <bytes>                  (default: 67108864)
//!
//!   ZNIPPY_GIT_EXPLODE        = off | graph | full          (default: see exploded_arrow.rs)
//!   ZNIPPY_GIT_BOUNDARY_DELTA = 0 | off | false to disable  (default: on)
//!   ZNIPPY_GIT_REACH_COMMITS  = <commits>                   (default: 512)
//!   ZNIPPY_GIT_EMIT_WORKERS   = <workers>, 0 = all cores    (default: 4)
//! ```
//!
//! The first four are per store and are what [`StoreConfig`] holds and prints;
//! the last four are per process. [`ALL_ENV`] is all thirteen (eight + the five memory/batch knobs of 2026-08-30), and a consumer that
//! has to carry any of them across a boundary asserts its list against it.
//!
//! An unset variable takes the default. A variable set to something else is an
//! **error**, named and listing what is accepted — a typo that silently fell
//! back to the default would make an operator believe a measurement came from an
//! arm that never ran, which is worse than a failed open.
//!
//! ⚠ The fourth one is **not an arm**, and it is the only variable here that
//! [`GitStore::open`](crate::git_ops::GitStore::open) reads. It selects no
//! implementation and changes no contract — it is a memory ceiling, and redb's
//! own default for it is 1 GiB *per database*. See [`redb_cache_bytes`].

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

use anyhow::{Result, bail};

use crate::archive_write::{ArchiveWrite, FastWriter, SafeWriter};
use crate::gc::{CompactInPlace, Gc, NewGeneration};

/// `getenv` calls this module has made since the process started.
///
/// The counter behind the read-once law. Every path that reads the environment
/// in this crate goes through [`read_env`], which bumps it; nothing else does.
/// A caller — or a guard — can therefore prove that serving N objects cost zero
/// environment reads, which is the only way to state the law as an assertion
/// rather than as a comment.
pub fn env_reads() -> u64 {
    ENV_READS.load(Ordering::Relaxed)
}

/// The same count, **for the calling thread only**.
///
/// [`env_reads`] is process-wide, which makes it the right number to report and
/// the wrong number to *assert on*: any other thread opening a store moves it,
/// so a guard written against it is a race dressed as a test. It was only ever
/// accidentally deterministic — until 2026-08-10 the sole reader was
/// [`StoreConfig::from_env`], every caller of which took `ENV_LOCK`, so the
/// tests that counted were the only tests that counted. That stopped being true
/// when [`redb_cache_bytes`] joined it: **every** store open reads the
/// environment now, including [`GitStore::open`](crate::git_ops::GitStore::open),
/// and most of the suite opens stores without the lock.
///
/// So the law is asserted per thread, where a call path actually lives. It is
/// also the stronger statement: "this thread served 2687 objects and touched the
/// environment zero times" is what the law says, and it is now true regardless
/// of what else the process is doing.
pub fn env_reads_here() -> u64 {
    ENV_READS_HERE.with(|c| c.get())
}

static ENV_READS: AtomicU64 = AtomicU64::new(0);

thread_local! {
    static ENV_READS_HERE: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}

/// The one place this crate touches the environment.
///
/// `pub(crate)` so the per-process knobs in `delta.rs`, `exploded_arrow.rs`
/// and `git_ops.rs` read through the same counted door rather than through
/// their own `std::env::var` — every read this crate makes moves
/// [`env_reads`], or the read-once law is a comment.
pub(crate) fn read_env(key: &str) -> Option<String> {
    ENV_READS.fetch_add(1, Ordering::Relaxed);
    ENV_READS_HERE.with(|c| c.set(c.get() + 1));
    match std::env::var(key) {
        Ok(v) if !v.trim().is_empty() => Some(v.trim().to_string()),
        _ => None,
    }
}

// ── the writer arm ────────────────────────────────────────────────────────────

/// Which [`ArchiveWrite`] the push path appends through.
///
/// **The arm changes what `append` promises and nothing else.** All three write
/// the pushed pack's bytes verbatim, at the same offset, into the same file; a
/// store built on any of them holds byte-identical payload. What differs is how
/// much of the write is durable when `append` returns, and that is the whole
/// axis.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum WriterArm {
    /// [`FastWriter`] — **one `pwrite`, no fsync, no journal.**
    ///
    /// `append` returns as soon as the bytes are in the kernel's page cache.
    /// Stated plainly, because an operator selecting this is choosing it:
    ///
    /// * **A machine crash (power loss, panic, a hard reset) after `append`
    ///   returns loses the bytes.** Not "may lose" — nothing has told the device
    ///   about them.
    /// * There is **no journal**, so a store on this arm has no durable record
    ///   that a pack was ever acked. It therefore has **no crash recovery**: a
    ///   reopen cannot re-queue an interrupted pack because there is nothing to
    ///   diff the index against, and pack ordinals restart at 0
    ///   (`indexer::packs_already_acked`).
    /// * A *process* crash alone keeps the bytes: page cache survives `exit`.
    ///   That is the only crash it survives.
    ///
    /// It is here because it is the **ceiling the other two are measured
    /// against** — at 8 KiB it acks in 3.9 µs against `SafeWriter`'s 132 µs
    /// (`examples/push_path_bench.rs`, oden 2026-08-07) — and because there are
    /// real workloads whose durability contract is not git's: a rebuildable
    /// mirror, a bulk import that is re-run on failure, a benchmark. Selecting
    /// it is not refused and not warned about. It is the operator's call and
    /// the contract above is what they are choosing.
    Fast,
    /// [`SafeWriter`] — blob `fsync`, **then** the journal row, then the
    /// journal's `fsync`. **The default, and git's contract**: a push that was
    /// acked survives a crash. Four syscalls per append.
    #[default]
    Safe,
    /// [`UringWriter`](crate::uring_write::UringWriter) — the **same ordering**
    /// as [`Safe`](WriterArm::Safe), enforced by the kernel through an
    /// `IOSQE_IO_LINK` chain instead of by the caller blocking between four
    /// syscalls. One `io_uring_enter` carrying four linked ops.
    ///
    /// MEASURED 2026-08-07 (PLAN §8.3): **it did not win** — within noise of
    /// `SafeWriter` at every pack size, because the cost is the two device
    /// flushes and not the syscall count. Kept as a selectable arm precisely so
    /// that finding can be re-run rather than remembered. Linux only.
    ///
    /// **What this arm costs that the other two do not: pinned memory, and a
    /// ceiling on how many stores a process can hold.** Its ring registers a
    /// staging buffer, and `IORING_REGISTER_BUFFERS` pins pages against
    /// `RLIMIT_MEMLOCK` — a limit the kernel counts on the `user_struct`, so it
    /// is shared by every process the uid is running. One writer per store means
    /// the number of repositories this arm can serve is
    /// `RLIMIT_MEMLOCK / page`, and `Fast` and `Safe` have no such ceiling
    /// because they register nothing. It was **~107 stores** on a stock 8 MiB
    /// limit until 2026-08-14, when the registration was cut from 64 KiB to the
    /// one page a journal row actually needs; see
    /// [`uring_write`](crate::uring_write)'s module docs for the arithmetic, the
    /// measurement and the failure it produced in gunnar's sweep.
    ///
    /// It used to differ from `Safe` in one more way, and that is now closed:
    /// [`UringWriter::create`](crate::uring_write::UringWriter::create) opened
    /// its journal with `File::create`, so a **reopen truncated the journal**
    /// where `SafeWriter` appends to it — the durability of `Safe` within a
    /// process and the crash recovery of `Fast` across one. Both durable arms
    /// now open the same log the same way, which is what "one journal format,
    /// two transports" (LAW 5) always claimed of them.
    Uring,
}

impl WriterArm {
    /// Every arm, for a bench that sweeps them.
    pub const ALL: [WriterArm; 3] = [WriterArm::Fast, WriterArm::Safe, WriterArm::Uring];

    /// The spelling an operator types.
    pub fn as_str(self) -> &'static str {
        match self {
            WriterArm::Fast => "fast",
            WriterArm::Safe => "safe",
            WriterArm::Uring => "uring",
        }
    }

    /// One line on what this arm's `append` promises. Same string the trait's
    /// own [`ArchiveWrite::durability`] returns, so a bench row and a config
    /// dump cannot disagree.
    pub fn durability(self) -> &'static str {
        match self {
            WriterArm::Fast => {
                "none — page cache only; a machine crash after return loses the bytes, and there \
                 is no journal, so no crash recovery and no stable pack ordinals"
            }
            WriterArm::Safe | WriterArm::Uring => {
                "full — blob fsynced, then a journal row fsynced; crash after return keeps both"
            }
        }
    }

    /// Whether this arm keeps the durable extent log that
    /// [`GitStore::open_with_arms`](crate::git_ops::GitStore::open_with_arms)
    /// derives §13.12's `indexed` bit from.
    ///
    /// `None` is [`Fast`](WriterArm::Fast) and it is load-bearing rather than
    /// cosmetic: with no journal there is no durable record of an ack, so the
    /// crash-recovery diff has nothing to run against and a reopen re-queues
    /// nothing. Returning the path of a journal this arm does not write would
    /// make a reopened store diff against a **stale** log and re-absorb packs
    /// that a different arm acked.
    pub fn journal(self, blobs: &Path) -> Option<PathBuf> {
        match self {
            WriterArm::Fast => None,
            WriterArm::Safe | WriterArm::Uring => Some(crate::archive_write::journal_path(blobs)),
        }
    }

    /// Build the writer. `blobs` is the file the pushed packs are appended to.
    pub fn create(self, blobs: &Path) -> Result<Box<dyn ArchiveWrite>> {
        Ok(match self {
            WriterArm::Fast => Box::new(FastWriter::create(blobs)?),
            WriterArm::Safe => Box::new(SafeWriter::create(blobs)?),
            #[cfg(target_os = "linux")]
            WriterArm::Uring => Box::new(crate::uring_write::UringWriter::create(blobs)?),
            #[cfg(not(target_os = "linux"))]
            WriterArm::Uring => bail!(
                "the `uring` writer arm is Linux-only — this target has no io_uring, and this \
                 crate does not substitute a pwrite path under an io_uring name"
            ),
        })
    }

    /// Parse the spelling an operator types. Errors name what is accepted.
    pub fn parse(s: &str) -> Result<Self> {
        Ok(match s.trim().to_ascii_lowercase().as_str() {
            "fast" | "fastwriter" => WriterArm::Fast,
            "safe" | "safewriter" => WriterArm::Safe,
            "uring" | "uringwriter" | "io_uring" => WriterArm::Uring,
            other => bail!(
                "'{other}' is not a writer arm — expected one of fast, safe, uring \
                 ({}={other})",
                ENV_WRITER
            ),
        })
    }
}

// ── the index arm ─────────────────────────────────────────────────────────────

/// Which payload layout sits behind the read stack's `stree`.
///
/// **This one is a *type*, not a value**, and that is deliberate: it is the hot
/// path. `GitStore` is generic over it
/// (`GitStore<S = OneTableFourColumns>`), so a caller that knows its arm at
/// compile time — which includes every existing caller, through the default —
/// pays no dispatch at all. This enum exists so a *runtime* selector can pick
/// one; see [`crate::git_ops::open_from_env`], which monomorphises all three and
/// hands back a `Box<dyn GitOps>`.
///
/// MEASURED 2026-08-07 (PLAN §13): a packed 25-byte column beats four columns by
/// 19–76% on the payload gather but loses column scans by 3.3x–14.4x, and no git
/// operation does a full-row fetch. That is why `OneTableFourColumns` is the
/// default — and why the losing arms stay selectable, because that conclusion is
/// a measurement and measurements get re-run.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum IndexArm {
    /// [`OneTableFourColumns`](crate::index_layout::OneTableFourColumns) — the
    /// default. One Arrow IPC section, four columns.
    #[default]
    OneTableFourColumns,
    /// [`FourTables`](crate::index_layout::FourTables) — four independent IPC
    /// sections joined on the ordinal.
    FourTables,
    /// [`PackedPayload`](crate::index_layout::PackedPayload) — one 25-byte
    /// fixed-size-binary column holding the whole row.
    PackedPayload,
}

impl IndexArm {
    pub const ALL: [IndexArm; 3] = [
        IndexArm::OneTableFourColumns,
        IndexArm::FourTables,
        IndexArm::PackedPayload,
    ];

    /// The spelling an operator types.
    pub fn as_str(self) -> &'static str {
        match self {
            IndexArm::OneTableFourColumns => "one-table",
            IndexArm::FourTables => "four-tables",
            IndexArm::PackedPayload => "packed",
        }
    }

    /// The name [`ObjectIndex::name`](crate::index_layout::ObjectIndex::name)
    /// reports for the projection this arm builds. Distinct from
    /// [`as_str`](IndexArm::as_str) on purpose: one is what an operator types,
    /// the other is what the built object calls itself, and a guard that
    /// compares them is comparing the selector against applied output.
    pub fn projection_name(self) -> &'static str {
        match self {
            IndexArm::OneTableFourColumns => "OneTableFourColumns",
            IndexArm::FourTables => "FourTables",
            IndexArm::PackedPayload => "PackedPayload",
        }
    }

    pub fn parse(s: &str) -> Result<Self> {
        Ok(match s.trim().to_ascii_lowercase().as_str() {
            "one-table" | "one_table" | "onetablefourcolumns" | "one" => {
                IndexArm::OneTableFourColumns
            }
            "four-tables" | "four_tables" | "fourtables" | "four" => IndexArm::FourTables,
            "packed" | "packedpayload" => IndexArm::PackedPayload,
            other => bail!(
                "'{other}' is not an index arm — expected one of one-table, four-tables, packed \
                 ({ENV_INDEX}={other})"
            ),
        })
    }
}

// ── the gc arm ────────────────────────────────────────────────────────────────

/// Which [`Gc`] [`GitOps::gc`](crate::git_ops::GitOps::gc) runs as its third
/// step, after reachability and after the dead index rows are dropped.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum GcArm {
    /// [`NewGeneration`] — the default. Hard-link, compact the link, **verify
    /// the result by reading every entry back**, rename to `x.gN.znippy`,
    /// unlink the old generation last. A death at any point leaves at least one
    /// complete readable archive.
    #[default]
    NewGeneration,
    /// [`CompactInPlace`] — base znippy's `compact_archive` against the archive
    /// under its own name: stage beside it, rename over it. Verification is
    /// **post-hoc**: by the time there is something to read back, the original
    /// is already gone. One archive on disk instead of two, and no generation
    /// suffix — which is the reason to pick it — at the price of that window.
    CompactInPlace,
}

impl GcArm {
    pub const ALL: [GcArm; 2] = [GcArm::NewGeneration, GcArm::CompactInPlace];

    pub fn as_str(self) -> &'static str {
        match self {
            GcArm::NewGeneration => "new-generation",
            GcArm::CompactInPlace => "in-place",
        }
    }

    /// The name the built implementation reports through [`Gc::name`], which is
    /// also what lands in [`GcReport::strategy`](crate::gc::GcReport::strategy)
    /// — the applied output a guard reads.
    pub fn strategy(self) -> &'static str {
        match self {
            GcArm::NewGeneration => "NewGeneration",
            GcArm::CompactInPlace => "CompactInPlace",
        }
    }

    /// Build it. `Send + Sync` because a `GitStore` is shared across the
    /// indexer's threads.
    pub fn create(self) -> Box<dyn Gc + Send + Sync> {
        match self {
            GcArm::NewGeneration => Box::new(NewGeneration::new()),
            GcArm::CompactInPlace => Box::new(CompactInPlace::new()),
        }
    }

    pub fn parse(s: &str) -> Result<Self> {
        Ok(match s.trim().to_ascii_lowercase().as_str() {
            "new-generation" | "new_generation" | "newgeneration" | "new" | "generation" => {
                GcArm::NewGeneration
            }
            "in-place" | "in_place" | "inplace" | "compactinplace" | "compact" => {
                GcArm::CompactInPlace
            }
            other => bail!(
                "'{other}' is not a gc arm — expected one of new-generation, in-place \
                 ({ENV_GC}={other})"
            ),
        })
    }
}

// ── the config ────────────────────────────────────────────────────────────────

/// Environment variable naming the [`ArchiveWrite`] arm.
pub const ENV_WRITER: &str = "ZNIPPY_GIT_WRITER";
/// Environment variable naming the
/// [`ObjectIndex`](crate::index_layout::ObjectIndex) arm.
pub const ENV_INDEX: &str = "ZNIPPY_GIT_INDEX";
/// Environment variable naming the [`Gc`] arm.
pub const ENV_GC: &str = "ZNIPPY_GIT_GC";
/// Environment variable bounding **redb's page cache**, in bytes. See
/// [`redb_cache_bytes`].
pub const ENV_REDB_CACHE: &str = "ZNIPPY_GIT_REDB_CACHE_BYTES";
/// Environment variable switching the **boundary re-delta** off for an A/B
/// (`0` / `off` / `false`; anything else, or unset, is on). Read once per
/// process by [`crate::delta::enabled`].
pub const ENV_BOUNDARY_DELTA: &str = "ZNIPPY_GIT_BOUNDARY_DELTA";
/// Environment variable naming the **explode policy** of the
/// `objects.exploded` table (`off` · `graph` · `full`). Read once per store
/// open by `ExplodePolicy::from_env`.
pub const ENV_EXPLODE: &str = "ZNIPPY_GIT_EXPLODE";
/// Environment variable setting the **explode cache budget** — the byte cap on
/// the `objects.exploded` pools, `0`/`unlimited` for none. Read once per store
/// open by `crate::exploded_cache::budget_from_env`, which supplies the
/// filesystem-derived default when it is unset. The tier it bounds MEASURED
/// 141.7 GiB for an 8.12 GiB pack before the cap existed.
pub const ENV_EXPLODE_BUDGET: &str = "ZNIPPY_GIT_EXPLODE_BUDGET";
/// Environment variable capping the **live reachability walk** in commits. An
/// operator knob, not an arm — see `live_reach_policy` in `git_ops.rs`. Read
/// once per process.
pub const ENV_REACH_COMMITS: &str = "ZNIPPY_GIT_REACH_COMMITS";
/// Environment variable setting the **phase-1 emit workers per request**. An
/// operator knob, not an arm — see `emit_workers` in `git_ops.rs`. Read once
/// per process.
pub const ENV_EMIT_WORKERS: &str = "ZNIPPY_GIT_EMIT_WORKERS";
/// Environment variable setting the **memory watchdog's floor** as a percent
/// of `MemTotal` (default 20): below it the indexer flushes and waits. See
/// [`crate::memguard`]. Read once per process.
pub const ENV_MEM_BUDGET_PCT: &str = "ZNIPPY_GIT_MEM_BUDGET_PCT";
/// Environment variable setting the same floor in **absolute bytes**, which
/// wins over the percent when both are set. Read once per process.
pub const ENV_MEM_BUDGET_BYTES: &str = "ZNIPPY_GIT_MEM_BUDGET_BYTES";
/// Environment variable naming the file the watchdog reads instead of
/// `/proc/meminfo` — a test's fake. Read once per process.
pub const ENV_MEMINFO: &str = "ZNIPPY_GIT_MEMINFO";
/// Environment variable bounding the **held delta-base set** of the resolver in
/// bytes (default 25 % of `MemTotal`): over it the base whose next use is
/// farthest is dropped and re-inflated from the mmapped pack when needed. See
/// [`crate::resolve`]. Read once per process.
pub const ENV_BASE_BUDGET_BYTES: &str = "ZNIPPY_GIT_BASE_BUDGET_BYTES";
/// Environment variable setting how many **index rows are flushed per batch**
/// by the absorber (default 262 144) instead of the whole pack's rows at once.
/// Read once per process.
pub const ENV_INDEX_BATCH_ROWS: &str = "ZNIPPY_GIT_INDEX_BATCH_ROWS";
/// Environment variable setting the **absolute resume margin** in bytes
/// (default 64 MiB): a pause ends once `MemAvailable` is the floor plus this.
/// It used to be `budget / 8`, which on a 16 GB box demanded 410 MB of headroom
/// to leave a pause that a 75 MB shortfall opened — the livelock lane P2
/// measured. See [`crate::memguard`]. Read once per process.
pub const ENV_MEM_RESUME_MARGIN_BYTES: &str = "ZNIPPY_GIT_MEM_RESUME_MARGIN_BYTES";
/// Environment variable setting **how many subtrees the resolver walks at
/// once** (default: one per core). `1` is the serial resolver, and the stream
/// out is the same at every value. See [`crate::resolve::workers_from_env`].
/// Read once per process.
pub const ENV_RESOLVE_WORKERS: &str = "ZNIPPY_GIT_RESOLVE_WORKERS";
/// Environment variable bounding **how far the resolver's workers may run
/// ahead of its ordered output**, in bytes (default 128 MiB). Read once per
/// process.
pub const ENV_RESOLVE_LOOKAHEAD_BYTES: &str = "ZNIPPY_GIT_RESOLVE_LOOKAHEAD_BYTES";
/// Environment variable bounding **how long one pause may last**, in
/// milliseconds (default 300 000 = 5 minutes). Past it the worker stops
/// waiting and the pack is failed BY NAME rather than parked forever. See
/// [`crate::memguard::PauseEnd`]. Read once per process.
pub const ENV_MEM_PAUSE_MAX_MS: &str = "ZNIPPY_GIT_MEM_PAUSE_MAX_MS";

/// **Every environment variable this crate reads outside its tests**, in one
/// place, so a consumer that carries them across a process or container
/// boundary can assert its list against this one instead of against a count
/// it remembers.
///
/// # Why a list, and why here
///
/// gunnar's forge runs `gunnar serve` inside a podman container, and podman
/// passes nothing implicitly: a variable set on the host reaches the server
/// only if the forge names it in its crossing list. On 2026-08-11 five separate
/// wrong conclusions were drawn on one box from knobs that were "set" and never
/// arrived — the server ran on its default and the probe reported "no effect",
/// a null result shaped exactly like a real one. The forge's own list was three
/// entries long against a crate that read four keys, and **nothing went red
/// when this crate grew a key**, because the only list that looked was the
/// consumer's.
///
/// This is the producer's list. It is the one a consumer asserts against — "my
/// crossing list plus my named exclusions equals `ALL_ENV`" — so the day this
/// crate grows a ninth key, the consumer's guard is what says so, and the
/// failure names the key rather than producing a quiet default.
///
/// # It is asserted complete, not trusted
///
/// `every_environment_read_in_this_crate_is_named_in_all_env` (in this module's
/// tests) walks the crate's own source and checks that every `std::env::var`
/// call outside a `#[cfg(test)]` region names a key on this list — by literal
/// or by one of the `ENV_*` constants above. Grow a read without growing the
/// list and that test is what goes red.
///
/// Three of the thirteen are **arms** (they select an implementation and change
/// what the store promises); the other five are knobs and ceilings that select
/// nothing. [`StoreConfig`] holds the arms and the cache ceiling, and its
/// `Display` prints those four; the remaining four are per-process knobs that
/// never vary between two stores in one process.
pub const ALL_ENV: &[&str] = &[
    ENV_WRITER,
    ENV_INDEX,
    ENV_GC,
    ENV_REDB_CACHE,
    ENV_BOUNDARY_DELTA,
    ENV_EXPLODE,
    ENV_EXPLODE_BUDGET,
    ENV_REACH_COMMITS,
    ENV_EMIT_WORKERS,
    ENV_MEM_BUDGET_PCT,
    ENV_MEM_BUDGET_BYTES,
    ENV_MEMINFO,
    ENV_BASE_BUDGET_BYTES,
    ENV_INDEX_BATCH_ROWS,
    ENV_MEM_RESUME_MARGIN_BYTES,
    ENV_MEM_PAUSE_MAX_MS,
    ENV_RESOLVE_WORKERS,
    ENV_RESOLVE_LOOKAHEAD_BYTES,
];

/// **The page-cache ceiling for one repository's redb database**, in bytes.
///
/// # Why this exists at all
///
/// `redb::Database::create(path)` is `Builder::new().create(path)`, and
/// `Builder::new` ends with `set_cache_size(1024 * 1024 * 1024)` — redb-2.6.3
/// `db.rs:1140`. **One GiB, per database, by default**, split 90 % read /
/// 10 % write by `set_cache_size` (`db.rs:1184`), so the stock ceiling is 1 GiB
/// *per repository* and a long-lived server holding N repositories has N times
/// that.
///
/// # "Two databases per repository" was stale, and it cost a wrong diagnosis
///
/// This paragraph said "a store opens two of them, so the stock ceiling is
/// 2 GiB per repository" until 2026-08-31. **A store opens one.**
/// [`GitStore::open_with_arms`](crate::git_ops::GitStore::open_with_arms)
/// constructs exactly one [`ObjectReadStack`](crate::read_stack::ObjectReadStack)
/// — `objects.tail` — and `tests/fd_per_store.rs` counts exactly one
/// `objects.tail` descriptor per store and asserts the total is three. The
/// second database was the exploded table's, and it stopped being a redb when
/// §14 moved that tier to append-only Arrow IPC
/// ([`crate::exploded_arrow`]); the sentence outlived it.
///
/// It is recorded rather than quietly deleted because a stale doc is a second
/// truth and this one was believed: `gunnar/.nornir/stinky.md` S-079 doubled
/// this figure to 128 MiB per store on the strength of it, and built an
/// explanation of a live appliance's 10 G on top. See S-082 for the
/// measurement that refuted it.
///
/// It is a ceiling and not a reservation, so a small repository never noticed.
/// MEASURED 2026-08-10 (t14s, `h2h-linear-sha1-2048c-1024f-16k`, a 1 078 472 704-byte
/// `objects.exploded`): a single clone parked **594 MB resident** and kept it for
/// the life of the process, because the pages a full-file traversal touched all
/// fit under the ceiling and nothing evicted them.
///
/// # How the default was chosen
///
/// A B-tree read cache earns its keep on the **interior** nodes, which every
/// lookup re-touches, and earns nothing on a single sequential pass over the
/// leaves, which is what a `refold` is. The interior levels of a 4 KiB-page
/// B-tree over that 1.078 GB table are ~7 MB; 64 MiB holds all of them nine
/// times over and leaves 6.4 MiB of write cache, which is more than one absorb
/// batch dirties. Measured against the 1 GiB default on the clone path it costs
/// nothing and returns most of the resident set — see the sweep in
/// `agentAA-report.md`.
///
/// # This is the one environment read that is not an arm
///
/// Every other variable in this module selects an *implementation* and therefore
/// changes what the store promises, which is why
/// [`GitStore::open`](crate::git_ops::GitStore::open) deliberately reads none of
/// them. This one selects nothing: it is a memory ceiling, every arm behaves
/// identically under any value of it, and the bytes on disk are the same either
/// way. So it *is* read on the `open` path — an operator who has to cap a
/// server's footprint cannot be told to use a different constructor.
///
/// An unparseable or zero value is an **error**, for the same reason a
/// misspelled arm is: a silent fallback would let an operator believe a
/// measurement came from a ceiling that was never applied.
pub fn redb_cache_bytes() -> Result<usize> {
    let Some(raw) = read_env(ENV_REDB_CACHE) else {
        return Ok(DEFAULT_REDB_CACHE_BYTES);
    };
    match raw.parse::<usize>() {
        Ok(0) | Err(_) => bail!(
            "{ENV_REDB_CACHE}={raw:?} is not a positive byte count; it bounds redb's page \
             cache per database and the default is {DEFAULT_REDB_CACHE_BYTES}"
        ),
        Ok(n) => Ok(n),
    }
}

/// 64 MiB. See [`redb_cache_bytes`] for how that number was arrived at.
pub const DEFAULT_REDB_CACHE_BYTES: usize = 64 * 1024 * 1024;

/// One implementation of each trait, chosen.
///
/// A plain `Copy` value with no interior state: it is *read* once, at
/// construction, and from then on the store holds the built objects rather than
/// this. Handing it around after that is a description of what was built, not a
/// switch anything consults.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StoreConfig {
    pub writer: WriterArm,
    pub index: IndexArm,
    pub gc: GcArm,
    /// The redb page-cache ceiling the store's `objects.tail` database is opened
    /// under. One database per store — see [`redb_cache_bytes`].
    /// **Not an arm** — it selects nothing — but it is part of what was built,
    /// and a `Display` of the config that left it out would have hidden the
    /// one key that was "falsified" on 2026-08-11 by never reaching the
    /// container at all. [`StoreConfig::from_env`] fills it from
    /// [`redb_cache_bytes`]; [`DEFAULT`](Self::DEFAULT) is
    /// [`DEFAULT_REDB_CACHE_BYTES`].
    pub redb_cache_bytes: usize,
}

impl Default for StoreConfig {
    /// [`StoreConfig::DEFAULT`], spelled out. A derived `Default` would put a
    /// zero-byte page cache on the shipping combination.
    fn default() -> Self {
        Self::DEFAULT
    }
}

impl StoreConfig {
    /// **What shipped before this module existed, and what still ships.**
    ///
    /// `SafeWriter` + `ObjectReadStack<OneTableFourColumns>` + `NewGeneration`.
    /// [`GitStore::open`](crate::git_ops::GitStore::open) builds exactly this
    /// and consults no environment to do it.
    pub const DEFAULT: StoreConfig = StoreConfig {
        writer: WriterArm::Safe,
        index: IndexArm::OneTableFourColumns,
        gc: GcArm::NewGeneration,
        redb_cache_bytes: DEFAULT_REDB_CACHE_BYTES,
    };

    /// Read the three arms and the cache ceiling. **Four `getenv`s, once, here.**
    ///
    /// It is called from a store's constructor and from nowhere on a serving
    /// path; [`env_reads`] is what makes that assertable rather than asserted-
    /// by-comment. An unset or empty variable takes the default; a variable set
    /// to an unknown value is an error rather than a silent fallback, so a typo
    /// cannot make a measurement look like it came from an arm that never ran.
    ///
    /// The ceiling rides along because it is read on the same construction path
    /// and printed by the same `Display`: a caller that logs `from_env()`'s
    /// result once at store open has then logged **every** per-store variable
    /// this crate consults, and `ZNIPPY_GIT_REDB_CACHE_BYTES` is no longer the
    /// one that can be set on a host, stop at a container boundary, and leave
    /// no trace of having done so.
    pub fn from_env() -> Result<Self> {
        let mut cfg = StoreConfig::DEFAULT;
        if let Some(v) = read_env(ENV_WRITER) {
            cfg.writer = WriterArm::parse(&v)?;
        }
        if let Some(v) = read_env(ENV_INDEX) {
            cfg.index = IndexArm::parse(&v)?;
        }
        if let Some(v) = read_env(ENV_GC) {
            cfg.gc = GcArm::parse(&v)?;
        }
        cfg.redb_cache_bytes = redb_cache_bytes()?;
        Ok(cfg)
    }

    /// Same three arms with a different writer. For a bench that sweeps one axis.
    pub fn with_writer(mut self, w: WriterArm) -> Self {
        self.writer = w;
        self
    }

    pub fn with_index(mut self, i: IndexArm) -> Self {
        self.index = i;
        self
    }

    pub fn with_gc(mut self, g: GcArm) -> Self {
        self.gc = g;
        self
    }

    /// Same arms, a different redb page-cache ceiling. What
    /// [`GitStore::open`](crate::git_ops::GitStore::open) uses to put the
    /// environment's ceiling on the default arms without reading any arm.
    pub fn with_redb_cache_bytes(mut self, bytes: usize) -> Self {
        self.redb_cache_bytes = bytes;
        self
    }
}

/// `KEY=value` for every per-store variable, in the spelling an operator
/// would type — so one log line at store open is a complete, reproducible
/// statement of what the environment selected, including the ceiling.
impl std::fmt::Display for StoreConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "{}={} {}={} {}={} {}={}",
            ENV_WRITER,
            self.writer.as_str(),
            ENV_INDEX,
            self.index.as_str(),
            ENV_GC,
            self.gc.as_str(),
            ENV_REDB_CACHE,
            self.redb_cache_bytes
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::archive_write::read_journal;
    use crate::git_ops::{GitOps, GitStore, open_from_env, open_selected};
    use crate::index_layout::{OneTableFourColumns, PackedPayload};
    use crate::object::GitHashKind;
    use crate::store::tests::{corpus_pack, tmpdir};
    use std::sync::Mutex;

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

    /// The environment is process-global and this crate's tests run in parallel
    /// threads, so every guard that sets or counts an environment read takes
    /// this first. Without it [`env_reads`] would be a shared counter two tests
    /// moved at once, and the read-once guard below would be a coin toss.
    static ENV_LOCK: Mutex<()> = Mutex::new(());

    /// Run `f` with this module's variables set to `vars`, restoring whatever
    /// was there before.
    ///
    /// `ENV_REDB_CACHE` is cleared along with the three arms even though no test
    /// sets it: it is read on the same construction path, and a value inherited
    /// from the surrounding shell would otherwise change the footprint a test
    /// measures without appearing anywhere in the test.
    fn with_env<R>(vars: &[(&str, &str)], f: impl FnOnce() -> R) -> R {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        let saved: Vec<(String, Option<String>)> = [ENV_WRITER, ENV_INDEX, ENV_GC, ENV_REDB_CACHE]
            .iter()
            .map(|k| (k.to_string(), std::env::var(k).ok()))
            .collect();
        // SAFETY (edition 2024): the process-wide environment is mutated only
        // under ENV_LOCK, and no thread this crate spawns reads it.
        unsafe {
            for (k, _) in &saved {
                std::env::remove_var(k);
            }
            for (k, v) in vars {
                std::env::set_var(k, v);
            }
        }
        let out = f();
        unsafe {
            for (k, v) in &saved {
                match v {
                    Some(v) => std::env::set_var(k, v),
                    None => std::env::remove_var(k),
                }
            }
        }
        out
    }

    /// **The writer arm really is the writer that runs — asserted on the files
    /// on disk, not on a name.**
    ///
    /// The two arms differ in exactly one observable thing and it is a file: a
    /// durable arm writes an Arrow IPC journal row naming the extent before
    /// `append` returns, and [`WriterArm::Fast`] writes no journal at all. So
    /// the same pack is pushed into two stores that differ only in
    /// [`StoreConfig::writer`], and what is asserted is the **payload byte for
    /// byte in both** (the arm must not change what is stored) and the
    /// **journal present in one and absent in the other** (the arm must change
    /// what is promised).
    ///
    /// Seen RED by ignoring the selection in `GitStore::open_with_arms` —
    /// `arms.writer.create(&blobs)?` → `WriterArm::Safe.create(&blobs)?`, which
    /// is the hardcoded `SafeWriter::create(&blobs)?` this whole change removes:
    /// "the fast arm left a journal at
    /// /tmp/…-arm-writer-fast-…/objects.pack.journal — a writer that was not
    /// selected ran". Restored.
    ///
    /// `writer_name()` is asserted **after** the two directory listings, on
    /// purpose: with it first the same break failed on a `&'static str`
    /// (`left: "SafeWriter", right: "FastWriter"`), which proves a label was
    /// copied and not that a byte moved.
    ///
    /// **What it does NOT catch, found by trying.** Making `WriterArm::journal`
    /// return `Some` for `Fast` as well leaves this guard **green**: no
    /// `SafeWriter` runs, so no journal file is ever created, and
    /// `!fast_journal.exists()` still holds. That break is caught one level down
    /// by [`only_the_arms_that_write_a_journal_name_one`], which asserts on the
    /// path itself rather than on a file — recorded here rather than quietly
    /// claimed, because a guard's blind spots are the part worth writing down.
    #[test]
    fn each_writer_arm_leaves_its_own_durability_on_disk() {
        let (pack, rows) = corpus_pack();

        // ── the fast arm: bytes, and nothing that says they were acked ───────
        let fast_dir = tmpdir("arm-writer-fast");
        let fast = GitStore::<OneTableFourColumns>::open_with_arms(
            &fast_dir,
            "rickard",
            GitHashKind::Sha1,
            StoreConfig::DEFAULT.with_writer(WriterArm::Fast),
        )
        .unwrap();
        let fast_tx = fast.put_pack(&pack).unwrap();
        let (fo, fl) = fast_tx.extent.unwrap();
        let fast_blobs = fast_dir.join("objects.pack");
        let fast_journal = crate::archive_write::journal_path(&fast_blobs);
        assert_eq!(
            &std::fs::read(&fast_blobs).unwrap()[fo as usize..(fo + fl) as usize],
            &pack[..],
            "the fast arm did not store the pack verbatim — the arm may change the promise, \
             never the payload"
        );
        assert!(
            !fast_journal.exists(),
            "the fast arm left a journal at {} — a writer that was not selected ran",
            fast_journal.display()
        );

        // ── the safe arm: the same bytes, plus the durable row that claims them
        let safe_dir = tmpdir("arm-writer-safe");
        let safe = GitStore::<OneTableFourColumns>::open_with_arms(
            &safe_dir,
            "rickard",
            GitHashKind::Sha1,
            StoreConfig::DEFAULT.with_writer(WriterArm::Safe),
        )
        .unwrap();
        let safe_tx = safe.put_pack(&pack).unwrap();
        let (so, sl) = safe_tx.extent.unwrap();
        let safe_blobs = safe_dir.join("objects.pack");
        let safe_journal = crate::archive_write::journal_path(&safe_blobs);
        assert_eq!(
            &std::fs::read(&safe_blobs).unwrap()[so as usize..(so + sl) as usize],
            &pack[..],
            "the safe arm did not store the pack verbatim"
        );
        assert!(
            safe_journal.exists(),
            "the safe arm wrote no journal — the durable arm did not run"
        );
        assert_eq!(
            read_journal(&safe_journal).unwrap(),
            vec![(so, sl)],
            "the journal does not name the extent that was acked"
        );

        // Both arms store the same bytes at the same offset: the selection
        // moves the durability contract and nothing else.
        assert_eq!((fo, fl), (so, sl), "the two arms disagree about the extent");
        // The labels, asserted **after** the evidence — a name that agreed with
        // a directory listing that did not would be the wrong thing to fail on.
        assert_eq!(fast.writer_name(), "FastWriter");
        assert_eq!(safe.writer_name(), "SafeWriter");
        fast.wait_indexed();
        safe.wait_indexed();
        assert_eq!(fast.object_count(), rows.len());
        assert_eq!(safe.object_count(), rows.len());
        eprintln!(
            "load {}; {} objects: fast arm {} journal row(s), safe arm {} — {}",
            loadavg(),
            rows.len(),
            if fast_journal.exists() { 1 } else { 0 },
            read_journal(&safe_journal).unwrap().len(),
            WriterArm::Fast.durability(),
        );
    }

    /// **The third writer arm is selectable through a store too, and it lands
    /// the same durable journal.**
    ///
    /// The `fast`/`safe` guard above is the contrast; this one is the coverage.
    /// It asserts on applied output — the pack verbatim on disk and one journal
    /// row naming its extent — which is the same pair
    /// [`each_writer_arm_leaves_its_own_durability_on_disk`] asserts for `safe`,
    /// because the two durable arms write **one** journal format through two
    /// transports (LAW 5) and a store must not be able to tell them apart.
    ///
    /// A kernel that cannot run the chain is a **failure, not a skip**: the same
    /// policy `tests/archive_write.rs` already takes, because a silently skipped
    /// arm is an arm nobody notices stopped working.
    ///
    /// Seen RED by pointing the arm at the wrong writer —
    /// `WriterArm::Uring => Box::new(SafeWriter::create(blobs)?)` in
    /// `WriterArm::create`: "the uring arm ran a different writer — left:
    /// \"SafeWriter\", right: \"UringWriter\"". Restored.
    ///
    /// **That red lands on a name, and here that is the honest ceiling.** The
    /// two durable arms are *required* to be indistinguishable on disk — one
    /// journal format, one encoder, two transports — so there is no applied
    /// output that separates them, and inventing one would mean breaking LAW 5
    /// to make a guard feel better. The applied-output assertions above prove
    /// the chain did the job; the name is the only thing that can say which
    /// transport did it.
    #[cfg(target_os = "linux")]
    #[test]
    fn the_uring_arm_is_selectable_through_a_store_and_lands_the_same_journal() {
        let dir = tmpdir("arm-writer-uring");
        let store = GitStore::<OneTableFourColumns>::open_with_arms(
            &dir,
            "rickard",
            GitHashKind::Sha1,
            StoreConfig::DEFAULT.with_writer(WriterArm::Uring),
        )
        .expect(
            "the io_uring arm could not be built on this kernel — not skipped, this is a real \
             failure",
        );
        let (pack, oid) = crate::store::tests::one_blob_pack(b"a push down the io_uring chain");
        let tx = store.put_pack(&pack).unwrap();
        let (o, l) = tx.extent.unwrap();

        let blobs = dir.join("objects.pack");
        assert_eq!(
            &std::fs::read(&blobs).unwrap()[o as usize..(o + l) as usize],
            &pack[..],
            "the uring arm did not store the pack verbatim"
        );
        assert_eq!(
            read_journal(&crate::archive_write::journal_path(&blobs)).unwrap(),
            vec![(o, l)],
            "the uring arm's journal does not name the extent it acked — the two durable arms \
             must write one journal format"
        );
        assert_eq!(store.writer_name(), "UringWriter", "the uring arm ran a different writer");
        store.wait_indexed();
        assert!(store.has(&oid).unwrap(), "the pushed object never reached the index");
    }

    /// **The gc arm really is the gc that runs — asserted on which files exist
    /// afterwards.**
    ///
    /// [`GcArm::NewGeneration`] renames the compacted archive to `x.g1.znippy`
    /// and unlinks the original **last**; [`GcArm::CompactInPlace`] renames over
    /// the original name and produces no generation. Those are two different
    /// sets of files on disk for the same input, which is what is asserted —
    /// `GcReport::strategy` is checked too, but it is a label and the
    /// directory listing is the evidence.
    ///
    /// Seen RED by ignoring the selection in `GitStore::open_with_arms` —
    /// `gc: arms.gc.create()` → `gc: GcArm::NewGeneration.create()`: "the
    /// in-place arm produced a new generation at
    /// /tmp/…-arm-gc-in-place-…/repository.g1.znippy — a gc that was not
    /// selected ran". Restored.
    ///
    /// `report.strategy` is asserted **after** the directory listing, on
    /// purpose: with the strategy check first the same break failed on a
    /// `&'static str` (`left: "NewGeneration", right: "CompactInPlace"`), which
    /// is a label agreeing with itself and says nothing about what happened to
    /// the archive.
    #[test]
    fn each_gc_arm_leaves_its_own_generation_on_disk() {
        for arm in GcArm::ALL {
            let dir = tmpdir(&format!("arm-gc-{}", arm.as_str()));
            let store = GitStore::<OneTableFourColumns>::open_with_arms(
                &dir,
                "rickard",
                GitHashKind::Sha1,
                StoreConfig::DEFAULT.with_gc(arm),
            )
            .unwrap();
            let (pack, _) = corpus_pack();
            store.put(&pack, &[]).unwrap();
            store.absorb_pending().unwrap();

            let root = store
                .graph_snapshot()
                .into_iter()
                .find(|c| c.generation == 1)
                .expect("a root commit");
            let root_raw = hex::decode(&root.oid).unwrap();
            store
                .update_ref("refs/heads/root", None, Some(&root_raw))
                .unwrap();

            // A real znippy archive for the compaction step to work on — the
            // same fixture `store::tests`' own gc guard builds.
            let files = vec![
                ("pack-0.pack".to_string(), pack.clone()),
                ("pack-1.pack".to_string(), pack.clone()),
            ];
            znippy_common::create_archive(store.archive_path(), &files, 3).unwrap();
            let original = store.archive_path().to_path_buf();
            let generation = crate::gc::next_generation(&original).unwrap();

            let report = store.gc().unwrap();
            match arm {
                GcArm::NewGeneration => {
                    assert_eq!(report.archive, generation);
                    assert!(
                        generation.exists(),
                        "the new-generation arm produced no {}",
                        generation.display()
                    );
                    assert!(
                        !original.exists(),
                        "the new-generation arm kept the old generation at {}",
                        original.display()
                    );
                    assert_eq!(report.retired.as_deref(), Some(original.as_path()));
                    assert!(report.verified, "the new generation was not read back");
                }
                GcArm::CompactInPlace => {
                    assert!(
                        !generation.exists(),
                        "the in-place arm produced a new generation at {} — a gc that was not \
                         selected ran",
                        generation.display()
                    );
                    assert!(
                        original.exists(),
                        "the in-place arm removed the archive it compacts into"
                    );
                    assert_eq!(report.archive, original);
                    assert_eq!(report.retired, None);
                }
            }
            // The label, after the evidence.
            assert_eq!(report.strategy, arm.strategy());
            assert!(
                report.bytes_after <= report.bytes_before,
                "{} grew the archive: {} → {}",
                arm.strategy(),
                report.bytes_before,
                report.bytes_after
            );
            eprintln!(
                "load {}; gc arm {}: {} → {} bytes, archive now {}",
                loadavg(),
                arm.strategy(),
                report.bytes_before,
                report.bytes_after,
                report.archive.display()
            );
        }
    }

    /// **The index arm really is the layout that gets built — asserted on the
    /// Arrow bytes it materialised, and on all three agreeing about every
    /// object.**
    ///
    /// A type name would prove nothing here (all three stacks report
    /// `ObjectReadStack`, because that is what they are). What distinguishes the
    /// arms is the projection they actually built:
    /// [`ObjectIndex::ipc_bytes`] is the size of the Arrow IPC payload held
    /// resident, and one packed 25-byte column, four columns in one section and
    /// four independent sections are three different numbers for the same 1352
    /// objects. Pairwise-distinct is the assertion.
    ///
    /// The other half matters more: the arms are a **layout** choice, so all
    /// three must answer identically. Every oid's full row is compared across
    /// the three, so an arm that decoded its own payload wrongly fails here
    /// rather than in production.
    ///
    /// Driven through [`open_selected`] — the runtime door — so the guard fails
    /// when the *selector* stops selecting rather than only when a layout is
    /// broken.
    ///
    /// Seen RED by pinning the type in `open_selected`, every match arm building
    /// `SelectedStore::OneTableFourColumns(GitStore::<OneTableFourColumns>…)`,
    /// which is the state this change starts from: "one-table and four-tables
    /// materialised the same 145544 IPC bytes — the index arm was not
    /// selected". Restored.
    ///
    /// **`store.arms().index` stayed green under that break**, and that is the
    /// reason it is not the assertion: `arms` is the value that was *requested*,
    /// so it agrees with the caller no matter which type got built. The byte
    /// count is what the store actually did.
    #[test]
    fn each_index_arm_builds_its_own_projection_and_all_three_agree() {
        let (pack, rows) = corpus_pack();

        let mut built = Vec::new();
        for arm in IndexArm::ALL {
            let store = open_selected(
                &tmpdir(&format!("arm-index-{}", arm.as_str())),
                "rickard",
                GitHashKind::Sha1,
                StoreConfig::DEFAULT.with_index(arm),
            )
            .unwrap();
            store.put_pack(&pack).unwrap();
            store.wait_indexed();
            // A rebuild, so the projection under test — not the redb tail — is
            // what answers below.
            store.rebuild_projection().unwrap();
            assert_eq!(store.arms().index, arm);
            // The name the BUILT projection reports, against the name this arm
            // says it builds. `arms().index` above is the request echoed back
            // and stays green when the selector selects nothing;
            // `index_name()` goes through `ObjectReadStack::projection_name`
            // to `S::name()`, so it can only answer what was monomorphised.
            //
            // It is what a server logs to say which layout it is running —
            // gunnar's `store.znippy_arms_selected` — and the reason
            // `ObjectIndex::name` on the stack itself is no use for that: the
            // stack answers "ObjectReadStack" for all three, because that is
            // what the stack is.
            assert_eq!(
                store.index_name(),
                arm.projection_name(),
                "the {} arm built a projection calling itself {}",
                arm.as_str(),
                store.index_name()
            );
            built.push((arm, store));
        }

        // Applied output: three layouts, three different quantities of Arrow IPC
        // materialised for the same objects.
        for (i, (a, sa)) in built.iter().enumerate() {
            for (b, sb) in &built[i + 1..] {
                assert_ne!(
                    sa.index_ipc_bytes(),
                    sb.index_ipc_bytes(),
                    "{} and {} materialised the same {} IPC bytes — the index arm was not selected",
                    a.as_str(),
                    b.as_str(),
                    sa.index_ipc_bytes()
                );
            }
        }

        // …and they are three spellings of one answer. `get` goes through the
        // whole stack — index row, extent, bytes off disk — so a layout that
        // decoded its own payload wrongly cannot agree here by accident.
        for r in rows.iter().take(512) {
            let first = built[0].1.get(&r.oid).unwrap();
            assert!(first.is_some(), "{} is missing from one-table", hex::encode(&r.oid));
            for (arm, store) in &built[1..] {
                assert_eq!(
                    store.get(&r.oid).unwrap(),
                    first,
                    "{} disagrees with one-table about {}",
                    arm.as_str(),
                    hex::encode(&r.oid)
                );
            }
        }
        eprintln!(
            "load {}; {} objects: {}",
            loadavg(),
            rows.len(),
            built
                .iter()
                .map(|(a, s)| format!("{} {} B", a.as_str(), s.index_ipc_bytes()))
                .collect::<Vec<_>>()
                .join(", ")
        );
    }

    /// **The default did not move, and no environment variable can move it.**
    ///
    /// The compatibility half of this change. `GitStore::open` built
    /// `SafeWriter` + `ObjectReadStack<OneTableFourColumns>` + `NewGeneration`
    /// before the selector existed and must build exactly that after it —
    /// including in a process whose environment names all three *other* arms,
    /// because a caller that never asked for a selector must not have its
    /// durability contract changed by somebody's shell.
    ///
    /// Asserted on applied output rather than on `arms()`: the journal row that
    /// only a durable writer produces, and the generation file that only
    /// `NewGeneration` produces.
    ///
    /// Seen RED by making the old constructor read the environment —
    /// `Self::open_with_arms(root, account, hash, StoreConfig::DEFAULT)` →
    /// `Self::open_with_arms(root, account, hash, StoreConfig::from_env()?)` in
    /// `GitStore::open_with`: "GitStore::open wrote no journal — an environment
    /// variable moved the default durability contract". Restored.
    ///
    /// `arms()` is asserted **after** the journal, for the third time in this
    /// module and for the same reason: with it first the break failed on
    /// `StoreConfig { writer: Fast, … }` vs `StoreConfig { writer: Safe, … }`,
    /// which is the request echoed back rather than the contract that was kept.
    #[test]
    fn the_default_is_unchanged_and_the_environment_cannot_move_it() {
        with_env(
            &[
                (ENV_WRITER, "fast"),
                (ENV_INDEX, "packed"),
                (ENV_GC, "in-place"),
            ],
            || {
                let dir = tmpdir("arm-default");
                let store = GitStore::open(&dir, "rickard").unwrap();
                let (pack, rows) = corpus_pack();
                let tx = store.put(&pack, &[]).unwrap();
                let journal = crate::archive_write::journal_path(&dir.join("objects.pack"));
                assert!(
                    journal.exists(),
                    "GitStore::open wrote no journal — an environment variable moved the default \
                     durability contract"
                );
                assert_eq!(read_journal(&journal).unwrap(), vec![tx.extent.unwrap()]);
                store.wait_indexed();
                assert_eq!(store.object_count(), rows.len());
                // The labels, after the evidence.
                assert_eq!(
                    store.arms(),
                    StoreConfig::DEFAULT,
                    "GitStore::open did not build the shipping arms"
                );
                assert_eq!(store.writer_name(), "SafeWriter");

                // And the gc arm is still the one that keeps the old generation
                // until the new one is proven.
                let root = store
                    .graph_snapshot()
                    .into_iter()
                    .find(|c| c.generation == 1)
                    .expect("a root commit");
                let root_raw = hex::decode(&root.oid).unwrap();
                store
                    .update_ref("refs/heads/root", None, Some(&root_raw))
                    .unwrap();
                let files = vec![("pack-0.pack".to_string(), pack.clone())];
                znippy_common::create_archive(store.archive_path(), &files, 3).unwrap();
                let generation = crate::gc::next_generation(store.archive_path()).unwrap();
                let report = store.gc().unwrap();
                assert_eq!(report.strategy, "NewGeneration");
                assert!(
                    generation.exists(),
                    "GitStore::open's gc arm is not NewGeneration — the environment moved it"
                );
            },
        );
    }

    /// **The selector is read once, at construction, and never per operation.**
    ///
    /// gunnar's own rule, stated in its source: *read once, here, and not per
    /// entry — a getenv inside the copy loop would be one syscall per object
    /// served*. A per-object `getenv` was a real defect found and fixed there,
    /// so this is asserted with a counter rather than trusted to review.
    ///
    /// [`env_reads`] counts every environment read this crate makes. The whole
    /// corpus pack is then pushed and served — one `put`, 1352 `has`, a batch
    /// `extents`, `get`, `size`, `refs` — across a **counter that does not
    /// move**.
    ///
    /// **Five reads at construction since 2026-08-21** (four since 2026-08-10,
    /// three before). Three are [`StoreConfig::from_env`]'s arms; the fourth is
    /// [`redb_cache_bytes`], which `from_env` now reads alongside them so the
    /// ceiling is part of the config a caller logs; the fifth is the explode
    /// policy `ExplodedArchive::open` reads through the same counted door.
    /// None of the extra two is an arm — they select no implementation — but
    /// each is an environment read, so each is counted, and each is read
    /// exactly once per store rather than once per database or once per lookup.
    ///
    /// Seen RED by re-reading the selector on the lookup path — adding
    /// `let _ = crate::arms::StoreConfig::from_env()?;` at the top of
    /// `GitStore::lookup_one`: "the selector was read **8070** times while
    /// serving 2692 operations — it must be read once, at construction; left:
    /// 8070, right: 3". Restored.
    ///
    /// Seen RED a second time, and this one is what keeps the guard from being
    /// vacuous: `StoreConfig::from_env` stubbed to `return
    /// Ok(StoreConfig::DEFAULT)` before it reads anything gives "opening a store
    /// read the environment **0** times, not once per variable". A selector that
    /// is never read at all also never moves the counter during serving, so
    /// without the construction-time line the guard would pass over a selector
    /// that does nothing. Restored.
    #[test]
    fn the_selector_is_read_once_at_construction_and_never_per_operation() {
        with_env(
            &[
                (ENV_WRITER, "safe"),
                (ENV_INDEX, "four-tables"),
                (ENV_GC, "in-place"),
            ],
            || {
                let dir = tmpdir("arm-read-once");
                let (pack, rows) = corpus_pack();

                // `env_reads_here`, not `env_reads`: the process-wide counter
                // is moved by every other test thread that opens a store, and
                // since `redb_cache_bytes` joined the readers that is most of
                // them. Per thread it is exact.
                let before = env_reads_here();
                let store = open_from_env(&dir, "rickard", GitHashKind::Sha1).unwrap();
                let at_open = env_reads_here();
                // Three arms, the redb cache ceiling, the explode policy and
                // the explode budget. Once each — a store opens TWO redb
                // databases and still reads the ceiling once.
                assert_eq!(
                    at_open - before,
                    6,
                    "opening a store read the environment {} times, not once per variable",
                    at_open - before
                );

                // Everything below this line is serving. Nothing here may read
                // the environment even once.
                let mut ops = 0u64;
                store.put(&pack, &[]).unwrap();
                ops += 1;
                let oids: Vec<&[u8]> = rows.iter().map(|r| r.oid.as_slice()).collect();
                for oid in &oids {
                    assert!(store.has(oid).unwrap());
                    ops += 1;
                }
                let ext = store.extents(&oids).unwrap();
                ops += 1;
                assert_eq!(ext.len(), oids.len());
                assert!(ext.iter().all(Option::is_some));
                assert!(store.get(oids[0]).unwrap().is_some());
                assert!(store.size(oids[0]).unwrap().is_some());
                store.refs().unwrap();
                ops += 3;

                let after = env_reads_here();
                assert_eq!(
                    after, at_open,
                    "the selector was read {} times while serving {ops} operations — it must be \
                     read once, at construction",
                    after - before
                );
                eprintln!(
                    "load {}; {ops} operations over {} objects: {} environment read(s), all of \
                     them at construction",
                    loadavg(),
                    rows.len(),
                    at_open - before,
                );
            },
        );
    }

    /// **A bad value is an error, not a silent default** — through the real
    /// front door, so the refusal cannot be a parser test that nothing calls.
    #[test]
    fn an_unknown_arm_refuses_to_open_rather_than_falling_back() {
        with_env(&[(ENV_WRITER, "safest")], || {
            let dir = tmpdir("arm-typo");
            let e = match open_from_env(&dir, "rickard", GitHashKind::Sha1) {
                Ok(_) => panic!("a typo opened a store on the default arm"),
                Err(e) => e,
            };
            let msg = format!("{e:#}");
            assert!(
                msg.contains("'safest' is not a writer arm") && msg.contains(ENV_WRITER),
                "the refusal must name the bad value and the variable it came from: {msg}"
            );
        });
        // …and an empty environment is the shipping default, not an error.
        with_env(&[], || {
            assert_eq!(StoreConfig::from_env().unwrap(), StoreConfig::DEFAULT);
        });
    }

    /// **The runtime door and the typed door build the same store.**
    ///
    /// [`open_selected`] exists so a value can choose the index arm, which is a
    /// type; the risk it carries is that the dispatched path drifts from the
    /// monomorphised one. Asserted on applied output: the same pack pushed
    /// through both answers the same for every oid.
    #[test]
    fn the_selected_store_and_the_typed_store_answer_alike() {
        let (pack, rows) = corpus_pack();
        let arms = StoreConfig::DEFAULT.with_index(IndexArm::PackedPayload);

        let boxed = open_selected(
            &tmpdir("arm-selected"),
            "rickard",
            GitHashKind::Sha1,
            arms,
        )
        .unwrap();
        let typed = GitStore::<PackedPayload>::open_with_arms(
            &tmpdir("arm-typed"),
            "rickard",
            GitHashKind::Sha1,
            arms,
        )
        .unwrap();

        boxed.put(&pack, &[]).unwrap();
        typed.put(&pack, &[]).unwrap();
        for r in rows.iter().take(512) {
            let a = boxed.get(&r.oid).unwrap();
            let b = typed.get(&r.oid).unwrap();
            assert_eq!(a, b, "the boxed and typed stores disagree about {}", hex::encode(&r.oid));
            assert!(a.is_some());
        }
    }

    /// Round-trips through every spelling, and a typo is an error rather than a
    /// default.
    #[test]
    fn every_arm_parses_from_the_name_an_operator_types() {
        for a in WriterArm::ALL {
            assert_eq!(WriterArm::parse(a.as_str()).unwrap(), a);
            assert_eq!(WriterArm::parse(&a.as_str().to_uppercase()).unwrap(), a);
        }
        for a in IndexArm::ALL {
            assert_eq!(IndexArm::parse(a.as_str()).unwrap(), a);
        }
        for a in GcArm::ALL {
            assert_eq!(GcArm::parse(a.as_str()).unwrap(), a);
        }
        // A typo must not become the default: an operator who mistyped `safe`
        // and got a measurement labelled `safe` would be reading a lie.
        for bad in ["saef", "", "fastwriter2", "none"] {
            assert!(
                WriterArm::parse(bad).is_err(),
                "'{bad}' parsed as a writer arm"
            );
        }
        assert!(IndexArm::parse("one-tabel").is_err());
        assert!(GcArm::parse("newgen").is_err());
    }

    /// **[`ALL_ENV`] is complete, checked against the crate's own source.**
    ///
    /// Walks every `.rs` under this crate's `src/`, cuts each file at its first
    /// `#[cfg(test)]` (the convention here: one tests module, at the bottom),
    /// and finds every environment read in what is left — `std::env::var(…)`,
    /// `env::var_os(…)` and [`read_env`]`(…)`. The key each one names, whether
    /// a string literal or one of the `ENV_*` constants declared in this file,
    /// must be on [`ALL_ENV`]. The only `std::env::var` allowed outside that
    /// rule is the one inside [`read_env`] itself, which takes its key as a
    /// parameter.
    ///
    /// This is what goes red the day the crate grows a key without naming it —
    /// the defect gunnar's forge could not see from its side (it asserted its
    /// crossing list against a count it remembered, and the count was wrong
    /// for four months of one day). A source scan is the honest shape for it:
    /// the keys are string literals and the reads are call sites, and no
    /// runtime counter can tell which *names* were read.
    ///
    /// Seen RED by adding `std::env::var("ZNIPPY_GIT_NEW_KNOB")` to
    /// `git_ops.rs` outside its tests: "git_ops.rs reads ZNIPPY_GIT_NEW_KNOB
    /// and ALL_ENV does not name it". Restored.
    #[test]
    fn every_environment_read_in_this_crate_is_named_in_all_env() {
        let src = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
        let this_file = std::fs::read_to_string(src.join("arms.rs")).unwrap();

        // `pub const ENV_X: &str = "…";` → ENV_X ↦ "…", from this file only.
        let mut consts: std::collections::BTreeMap<String, String> = Default::default();
        for line in this_file.lines() {
            let t = line.trim();
            let Some(rest) = t.strip_prefix("pub const ENV_") else { continue };
            let Some((name, val)) = rest.split_once(": &str = \"") else { continue };
            let val = val.split('"').next().unwrap();
            consts.insert(format!("ENV_{name}"), val.to_string());
        }
        assert!(consts.len() >= 8, "fewer ENV_* constants than expected: {consts:?}");

        let mut seen: Vec<(String, String)> = Vec::new();
        let mut files = std::fs::read_dir(&src)
            .unwrap()
            .map(|e| e.unwrap().path())
            .filter(|p| p.extension().is_some_and(|e| e == "rs"))
            .collect::<Vec<_>>();
        files.sort();
        assert!(!files.is_empty());
        for path in files {
            let file = path.file_name().unwrap().to_string_lossy().into_owned();
            let text = std::fs::read_to_string(&path).unwrap();
            let non_test = match text.find("\n#[cfg(test)]") {
                Some(at) => &text[..at],
                None => &text[..],
            };
            for needle in ["env::var(", "env::var_os(", "read_env("] {
                let mut from = 0;
                while let Some(at) = non_test[from..].find(needle) {
                    let call_at = from + at;
                    from = call_at + needle.len();
                    // The definition of `read_env` itself, and doc/comment lines
                    // that merely mention a call, are not reads.
                    let line_start = non_test[..call_at].rfind('\n').map_or(0, |i| i + 1);
                    let line = non_test[line_start..].lines().next().unwrap_or("").trim_start();
                    if line.starts_with("//") || line.starts_with("pub(crate) fn read_env") {
                        continue;
                    }
                    // The `env::var(key)` inside `read_env` — the one door.
                    if file == "arms.rs" && needle == "env::var(" && line.contains("var(key)") {
                        continue;
                    }
                    let arg = non_test[from..].trim_start();
                    let key = if let Some(lit) = arg.strip_prefix('"') {
                        lit.split('"').next().unwrap().to_string()
                    } else {
                        let ident: String = arg
                            .chars()
                            .take_while(|c| c.is_ascii_alphanumeric() || *c == '_' || *c == ':')
                            .collect();
                        let short = ident.rsplit("::").next().unwrap_or(&ident).to_string();
                        consts.get(&short).cloned().unwrap_or_else(|| {
                            panic!(
                                "{file} reads the environment through `{needle}{ident}…`, which \
                                 is neither a string literal nor an ENV_* constant this test \
                                 can resolve — name the key as a `pub const ENV_*` in arms.rs"
                            )
                        })
                    };
                    seen.push((file.clone(), key));
                }
            }
        }
        assert!(
            seen.len() >= 8,
            "the scan found only {} environment reads; it is supposed to find at least the \
             eight in ALL_ENV ({seen:?})",
            seen.len()
        );
        for (file, key) in &seen {
            assert!(
                ALL_ENV.contains(&key.as_str()),
                "{file} reads {key} and ALL_ENV does not name it"
            );
        }
        // And the other direction: a key on the list that nothing reads is a
        // stale entry a consumer would be carrying for nothing.
        for key in ALL_ENV {
            assert!(
                seen.iter().any(|(_, k)| k == key),
                "ALL_ENV names {key} but no non-test code in this crate reads it"
            );
        }
    }

    /// The default is the shipping combination, spelled out rather than derived
    /// — a derived `Default` that drifted would move the durability contract.
    #[test]
    fn the_default_config_is_the_shipping_combination() {
        assert_eq!(StoreConfig::default(), StoreConfig::DEFAULT);
        assert_eq!(StoreConfig::DEFAULT.writer, WriterArm::Safe);
        assert_eq!(StoreConfig::DEFAULT.index, IndexArm::OneTableFourColumns);
        assert_eq!(StoreConfig::DEFAULT.gc, GcArm::NewGeneration);
        assert_eq!(
            StoreConfig::DEFAULT.to_string(),
            "ZNIPPY_GIT_WRITER=safe ZNIPPY_GIT_INDEX=one-table ZNIPPY_GIT_GC=new-generation \
             ZNIPPY_GIT_REDB_CACHE_BYTES=67108864"
        );
    }

    /// **Only the durable arms name a journal**, and the fast one names none.
    ///
    /// Not cosmetic: the path returned here is what a reopen derives §13.12's
    /// `indexed` bit from, so an arm that writes no journal must not hand back
    /// the name of one — a store that ran on `SafeWriter` and is reopened on
    /// `FastWriter` would then diff against a log this writer is not appending
    /// to, and resume pack ordinals from it.
    ///
    /// Seen RED by `WriterArm::Fast => Some(journal_path(blobs))`: "left:
    /// Some(\"/nonexistent/objects.pack.journal\") right: None". Restored.
    /// This is the guard that catches that break — the disk-level one above
    /// stays green for it, which is why both exist.
    #[test]
    fn only_the_arms_that_write_a_journal_name_one() {
        let blobs = Path::new("/nonexistent/objects.pack");
        assert_eq!(WriterArm::Fast.journal(blobs), None);
        assert_eq!(
            WriterArm::Safe.journal(blobs),
            Some(PathBuf::from("/nonexistent/objects.pack.journal"))
        );
        assert_eq!(
            WriterArm::Uring.journal(blobs),
            WriterArm::Safe.journal(blobs),
            "the two durable arms share one journal format (LAW 5) and must share its name"
        );
    }

    /// The arm's stated durability is the writer's own, not a second copy of it
    /// that could drift.
    #[test]
    fn the_arms_durability_line_matches_the_writer_it_builds() {
        let dir = crate::store::tests::tmpdir("arm-durability");
        for arm in [WriterArm::Fast, WriterArm::Safe] {
            let w = arm.create(&dir.join(format!("{}.pack", arm.as_str()))).unwrap();
            match arm {
                WriterArm::Fast => {
                    assert_eq!(w.name(), "FastWriter");
                    assert!(
                        w.durability().starts_with("none"),
                        "FastWriter must say plainly that it promises nothing: {}",
                        w.durability()
                    );
                    assert!(arm.durability().starts_with("none"));
                }
                WriterArm::Safe => {
                    assert_eq!(w.name(), "SafeWriter");
                    assert_eq!(w.durability(), arm.durability());
                }
                WriterArm::Uring => unreachable!(),
            }
        }
    }
}