gunnar_server/

storage.rs

//! **Which storage engine this binary carries, and which one it is running.**
//!
//! There is one storage contract in the constellation and it is
//! [`git_storage_trait::GitOps`] — *the eleven* — which lives in znippy beside
//! its main implementation. This module is the *only* place in gunnar that names
//! an implementation of it.
//!
//! ```text
//!                   git-storage-trait :: GitOps
//!                    ┌───────────┴───────────┐
//!        znippy-plugin-git            storage-git-gix
//!        (Arrow-IPC archive)          (gix-odb / gix-ref)
//!            `engine-znippy`              `engine-gix`
//!                    └───────────┬───────────┘
//!                            Engine::open_repo
//!                        ──▶ Arc<OpenRepo>
//! ```
//!
//! # What used to be here, and why it is gone
//!
//! A second, older seam: `trait StorageEngine` with `install()`, `installed()`
//! and a `static ENGINE: OnceLock<&'static dyn StorageEngine>`. It existed
//! because the engine was chosen by *linking a different crate* — `gunnar-znippy`
//! — which had to run `install()` before `run()`, and therefore needed a `main`
//! of its own. That crate declared `[[bin]] gunnar` and `[[bin]] gunnar-server`,
//! two names the build declaration had already assigned to `gunnar-cli` and to
//! this package, while being absent from that declaration itself: an undeclared
//! second writer of two declared artifact names. It is why every matrix-built
//! binary answered *"this binary carries no storage engine"*.
//!
//! The engine is a **feature** now. There is nothing left to install, so there is
//! no `OnceLock`, no `install`, and no engine-free build — see the
//! [`compile_error!`] below. `gunnar-znippy` is deleted.
//!
//! # Both engines may be compiled in at once
//!
//! Cargo features are additive and must be, so this is the correct idiom rather
//! than a compromise:
//!
//! | build | engines linked | selection |
//! |---|---|---|
//! | `--features engine-znippy,engine-gix` | both | **runtime**, `GUNNAR_ENGINE`, defaulting to znippy |
//! | default | znippy only | nothing to select |
//! | `--no-default-features --features server,engine-gix` | gix only | nothing to select |
//!
//! **The rule.** When both are compiled the default is **znippy**, and asking for
//! an engine that was not compiled in fails loudly, at startup, before anything
//! is served. It never silently falls back to the other: a server quietly
//! serving from the wrong engine is the worst possible outcome of this seam, and
//! it is the one failure a measurement cannot see.
//!
//! That rule is kept **by construction** rather than by a check. [`Engine`]'s
//! variants are themselves `#[cfg]`-gated, so in a gix-only build the value
//! `Engine::Znippy` does not exist and cannot be constructed by any path,
//! mistaken or otherwise. [`Engine::from_name`] still *knows* the name, which is
//! what lets it refuse with "this binary was built without it" rather than with
//! "unknown engine".
//!
//! # znippy's build-time C toolchain, and why this module is where the cost lands
//!
//! Cargo reads a path dependency's manifest at **resolve** time whether or not
//! the feature selecting it is on. Naming `znippy-plugin-git` here therefore
//! makes a sibling znippy checkout a hard requirement of *every* build of this
//! workspace. MEASURED 2026-08-10 — see this crate's `Cargo.toml`, which records
//! the command, the output and the 30 packages it adds to `Cargo.lock`.

use std::io::BufRead;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{anyhow, bail, Context, Result};
use gunnar_store::git_storage_trait::GitServe;
use gunnar_store::RepoStore;

#[cfg(not(any(feature = "engine-znippy", feature = "engine-gix")))]
compile_error!(
    "gunnar-server must be built with at least one storage engine: `engine-znippy` (the \
     default) or `engine-gix`. A server binary with no engine is not a smaller server — it \
     is the defect this seam replaced: it built, it shipped, it answered every push with \
     `this binary carries no storage engine`, and the build matrix could not tell it from \
     the real one."
);

#[cfg(feature = "engine-gix")]
mod gix_engine;
mod pack_end;
mod store_cache;
#[cfg(feature = "engine-znippy")]
pub mod writer;
/// **The engine seam.** `znippy_plugin_git::` may be named here and nowhere
/// else — `xtask`'s `Rule::EngineByName` says so, because this file's own
/// header claims to be the only place in gunnar that names an implementation.
///
/// Both sides of the 2026-08-31 merge had built this module and given it a
/// different name and a different scope. Master's `archive` moved the SIX names
/// that broke the rule (`archive_in`/`out`/`out_to`, `parse_explode_budget`,
/// `human_bytes`); this one moves the whole adapter — `store_root`, `shutdown`,
/// `cancel_index`, `index_progress`, `explode_use`, `set_explode_budget`,
/// `drop_explode_cache`, `open`, `ingest`, `maintain` — and is the answer that
/// makes the next engine a file beside `gix_engine.rs` rather than a `#[cfg]`
/// in `admin.rs`. So the wider one wins and `archive.rs` retires into it, with
/// master's type discipline carried across: the archive receipts are the
/// plugin's, re-exported, never re-declared.
///
/// `pub` because `admin.rs` and `control_plane.rs` have to reach it: a seam
/// nothing outside can reach is a rule nobody can keep.
#[cfg(feature = "engine-znippy")]
pub mod znippy_engine;

/// **An encrypted repository's push, checked before a byte of it is kept**
/// (crypt-level2 item 4; `crate::encrypted_quarantine` re-exports it). Here,
/// not beside `transport.rs`, because it walks the pushed pack with the znippy
/// engine's own reader — and the engine is named in this seam and nowhere else.
#[cfg(feature = "engine-znippy")]
pub mod encrypted_quarantine;

/// The writer arm the znippy engine built its stores on (`uring` / `safe`),
/// for the dashboard's disk cell; `None` before the first store, or on an
/// engine that has no arms.
pub fn writer_arm() -> Option<&'static str> {
    #[cfg(feature = "engine-znippy")]
    {
        writer::selected().map(|s| s.arm.as_str())
    }
    #[cfg(not(feature = "engine-znippy"))]
    {
        None
    }
}

/// **Release every store this process cached, on the way out.**
///
/// Engine-neutral, and deliberately not a method on [`Engine`]: it is not "shut
/// down the engine this run selected", it is "let go of everything anything
/// opened", and a build carrying both engines has two caches whichever one the
/// request used.
///
/// The znippy arm is where this earns its keep — see
/// [`znippy_engine::shutdown`] for the measurement and for why the cache itself
/// stays. It is idempotent, and safe to call in a process that opened no
/// repository at all: the `OnceLock` behind the cache is initialised on demand
/// and an empty map clears to an empty map.
///
/// The gix arm now has something to release too. It has no database to be told
/// it was closed, but since 2026-08-11 it caches a `GixStore` per store root for
/// correctness — one handle so that a push refreshes the view the reader serves
/// from ([`store_cache`]) — and those hold mmaps over every `.pack` and `.idx`
/// they have touched. The line the old comment said would be needed "later" is
/// the second one below.
pub fn shutdown() {
    #[cfg(feature = "engine-znippy")]
    znippy_engine::shutdown();
    #[cfg(feature = "engine-gix")]
    gix_engine::shutdown();
}

/// The environment variable that picks an engine in a build carrying both.
///
/// An environment variable and not a flag, for now, because the flag would live
/// on [`gunnar_common::cli`]'s `ServeArgs` — a crate this change does not own —
/// and every verb that opens a repository needs the choice, not just `serve`.
/// The variable is read exactly once, in [`Engine::selected`], on the startup
/// path.
pub const ENGINE_ENV: &str = "GUNNAR_ENGINE";

/// **A storage engine this binary actually carries.**
///
/// The variants are `#[cfg]`-gated on purpose: a value of this type is proof
/// that the engine it names is linked. There is no "selected but absent" state
/// to guard against, because it cannot be represented.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Engine {
    /// `znippy-plugin-git` — Arrow-IPC in a znippy archive.
    #[cfg(feature = "engine-znippy")]
    Znippy,
    /// `storage-git-gix` — `gix-odb` / `gix-ref` on a plain filesystem.
    #[cfg(feature = "engine-gix")]
    Gix,
}

/// The engine a build carrying znippy runs unless told otherwise.
#[cfg(feature = "engine-znippy")]
pub const DEFAULT_ENGINE: Engine = Engine::Znippy;

/// With no znippy there is exactly one engine left, and it is the default by
/// arithmetic rather than by preference. The `compile_error!` above is what
/// makes this exhaustive.
#[cfg(all(not(feature = "engine-znippy"), feature = "engine-gix"))]
pub const DEFAULT_ENGINE: Engine = Engine::Gix;

impl std::fmt::Display for Engine {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.name())
    }
}

/// **Two resolver workers on a twin.** MEASURED 2026-09-15
/// (`tests/twin_snapshot_memory.rs`): one worker per core, each inflating a whole blob, grew
/// RssAnon +133 MiB for a 512 MiB snapshot install even with explode and graph off. Handed to
/// `znippy_plugin_git::set_resolve_workers_default` by [`Engine::hold_twin_memory_bound`].
pub const TWIN_RESOLVE_WORKERS: usize = 2;
/// **16 MiB of resolver lookahead on a twin** (same measurement) — the bytes the workers may
/// run ahead of the ordered output by. znippy's own default is 32 MiB.
pub const TWIN_RESOLVE_LOOKAHEAD_BYTES: u64 = 16 << 20;
/// **16 MiB explode budget on a twin** (same measurement). With the policy `off` nothing is
/// stored at all; the cap is what keeps a later policy change bounded.
pub const TWIN_EXPLODE_BUDGET_BYTES: u64 = 16 << 20;

/// **Where the nordisk repositories live** — the ONE place an engine's source
/// URL is spelled. Today codeberg; the day they move to gunnar.rs this is
/// `https://gunnar.rs/nordisk` and every engine's link follows.
pub const NORDISK_REPOS: &str = "https://codeberg.org/nordisk";

impl Engine {
    /// Every engine compiled into this binary, in preference order.
    pub const COMPILED: &'static [Engine] = &[
        #[cfg(feature = "engine-znippy")]
        Engine::Znippy,
        #[cfg(feature = "engine-gix")]
        Engine::Gix,
    ];

    /// The name this engine is selected by and reported under.
    pub fn name(self) -> &'static str {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => "znippy",
            #[cfg(feature = "engine-gix")]
            Engine::Gix => "gix",
        }
    }

    /// **Can replication read a repository this engine stores?**
    ///
    /// It reads one by opening it as a *filesystem git repository* — `gix::open`
    /// in [`crate::mirror`], then `gunnar-client`'s send-pack, which is proven
    /// against the real `git receive-pack` and must not be twinned. An engine
    /// that writes no `.git` layout cannot be read that way, and the failure
    /// lands inside the retry loop where nobody sees it: the server accepts
    /// every push and the standby stays empty.
    ///
    /// So it is asked at start-up instead, and a server that cannot do what it
    /// was configured to do refuses to come up. Closing the gap means a
    /// server-side send-pack over `GitServe` — the store already emits packs
    /// for `upload-pack` — at which point this predicate goes away rather than
    /// gaining an arm.
    pub fn replication_can_read_repositories(self) -> bool {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => false,
            #[cfg(feature = "engine-gix")]
            Engine::Gix => true,
        }
    }

    /// **The engine's crate**, as it is named on the Storage page and linked.
    pub fn plugin(self) -> &'static str {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => "znippy-plugin-git",
            #[cfg(feature = "engine-gix")]
            Engine::Gix => "storage-git-gix",
        }
    }

    /// **What it stores and how**, in one line — the Storage page's tooltip.
    pub fn summary(self) -> &'static str {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => "Arrow-IPC archive; objects + refs",
            #[cfg(feature = "engine-gix")]
            Engine::Gix => "gix-odb packs + gix-ref (gitoxide); objects + refs",
        }
    }

    /// **Where the engine's source lives**, off [`NORDISK_REPOS`]: znippy's
    /// own repository; `storage-git-gix` is a crate of edda's (built on
    /// gitoxide's `gix-odb` / `gix-ref`).
    pub fn repo_url(self) -> String {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => format!("{NORDISK_REPOS}/znippy"),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => format!("{NORDISK_REPOS}/edda"),
        }
    }

    /// **The engine, in parts** — name, one-line description, source
    /// repository — for `Admin.StorageInfo`'s `engine` (2026-10-02).
    pub fn report(self) -> gunnar_proto::StorageEngine {
        gunnar_proto::StorageEngine {
            name: self.plugin().to_owned(),
            description: self.summary().to_owned(),
            repo_url: self.repo_url(),
        }
    }

    /// What answers, in one line, for an operator reading `storage info`:
    /// `znippy-plugin-git (Arrow-IPC archive; objects + refs)`, made of the
    /// same parts [`Self::report`] sends apart.
    pub fn describe(self) -> String {
        format!("{} ({})", self.plugin(), self.summary())
    }

    /// The engines this binary carries, as a human-readable list.
    pub fn compiled_list() -> String {
        Self::COMPILED
            .iter()
            .map(|e| e.name())
            .collect::<Vec<_>>()
            .join(", ")
    }

    /// **Resolve a name to an engine, or refuse.**
    ///
    /// Three outcomes, and the middle one is the whole point:
    ///
    /// * a name this build carries → that engine;
    /// * a name this build *knows* but did not compile in → refused, saying so;
    /// * anything else → refused as unknown.
    ///
    /// Never a fallback. See the module docs.
    pub fn from_name(name: &str) -> Result<Engine> {
        match name.trim().to_ascii_lowercase().as_str() {
            #[cfg(feature = "engine-znippy")]
            "znippy" => Ok(Engine::Znippy),
            #[cfg(not(feature = "engine-znippy"))]
            "znippy" => Err(not_compiled_in("znippy")),

            #[cfg(feature = "engine-gix")]
            "gix" => Ok(Engine::Gix),
            #[cfg(not(feature = "engine-gix"))]
            "gix" => Err(not_compiled_in("gix")),

            other => Err(anyhow!(
                "unknown storage engine {other:?}; the engine names are `znippy` and `gix`, and \
                 this binary carries {}",
                Self::compiled_list()
            )),
        }
    }

    /// **The engine this process runs**, decided once, on the startup path.
    ///
    /// [`ENGINE_ENV`] if it is set and non-empty, [`DEFAULT_ENGINE`] otherwise.
    /// An unset variable is *not* "no engine": it is the default, which is
    /// znippy wherever znippy is compiled in.
    pub fn selected() -> Result<Engine> {
        match std::env::var(ENGINE_ENV) {
            Ok(raw) if !raw.trim().is_empty() => Self::from_name(&raw).with_context(|| {
                format!("{ENGINE_ENV}={raw:?} selects the storage engine for this process")
            }),
            _ => Ok(DEFAULT_ENGINE),
        }
    }

    /// Open — creating if absent — the repository at `git_dir`.
    ///
    /// Unconditional. "It holds nothing yet" is the normal state of a
    /// repository's first push, not a reason to answer with something else:
    /// gating on what a repository already holds is what refuses every first
    /// push, which is a mistake this seam has already made once.
    pub fn open_repo(self, git_dir: &Path, hash_kind: gix_hash::Kind) -> Result<Arc<OpenRepo>> {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::open(git_dir, hash_kind),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => gix_engine::open(git_dir, hash_kind),
        }
    }

    /// Stop any index work on the repository at `git_dir` — the engine half
    /// of `repo delete`. A no-op for an engine without a background indexer.
    pub fn cancel_index(self, git_dir: &Path) {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::cancel_index(git_dir),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => {
                let _ = git_dir;
            }
        }
    }

    /// Index work in flight across every open repository, as
    /// `(git_dir, snapshot)`. Empty for an engine without a background indexer.
    pub fn index_progress(self) -> Vec<(PathBuf, IndexSnapshot)> {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::index_progress()
                .into_iter()
                .map(|(dir, s)| {
                    (
                        dir,
                        IndexSnapshot {
                            active: s.active,
                            resumed: s.resumed,
                            pack_id: s.pack_id,
                            objects_done: s.objects_done,
                            objects_total: s.objects_total,
                            bytes_done: s.bytes_done,
                            bytes_total: s.bytes_total,
                            phase: s.phase.as_str(),
                            started_at_ms: s.started_at_ms,
                            reinflations: s.reinflations,
                            pauses: s.pauses,
                        },
                    )
                })
                .collect(),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => Vec::new(),
        }
    }

    /// **Every open repository's explode band**, as `(git_dir, band)`.
    ///
    /// The exploded tier is a read cache with a byte budget, not storage: a
    /// blob is written into it by the first reader that inflates it and
    /// unlinked from it when the budget says so. Both numbers are here because
    /// an operator asked to see a repository at its limit — an account at its
    /// budget still serves every object, only slower, and nothing else on the
    /// dashboard would say which repositories those are.
    ///
    /// A `metadata` call per pool, no store lock. Empty for an engine with no
    /// such tier.
    pub fn explode_use(self) -> Vec<(PathBuf, ExplodeBand)> {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::explode_use(),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => Vec::new(),
        }
    }

    /// **Indexed objects of every open store**, as `(git_dir, count)` — the
    /// `objects` meter of `Usage.Read`, i.e. what this process is paying RAM
    /// for (~382 B an object, measured). A closed store is absent, not zero.
    ///
    /// Empty on the gix engine: it keeps no in-RAM object index per open
    /// store (its odb is mmapped pack indexes), so there is no per-object RAM
    /// price to report and reporting one would be a made-up bill.
    pub fn open_object_counts(self) -> Vec<(PathBuf, u64)> {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::open_object_counts(),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => Vec::new(),
        }
    }

    /// **Set the explode budget of every open repository under `git_dir`.**
    ///
    /// A prefix, so one call moves a whole account. Returns how many stores
    /// were changed — applied output, so a control plane that reached nothing
    /// (every repository closed) can say so instead of reporting success.
    pub fn set_explode_budget(self, git_dir: &Path, bytes: u64) -> usize {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::set_explode_budget(git_dir, bytes),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => {
                let _ = (git_dir, bytes);
                0
            }
        }
    }

    /// **A twin keeps no derived memo** (`.nornir/twin-transaction-log.md` C2): the znippy
    /// engine's persisted commit graph (`objects.graph`) is off for every store this process
    /// opens from now on. Process-wide, because the box's role is — called when stores open as
    /// a twin and when a fresh box becomes one. The gix engine keeps no such file.
    pub fn hold_no_graph_memo(self) {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::set_graph_cache_default(false),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => {}
        }
    }

    /// **A twin holds no read cache and resolves narrow** (`twin-memory-bound`, and lane T3's
    /// one-medium run): the znippy engine's explode policy is `off`, its explode budget is
    /// [`TWIN_EXPLODE_BUDGET_BYTES`], and the indexer's resolver runs [`TWIN_RESOLVE_WORKERS`]
    /// wide with [`TWIN_RESOLVE_LOOKAHEAD_BYTES`] of lookahead, for every store this
    /// process opens from now on. MEASURED: +36 MiB of RssAnon for a 512 MiB snapshot install
    /// against a 64 MiB ceiling, against +149 MiB unbounded — and, with the policy left alone,
    /// five `objects.exploded` files on a twin's volume that a twin may not hold.
    ///
    /// Process-wide, because the box's role is. It is a SIBLING of
    /// [`Engine::hold_no_graph_memo`] and not folded into it, because the two answer different
    /// files (`objects.graph` against `objects.exploded` and the resolver's heap) and a reader
    /// who assumed the graph door covered the cache is exactly how the twin came to hold five
    /// of them. The gix engine has neither.
    pub fn hold_twin_memory_bound(self) {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::hold_twin_memory_bound(),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => {}
        }
    }

    /// **And the stores already open when that happened** — their exploded pools unlinked and
    /// capped, since a store keeps the policy it opened with. Returns the bytes unlinked.
    pub fn bound_open_stores_to_twin(self) -> u64 {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::bound_open_stores_to_twin(),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => 0,
        }
    }

    /// **`repo gc` reclaims the read cache**: every lazily-exploded object
    /// unlinked, the commit-and-tree pool kept because that is the graph.
    /// Returns the bytes that left the filesystem.
    pub fn drop_explode_cache(self, git_dir: &Path) -> u64 {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::drop_explode_cache(git_dir),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => {
                let _ = git_dir;
                0
            }
        }
    }

    /// **Take one pushed packfile straight into storage, off the wire.**
    ///
    /// The bytes go from the transport into the engine and are durable when this
    /// returns. Nothing is staged through a packfile on the way: a seam spelled
    /// in paths cannot express a server without them, which is why this one is
    /// spelled in bytes.
    ///
    /// Everything but `want_objects` is engine-neutral and lives in
    /// [`ingest_common`] — the pack header, the trailer and
    /// [`GitOps::put_pack`](git_storage_trait::GitOps::put_pack) are the same
    /// three facts whichever engine stored them. Enumerating the ids a push
    /// introduced is **not** neutral: a pack entry carries no oid, so an engine
    /// answers it out of its own index or not at all.
    ///
    /// # `progress`, and why it is threaded rather than fetched
    ///
    /// `Some(sink)` asks the engine to say how many objects of this pack it has
    /// taken in **while it is still taking them in**
    /// ([`GitOps::put_pack_observed`](git_storage_trait::GitOps::put_pack_observed)).
    /// The count belongs to the code that splits the pack and to nothing else —
    /// deriving it anywhere on the way here means inflating every entry a second
    /// time, which is a second pack decoder (one implementation) — so the only thing this
    /// layer can do is carry the sink down and let the walk publish.
    ///
    /// # `before_store`, and why the pushed pack is framed HERE
    ///
    /// Called with exactly the bytes the engine is about to store, before it stores them — the
    /// transaction log's `Pack` frame (`gunnar_core::txlog`, the twin). This path opens the
    /// engine by `git_dir` and never passes the `OpenRepo::logged` tee, so without it a pushed
    /// pack was never a frame while its ref update was: MEASURED 2026-09-15 under KVM, the twin
    /// and a restored primary both refused `refs/heads/main would point at …, which this
    /// repository does not have`. An error from the hook refuses the push before a byte lands.
    ///
    /// `None` is not a degraded mode: a push over SSH or smart HTTP has no job
    /// record and nobody polling one, so it asks for nothing and pays for
    /// nothing. The mirror queue is the caller that has a console watching.
    pub fn ingest_push(
        self,
        git_dir: &Path,
        hash_kind: gix_hash::Kind,
        pack: &mut dyn BufRead,
        objects: ObjectSet,
        progress: Option<&dyn git_storage_trait::PackProgress>,
        before_store: Option<BeforeStore<'_>>,
    ) -> Result<PushReceipt> {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::ingest(git_dir, hash_kind, pack, objects, progress, before_store),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => gix_engine::ingest(git_dir, hash_kind, pack, objects, progress, before_store),
        }
    }

    /// `gunnar gc` for one repository: seal, then reclaim.
    ///
    /// The order is the correctness argument rather than a preference. Sealing
    /// folds pending index work into the derived tables; reclaiming computes its
    /// live set out of those tables. Reclaiming first would judge liveness
    /// against a graph missing everything pushed since the last drain, and
    /// tombstone objects reachable only from those commits.
    ///
    /// Only the znippy arm has a seal. The gix arm's `gc` reads the packs and
    /// the ref store directly, so there is no pending index work for a seal to
    /// fold in, and it reports `sections_sealed: 0` — which is the truth and not
    /// a stub.
    pub fn maintain(self, git_dir: &Path, hash_kind: gix_hash::Kind) -> Result<MaintenanceReport> {
        match self {
            #[cfg(feature = "engine-znippy")]
            Engine::Znippy => znippy_engine::maintain(git_dir, hash_kind),
            #[cfg(feature = "engine-gix")]
            Engine::Gix => gix_engine::maintain(git_dir, hash_kind),
        }
    }
}

/// The refusal a build that does not carry `name` gives when asked for it.
///
/// Written out in full rather than as "unsupported", because the operator's next
/// action is a rebuild and this is where they find out which flag.
#[allow(dead_code)]
fn not_compiled_in(name: &str) -> anyhow::Error {
    anyhow!(
        "this binary was built without the `{name}` storage engine; it carries {}. Rebuild with \
         `--features engine-{name}`, or select an engine that is here. Refusing rather than \
         serving from a different engine: a server that quietly answers out of the wrong store \
         is worse than one that will not start.",
        Engine::compiled_list()
    )
}

/// **One repository, opened on whichever engine this process carries.**
///
/// The value [`Engine::open_repo`] hands back, and the *only* implementor of
/// [`RepoStore`] in gunnar. It is a **value**, not a seam: nothing in it can be
/// implemented by a backend and no engine can behave differently through it —
/// the engine-specific part is the one line in each adapter that constructs it.
/// It replaced two near-identical `RepoStore` impls (`ZnippyRepo`, `GixRepo`)
/// that differed only in which concrete store they held.
///
/// # Why it exists at all
///
/// `gunnar-server` is a *serving* process, and everything it does with a
/// repository needs the **reading** contract:
/// [`UploadPack::new`](gunnar_wire::UploadPack::new) and
/// [`ReceivePack::new`](gunnar_wire::ReceivePack::new) take `&dyn GitServe`,
/// `gunnar-policy`'s ancestry walk and signature gate take `&dyn GitServe`, and
/// `HEAD` is reachable only through [`GitServe::head`] / [`GitServe::set_head`]
/// because the `GitServe` HEAD ruling settled that it is not a row in `GitOps::refs`.
///
/// # ONE handle, since [`RepoStore::git`] widened — 2026-08-10
///
/// This struct carried **two** fat pointers to the same store, `ops` and
/// `serve`, for one day: `RepoStore::git()` returned `Arc<dyn GitOps>`, the
/// hand-off from the reading half to the storage half is one-way, and `dyn` has
/// no downcast — so a server that had only the narrow handle could not serve a
/// clone from it. The fix was one line in `gunnar-store`:
/// `git()` hands out `Arc<dyn GitServe>`, which is still one method and still
/// One repository's index work, engine-neutral. See
/// `znippy_plugin_git::progress::Snapshot` for the source of every field.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IndexSnapshot {
    pub active: bool,
    pub resumed: bool,
    pub pack_id: u64,
    pub objects_done: u64,
    pub objects_total: u64,
    pub bytes_done: u64,
    pub bytes_total: u64,
    pub phase: &'static str,
    pub started_at_ms: u64,
    pub reinflations: u64,
    pub pauses: u64,
}

/// One repository's exploded read cache: what it holds, what it may hold, and
/// what has already been evicted from it. Engine-neutral by construction —
/// `znippy_plugin_git::exploded_cache::CacheUse` copied field for field, so a
/// dashboard cell cannot tell which engine answered.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct ExplodeBand {
    /// The pool the indexer writes at push time (commits and trees). Never
    /// evicted: it is what `graph`, `reach` and a tag peel read.
    pub eager_bytes: u64,
    /// The pool first reads fill and the budget empties.
    pub lazy_bytes: u64,
    /// The cap. `0` is unlimited.
    pub budget: u64,
    pub segments: u64,
    pub evicted_segments: u64,
    /// Bytes that actually left the filesystem, measured before the unlink.
    pub evicted_bytes: u64,
    pub written_back: u64,
}

impl ExplodeBand {
    pub fn used(&self) -> u64 {
        self.eager_bytes + self.lazy_bytes
    }
    /// True while the repository is at or over its quota — the row an operator
    /// wants to find. Not an error state: it reads correctly and more slowly.
    pub fn at_limit(&self) -> bool {
        self.budget != 0 && self.used() >= self.budget
    }
}

impl IndexSnapshot {
    pub fn percent(&self) -> u64 {
        (self.objects_done * 100)
            .checked_div(self.objects_total)
            .map_or(0, |percent| percent.min(100))
    }
}

/// one authority because `GitServe` **is** `GitOps`.
///
/// So the second field is gone, and with it the paragraph that explained why
/// two fields were not two handles and why they could not be one upcast (trait
/// upcasting stabilised in Rust 1.86 and this workspace declares `1.85`). There
/// is one `Arc`, one refcount cell, and one view of one transaction.
pub struct OpenRepo {
    git_dir: PathBuf,
    hash_kind: gix_hash::Kind,
    serve: Arc<dyn GitServe>,
    /// The engine's store UNDER the transaction-log tee, when `serve` is the
    /// tee. The twin applies the primary's frames through this one, so its
    /// own log holds the primary's numbers and never a second copy.
    raw: Option<Arc<dyn GitServe>>,
}

impl OpenRepo {
    /// Wrap one engine's concrete store.
    ///
    /// `S: GitServe` is the requirement that keeps an engine honest: a backend
    /// that implements only the storage half cannot be opened by this server,
    /// and the refusal is a compile error in the engine's adapter rather than a
    /// repository that opens and then cannot answer a clone.
    pub(crate) fn new<S: GitServe + 'static>(
        git_dir: &Path,
        hash_kind: gix_hash::Kind,
        store: Arc<S>,
    ) -> Arc<Self> {
        Arc::new(OpenRepo {
            git_dir: git_dir.to_path_buf(),
            hash_kind,
            serve: store as Arc<dyn GitServe>,
            raw: None,
        })
    }

    /// **The reading contract**: inflated objects, headers, `HEAD`, `emit_pack`,
    /// `select` — and, by the `GitServe: GitOps` supertrait, the eleven as well.
    ///
    /// This is what the serving path holds. It blocks: every caller below is
    /// already inside `spawn_blocking` or on a control-plane worker, and
    /// The two-tier rule keeps the serving tier serial — request concurrency, no
    /// intra-request fan-out.
    pub fn serve(&self) -> &dyn GitServe {
        self.serve.as_ref()
    }

    /// **The same store, behind the transaction log.** Every write through
    /// the returned handle is a frame in `log` before the engine sees it
    /// (`gunnar_core::txlog::tee::LoggedGit`); every read is the engine's.
    /// `repo` is the catalog key and `hash` its hash kind, as the frame
    /// records them for the twin.
    pub(crate) fn logged(self: &Arc<Self>, log: Arc<gunnar_core::txlog::Log>, repo: &str, hash: &str) -> Arc<Self> {
        let tee = gunnar_core::txlog::tee::LoggedGit::new(Arc::clone(&self.serve), log, repo, hash);
        Arc::new(OpenRepo {
            git_dir: self.git_dir.clone(),
            hash_kind: self.hash_kind,
            serve: Arc::new(tee) as Arc<dyn GitServe>,
            raw: Some(Arc::clone(&self.serve)),
        })
    }

    /// **The store with no log in front of it.** The twin's apply path and
    /// nothing else: a write here is invisible to this box's transaction log.
    pub fn unlogged(&self) -> &dyn GitServe {
        self.raw.as_deref().unwrap_or(self.serve.as_ref())
    }

    /// Where this repository lives. The inherent twin of
    /// [`RepoStore::git_dir`], so callers that only want a path do not have to
    /// have the trait in scope.
    pub fn git_dir(&self) -> &Path {
        &self.git_dir
    }

    /// This repository's hash algorithm — asked, never defaulted. See
    /// [`RepoStore::hash_kind`].
    pub fn hash_kind(&self) -> gix_hash::Kind {
        self.hash_kind
    }
}

impl RepoStore for OpenRepo {
    fn git_dir(&self) -> &Path {
        &self.git_dir
    }

    fn hash_kind(&self) -> gix_hash::Kind {
        self.hash_kind
    }

    fn git(&self) -> Arc<dyn GitServe> {
        Arc::clone(&self.serve)
    }
}

/// **How badly the caller needs the object set a push introduced.**
///
/// A `bool` said only *"enumerate it"*, and that conflated two callers with
/// very different stakes:
///
/// * a **policy gate** over the pushed objects cannot decide without the set,
///   so an engine that cannot produce one must refuse the push — passing it on
///   an answer the engine never gave is the silent-yes this whole seam exists
///   to prevent;
/// * the **push attestation** wants it and can record its absence
///   (`new_object_oids_unknown`), so an engine that cannot produce one must still
///   serve the push.
///
/// With one flag, teaching the attestation to ask made every push refuse on the
/// `gix` arm, which cannot enumerate at all. Two answers, one type.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ObjectSet {
    /// A gate needs it. An engine that cannot answer must return an error.
    Required,
    /// Wanted for the record. An engine that cannot answer returns `None` and
    /// the push is served.
    Wanted,
    /// Nobody asked. Do not pay for it.
    NotNeeded,
}

/// The hook [`Engine::ingest_push`] runs on the pack before a byte of it is
/// stored; an error refuses the push.
pub type BeforeStore<'a> = &'a dyn Fn(&[u8]) -> Result<()>;

/// What one push stored, read back off what was written.
///
/// Nothing here is restated from the request: the object count and version come
/// off the stored pack's own 12-byte header, and `pack_id` is its trailing
/// checksum.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PushReceipt {
    /// From the pack's own header.
    pub num_objects: u32,
    /// The pack format version, likewise.
    pub version: u32,
    /// The pack's trailing checksum, which is its identity.
    pub pack_id: Vec<u8>,
    /// Bytes stored, which for a verbatim store is bytes received.
    pub pack_bytes: u64,
    /// The object ids this push introduced, when the caller asked for them.
    ///
    /// `None` is **not asked for**, never "none were introduced" — a gate handed
    /// an empty set would pass, and one handed `None` refuses.
    pub new_oids: Option<Vec<Vec<u8>>>,
    /// Wall time of the ingest.
    pub elapsed_ms: u64,
}

/// What one maintenance run did. Applied outputs, read off the result.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct MaintenanceReport {
    /// Which strategy ran, by its own name.
    pub strategy: String,
    /// Size before and after — the pair, because a "reclaimed" figure alone
    /// cannot tell a run that compacted from one that did nothing.
    pub bytes_before: u64,
    pub bytes_after: u64,
    /// Live index rows carried across. Maintenance never changes this, so a run
    /// that did is a bug and this is how it becomes visible.
    pub rows: u64,
    /// Packs whose every object was dead, and whose rows were tombstoned.
    pub retired_packs: u64,
    /// Whether the result was read back and checked. `false` is not failure —
    /// an in-place compaction verifies after it has already committed.
    pub verified: bool,
    /// Reserved sections the seal produced. Always `0` on an engine with no
    /// seal; see [`Engine::maintain`].
    pub sections_sealed: usize,
}

// ── the engine-neutral half of a push ────────────────────────────────────────

/// One pushed pack, read off the wire, with the three facts its own bytes carry.
///
/// [`Debug`] is written out rather than derived: `bytes` is a whole packfile,
/// and a derive would put tens of megabytes of it into any `expect_err` message
/// or `tracing` line that ever touched one.
pub(crate) struct PushedPack {
    pub bytes: PackBytes,
    /// From the pack's own 12-byte header, so nothing derived can disagree.
    pub num_objects: u32,
    pub version: u32,
}

/// Packs at or above this many bytes are spooled to disk while they arrive.
pub(crate) const SPOOL_ABOVE: usize = 64 << 20;

/// **How many of the first `consumed` bytes of a pushed pack are in its spool
/// file, on the disk** — the rule [`read_pushed_pack_spooling_at`] follows,
/// stated once for the space ledger (`crate::push_space::Gated`), which gives a
/// spooled byte's promise back the moment it is written: below the threshold
/// the pack is on the heap and nothing is on the disk; the moment it reaches
/// the threshold everything consumed so far is written out, and from then on
/// each chunk is written BEFORE it is consumed.
/// `the_spool_holds_what_spooled_on_disk_says_at_every_consume` holds the two
/// to each other.
pub(crate) fn spooled_on_disk(consumed: u64, spool_above: u64) -> u64 {
    if consumed >= spool_above {
        consumed
    } else {
        0
    }
}

/// The pushed bytes: on the heap for an ordinary push, **on disk and mapped**
/// for a big one.
///
/// MEASURED on the KVM rig 2026-08-30: the linux mirror push (8.12
/// GiB) killed an 8 GB and a 12 GB appliance *while the client was still
/// sending* — `Kernel panic: System is deadlocked on memory` with no
/// `store.znippy.ingested` line — because the whole pack was one `Vec<u8>`
/// grown by `extend_from_slice` (a 16 GiB capacity step past 8 GiB). Above
/// [`SPOOL_ABOVE`] the bytes now go to a temp file beside the store as they
/// arrive and come back as a read-only mapping: page cache the kernel can
/// reclaim, which `MemAvailable` (and the indexer's watchdog) counts as free.
/// The walk, the closure check and the verbatim append read the mapping the
/// same way they read a `Vec`.
pub(crate) enum PackBytes {
    Heap(Vec<u8>),
    Spooled {
        /// Removed when this drops — after `put_pack` has copied the bytes
        /// into the archive.
        _file: tempfile::NamedTempFile,
        map: memmap2::Mmap,
    },
}

impl std::ops::Deref for PackBytes {
    type Target = [u8];
    fn deref(&self) -> &[u8] {
        match self {
            PackBytes::Heap(v) => v,
            PackBytes::Spooled { map, .. } => map,
        }
    }
}

impl PartialEq<Vec<u8>> for PackBytes {
    fn eq(&self, other: &Vec<u8>) -> bool {
        &**self == other.as_slice()
    }
}

impl std::fmt::Debug for PackBytes {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            PackBytes::Heap(v) => write!(f, "{} byte(s) on the heap", v.len()),
            PackBytes::Spooled { map, .. } => write!(f, "{} byte(s) spooled", map.len()),
        }
    }
}

impl std::fmt::Debug for PushedPack {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PushedPack")
            .field("bytes", &format_args!("{} byte(s)", self.bytes.len()))
            .field("num_objects", &self.num_objects)
            .field("version", &self.version)
            .finish()
    }
}

/// **Read exactly one packfile off the wire and read its header back out of
/// it.**
///
/// One copy, called from both engine arms (one implementation): the header is a property of
/// the file, not of the engine that stored it.
///
/// # The shape that does not line up, and the deadlock it caused
///
/// The seam hands a `&mut dyn BufRead` positioned on the packfile's first byte
/// and gives **no length**, because the pkt-line framing does not carry one.
/// `GitOps::put_pack` needs the bytes *as a slice*, because storing them
/// verbatim is a single append of a caller-owned buffer. Until 2026-08-11 the
/// two were joined by `std::io::copy`, i.e. by reading to end of stream, under
/// a comment that said *"nothing follows the pack on that channel"*.
///
/// Nothing follows the pack, and that is the bug. A `receive-pack` client does
/// not hang up after its pack: it holds the channel open and waits for the
/// status report, which is the next thing this server owes it. So the server
/// waited for an EOF the client would only send once it had been answered, and
/// the client waited for an answer the server would only send once it had seen
/// EOF. `gunnar.real_repo_clone` sat in that state for 452 seconds with **0 s
/// of CPU on either side**, holding all 12 750 objects of the pack it was
/// waiting for more of; see [`pack_end`] for the full measurement.
///
/// It survived because every client the suite drives happens to close its write
/// end — stock `git push` does, smart HTTP does — and gunnar's own client, which
/// does not have to, is the one that found it.
///
/// So the pack is **delimited**, the way `git index-pack --stdin` delimits it:
/// [`pack_end::PackScanner`] walks the entries and stops on the last byte of the
/// trailer, and this loop consumes from the reader only what the scanner claims.
/// Everything after it — the report the client is waiting for, or nothing at all
/// — is left in the stream, untouched.
/// Both engine arms pass a directory to spool a big pack into; `None` keeps
/// every pack on the heap, which is what the tests below want and what
/// [`read_pushed_pack`] is.
pub(crate) fn read_pushed_pack_spooling(
    pack: &mut dyn BufRead,
    hash_kind: gix_hash::Kind,
    spool_dir: Option<&Path>,
) -> Result<PushedPack> {
    read_pushed_pack_spooling_at(pack, hash_kind, spool_dir, SPOOL_ABOVE)
}

/// [`read_pushed_pack_spooling`] with the threshold as a parameter — a test's
/// door to the spool path without a 64 MiB fixture.
pub(crate) fn read_pushed_pack_spooling_at(
    pack: &mut dyn BufRead,
    hash_kind: gix_hash::Kind,
    spool_dir: Option<&Path>,
    spool_above: usize,
) -> Result<PushedPack> {
    use std::io::Write as _;
    let mut scanner = pack_end::PackScanner::new(hash_kind.len_in_bytes());
    let mut bytes = Vec::new();
    let mut spool: Option<(tempfile::NamedTempFile, u64)> = None;
    while !scanner.done() {
        // `fill_buf`/`consume` and not `read`: the scanner is shown a *peek* and
        // answers with a prefix length, so the byte after the trailer is never
        // taken out of the reader. A `read` would already have removed it.
        let used = {
            let available = pack
                .fill_buf()
                .context("reading the pushed packfile off the wire")?;
            if available.is_empty() {
                if bytes.is_empty() {
                    bail!("an empty push is not a pack");
                }
                bail!(
                    "the pushed pack ended after {} byte(s), in the middle of {}. The client \
                     hung up mid-push; nothing was stored",
                    bytes.len(),
                    scanner.waiting_for()
                );
            }
            let used = scanner.feed(available)?;
            match &mut spool {
                Some((f, n)) => {
                    f.write_all(&available[..used])
                        .context("spooling the pushed pack")?;
                    *n += used as u64;
                }
                None => {
                    bytes.extend_from_slice(&available[..used]);
                    if let (Some(dir), true) = (spool_dir, bytes.len() >= spool_above) {
                        // Past the threshold: everything so far goes to disk
                        // and the heap copy is released. One row, once.
                        let mut f = tempfile::Builder::new()
                            .prefix("incoming-")
                            .suffix(".pack")
                            .tempfile_in(dir)
                            .with_context(|| {
                                format!("spooling the pushed pack under {}", dir.display())
                            })?;
                        f.write_all(&bytes).context("spooling the pushed pack")?;
                        let n = bytes.len() as u64;
                        bytes = Vec::new();
                        tracing::info!(
                            target: "gunnar::store",
                            event = "push.spooled",
                            dir = %dir.display(),
                            threshold = spool_above,
                            "a pushed pack over {} MiB is spooled to disk while it arrives, not held in RAM",
                            spool_above >> 20
                        );
                        spool = Some((f, n));
                    }
                }
            }
            used
        };
        pack.consume(used);
    }
    let (version, num_objects) = scanner.header();
    let bytes = match spool {
        None => PackBytes::Heap(bytes),
        Some((f, n)) => {
            f.as_file().sync_data().ok();
            // SAFETY: the file is ours alone — created with a unique name,
            // written only above, never written again while mapped, and
            // removed when the mapping's owner drops.
            let map =
                unsafe { memmap2::Mmap::map(f.as_file()) }.context("mapping the spooled pack")?;
            tracing::info!(
                target: "gunnar::store",
                event = "push.spooled.mapped",
                bytes = n,
                "the spooled pack is complete and mapped for the walk and the append"
            );
            PackBytes::Spooled { _file: f, map }
        }
    };
    Ok(PushedPack {
        bytes,
        num_objects,
        version,
    })
}

/// The pack's trailing checksum, which is its identity.
pub(crate) fn pack_trailer(bytes: &[u8], hash_kind: gix_hash::Kind) -> Result<Vec<u8>> {
    let len = hash_kind.len_in_bytes();
    bytes
        .len()
        .checked_sub(len)
        .and_then(|at| bytes.get(at..))
        .map(<[u8]>::to_vec)
        .ok_or_else(|| anyhow!("the pushed pack has no trailing {hash_kind} checksum"))
}

// `open_secret_log` used to live here — an Arrow secret log at
// `<git_dir>/gunnar-secrets`, created by both arms at `open_repo` time to
// satisfy `RepoStore::secrets()`.
//
// It is **deleted** (2026-08-10), and the removal is worth more than the eight
// lines it took back. It ran `create_dir_all` on the way past, so opening *any*
// repository minted a directory, an `O_EXCL` lock file and a `secrets.arrow`
// beside it — per repository, for a store that was never read, on an engine
// whose whole design goal is one file per repository. The only secret store the
// server ever read was the *control* repository's, and that one is now a table
// in `control.redb` (`gunnar_core::secrets`). Neither arm has a secret of its
// own to keep, so neither carries one.

#[cfg(test)]
mod tests {
    use super::pack_end::fixture::fixture_pack;
    use super::*;

    /// ★ **The engine is reported in parts: its name, what it is, and where
    /// its source lives** (owner, 2026-10-02: the Storage page read "Engine
    /// Backend znippy-plugin-git (Arrow-IPC archive; objects + refs)"; it
    /// shows the name, the description as a tooltip, and links the name). The
    /// repository URL comes from ONE constant per forge.
    #[test]
    fn the_engine_is_reported_as_name_description_and_source_repository() {
        for engine in Engine::COMPILED {
            let r = engine.report();
            assert!(!r.name.is_empty() && !r.name.contains(' '), "{engine:?}: the name is one word: {r:?}");
            assert!(!r.description.is_empty() && !r.description.contains(&r.name), "{engine:?}: {r:?}");
            assert!(r.repo_url.starts_with("https://"), "{engine:?}: {r:?}");
            assert_eq!(engine.describe(), format!("{} ({})", r.name, r.description), "the CLI's line is the same parts");
        }
        #[cfg(feature = "engine-znippy")]
        {
            let z = Engine::Znippy.report();
            assert_eq!((z.name.as_str(), z.description.as_str()), ("znippy-plugin-git", "Arrow-IPC archive; objects + refs"));
            assert_eq!(z.repo_url, format!("{NORDISK_REPOS}/znippy"));
            assert_eq!(z.repo_url, "https://codeberg.org/nordisk/znippy", "znippy is on codeberg, not on gunnar.rs (yet)");
        }
        #[cfg(feature = "engine-gix")]
        {
            let g = Engine::Gix.report();
            assert_eq!(g.name, "storage-git-gix");
            assert_eq!(g.repo_url, format!("{NORDISK_REPOS}/edda"), "storage-git-gix's source is edda");
        }
    }

    /// [`read_pushed_pack_spooling`] with no spool directory: every pack stays
    /// on the heap. **Test-only.** Both engine arms pass a real directory, so
    /// a `pub(crate)` version of this shim was dead code in every shipped
    /// binary while looking like a production entry point — it is the shape
    /// `dead_code` caught on 2026-09-03.
    fn read_pushed_pack(pack: &mut dyn BufRead, hash_kind: gix_hash::Kind) -> Result<PushedPack> {
        read_pushed_pack_spooling(pack, hash_kind, None)
    }

    /// **A `receive-pack` client, as it behaves after it has sent its pack: it
    /// waits, holding the channel open, for the status report.**
    ///
    /// Every byte of the pack is handed over. Then the reader does the only
    /// honest thing a socket can do here — it refuses to produce a byte,
    /// because there is not going to be one until the server speaks — and it
    /// counts how many times it was asked.
    ///
    /// **The refusal is an error rather than a block on purpose.** A reader
    /// that blocked would turn this into a test with a stopwatch in it, which
    /// passes on a fast machine and fails on a slow one and proves nothing
    /// about either. `reads_past_the_pack` is the assertion that cannot lie:
    /// the correct number is zero, and it is zero for reasons of protocol, not
    /// of timing.
    struct ClientWaitingForItsReport {
        stream: Vec<u8>,
        at: usize,
        pack_len: usize,
        reads_past_the_pack: usize,
    }

    impl ClientWaitingForItsReport {
        fn holding(pack: &[u8]) -> Self {
            ClientWaitingForItsReport {
                stream: pack.to_vec(),
                at: 0,
                pack_len: pack.len(),
                reads_past_the_pack: 0,
            }
        }

        /// The same client, with a report already sitting behind the pack. Used
        /// to prove the reader is left *positioned* correctly and not merely
        /// unblocked: a delimiter that stopped one byte late would pass the
        /// test above and desynchronise the session here.
        fn with_a_report_behind_the_pack(pack: &[u8], report: &[u8]) -> Self {
            let mut stream = pack.to_vec();
            stream.extend_from_slice(report);
            ClientWaitingForItsReport {
                stream,
                at: 0,
                pack_len: pack.len(),
                reads_past_the_pack: 0,
            }
        }

        fn what_is_left(&self) -> &[u8] {
            &self.stream[self.at..]
        }
    }

    impl std::io::Read for ClientWaitingForItsReport {
        fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
            let available = self.fill_buf()?.len().min(buf.len());
            buf[..available].copy_from_slice(&self.stream[self.at..self.at + available]);
            self.consume(available);
            Ok(available)
        }
    }

    impl BufRead for ClientWaitingForItsReport {
        fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
            if self.at >= self.pack_len && self.at >= self.stream.len() {
                self.reads_past_the_pack += 1;
                return Err(std::io::Error::new(
                    std::io::ErrorKind::WouldBlock,
                    "the client sent its whole pack and is waiting for the status report; this \
                     read would block for ever",
                ));
            }
            // Handed over in small pieces, because a socket does that and a
            // delimiter that only works on one whole buffer is a delimiter that
            // works in tests.
            let end = self.stream.len().min(self.at + 997);
            Ok(&self.stream[self.at..end])
        }

        fn consume(&mut self, amount: usize) {
            self.at += amount;
        }
    }

    /// **RED-WHEN-BROKEN: the push that deadlocked `gunnar.real_repo_clone`.**
    ///
    /// Watched go red against the `std::io::copy` this replaced, 2026-08-11:
    ///
    /// ```text
    /// ---- storage::tests::the_pack_read_stops_at_the_trailer_and_never_waits_for_eof stdout ----
    /// reading the pushed packfile off the wire
    /// Caused by: the client sent its whole pack and is waiting for the status
    ///            report; this read would block for ever
    /// ```
    ///
    /// On the wire that error is not an error. It is 452 seconds of two
    /// processes at 0% CPU with an empty socket between them.
    /// **A big pack is spooled, not held**: with the threshold at
    /// zero the same bytes come back through a mapping of a temp file beside
    /// the store, the file exists while the pack is held and is gone when it
    /// drops. Seen RED by returning `PackBytes::Heap` from the spool arm: the
    /// temp dir stays empty.
    #[test]
    fn a_pack_over_the_threshold_is_spooled_to_disk_and_mapped() {
        let pack = fixture_pack();
        let dir = tempfile::tempdir().expect("tempdir");
        let mut client = std::io::Cursor::new(pack.clone());
        let pushed =
            read_pushed_pack_spooling_at(&mut client, gix_hash::Kind::Sha1, Some(dir.path()), 0)
                .expect("a well-formed pack");
        assert!(
            matches!(pushed.bytes, PackBytes::Spooled { .. }),
            "{:?}",
            pushed.bytes
        );
        assert_eq!(pushed.bytes, pack, "the mapped pack is not the pack sent");
        assert_eq!(pushed.bytes.len(), pack.len());
        let spooled: Vec<_> = std::fs::read_dir(dir.path()).unwrap().flatten().collect();
        assert_eq!(
            spooled.len(),
            1,
            "one incoming-*.pack while the pack is held"
        );
        assert!(spooled[0]
            .file_name()
            .to_string_lossy()
            .starts_with("incoming-"));
        assert_eq!(
            pack_trailer(&pushed.bytes, gix_hash::Kind::Sha1)
                .unwrap()
                .len(),
            20
        );
        drop(pushed);
        assert_eq!(
            std::fs::read_dir(dir.path()).unwrap().count(),
            0,
            "the spool file outlived the pack"
        );
        // Under the threshold: the heap, and nothing on disk.
        let mut client = std::io::Cursor::new(pack.clone());
        let pushed = read_pushed_pack_spooling_at(
            &mut client,
            gix_hash::Kind::Sha1,
            Some(dir.path()),
            usize::MAX,
        )
        .unwrap();
        assert!(matches!(pushed.bytes, PackBytes::Heap(_)));
        assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0);
    }

    /// ★ The space ledger's rule for the spool ([`spooled_on_disk`]) is what
    /// the spool file actually holds, at every consume of the read — the ledger
    /// gives a byte's spool promise back at `consume`, so a rule that ran ahead
    /// of the write would be a byte nobody counts.
    #[test]
    fn the_spool_holds_what_spooled_on_disk_says_at_every_consume() {
        struct Probe<'a> {
            inner: std::io::Cursor<Vec<u8>>,
            dir: &'a Path,
            consumed: u64,
            threshold: u64,
            checked: u32,
        }
        impl std::io::Read for Probe<'_> {
            fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
                std::io::Read::read(&mut self.inner, buf)
            }
        }
        impl BufRead for Probe<'_> {
            fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
                let b = self.inner.fill_buf()?;
                Ok(&b[..b.len().min(37)])
            }
            fn consume(&mut self, n: usize) {
                self.inner.consume(n);
                self.consumed += n as u64;
                let on_disk: u64 = std::fs::read_dir(self.dir)
                    .unwrap()
                    .flatten()
                    .map(|e| e.metadata().unwrap().len())
                    .sum();
                assert_eq!(
                    on_disk,
                    spooled_on_disk(self.consumed, self.threshold),
                    "after {} byte(s) consumed",
                    self.consumed
                );
                self.checked += 1;
            }
        }
        let pack = fixture_pack();
        let threshold = (pack.len() / 3) as u64;
        let dir = tempfile::tempdir().unwrap();
        let mut probe = Probe {
            inner: std::io::Cursor::new(pack.clone()),
            dir: dir.path(),
            consumed: 0,
            threshold,
            checked: 0,
        };
        let pushed = read_pushed_pack_spooling_at(
            &mut probe,
            gix_hash::Kind::Sha1,
            Some(dir.path()),
            threshold as usize,
        )
        .unwrap();
        assert_eq!(pushed.bytes, pack);
        assert!(probe.checked > 3, "{} consume(s)", probe.checked);
    }

    #[test]
    fn the_pack_read_stops_at_the_trailer_and_never_waits_for_eof() {
        let pack = fixture_pack();
        let mut client = ClientWaitingForItsReport::holding(&pack);

        let pushed = read_pushed_pack(&mut client, gix_hash::Kind::Sha1)
            .expect("the pack is complete; nothing more is owed before the report");

        assert_eq!(
            client.reads_past_the_pack, 0,
            "the server made {} read(s) past the last byte of the pack. There is nobody to \
             answer them: the client is waiting for the status report those reads come before, \
             and on a socket this is not an error, it is a deadlock",
            client.reads_past_the_pack
        );
        assert_eq!(pushed.bytes, pack, "the stored pack is not the pack sent");
        assert_eq!(pushed.num_objects, 4);
        assert_eq!(pushed.version, 2);
    }

    /// The reader is left **positioned** on the byte after the trailer, not
    /// merely unblocked. Smart HTTP already worked with an over-reading server
    /// because its body ends; this is what a session that carries on needs.
    #[test]
    fn what_follows_the_pack_is_left_in_the_stream() {
        let pack = fixture_pack();
        let report = b"0009done\n0000";
        let mut client = ClientWaitingForItsReport::with_a_report_behind_the_pack(&pack, report);

        let pushed =
            read_pushed_pack(&mut client, gix_hash::Kind::Sha1).expect("a well-formed pack");

        assert_eq!(pushed.bytes.len(), pack.len());
        assert_eq!(
            client.what_is_left(),
            report,
            "the bytes after the pack were swallowed, so whatever spoke next would be read \
             out of the middle of the pack"
        );
        assert_eq!(client.reads_past_the_pack, 0);
    }

    /// A client that hangs up mid-pack is a **refusal**, not a stored half
    /// push. The old code could not tell this from a completed one: both ended
    /// at EOF.
    #[test]
    fn a_client_that_hangs_up_mid_pack_is_refused_and_says_where() {
        let pack = fixture_pack();
        let truncated = &pack[..pack.len() - 4];
        let mut client = std::io::Cursor::new(truncated.to_vec());

        let err = read_pushed_pack(&mut client, gix_hash::Kind::Sha1)
            .expect_err("half a pack must not be accepted as a push");
        let message = format!("{err:#}");
        assert!(
            message.contains("trailing checksum") && message.contains("nothing was stored"),
            "the refusal must name what never arrived: {message}"
        );
    }

    /// A push with no bytes at all keeps the message it always had.
    #[test]
    fn an_empty_push_is_still_not_a_pack() {
        let mut nothing = std::io::Cursor::new(Vec::new());
        let err = read_pushed_pack(&mut nothing, gix_hash::Kind::Sha1).expect_err("no bytes");
        assert!(format!("{err:#}").contains("an empty push is not a pack"));
    }

    /// The default is znippy wherever znippy is compiled in. Asserted rather
    /// than commented, because "defaulting to the other engine" is a change no
    /// output of a working server would show.
    #[test]
    #[cfg(feature = "engine-znippy")]
    fn the_default_is_znippy() {
        assert_eq!(DEFAULT_ENGINE, Engine::Znippy);
        assert_eq!(Engine::COMPILED[0], Engine::Znippy);
    }

    /// Every compiled engine answers to its own name. A rename that broke this
    /// would make `GUNNAR_ENGINE` silently unusable.
    #[test]
    fn every_compiled_engine_answers_to_its_name() {
        for engine in Engine::COMPILED {
            assert_eq!(
                Engine::from_name(engine.name()).expect("a compiled engine resolves"),
                *engine
            );
        }
    }

    /// An unknown name is refused, and the message says what is here.
    #[test]
    fn an_unknown_engine_is_refused() {
        let err = Engine::from_name("postgres").expect_err("not an engine");
        let msg = format!("{err:#}");
        assert!(msg.contains("unknown storage engine"), "{msg}");
        assert!(msg.contains(&Engine::compiled_list()), "{msg}");
    }

    /// **The rule, asserted in the build that can violate it.**
    ///
    /// In a znippy-only build, asking for gix must refuse and say the engine was
    /// not compiled in — never fall back to znippy. This test only exists in the
    /// build where the fallback is possible, which is the build where it matters.
    #[test]
    #[cfg(all(feature = "engine-znippy", not(feature = "engine-gix")))]
    fn an_engine_that_was_not_compiled_in_refuses_rather_than_falling_back() {
        let err = Engine::from_name("gix").expect_err("gix is not in this build");
        let msg = format!("{err:#}");
        assert!(
            msg.contains("built without the `gix` storage engine"),
            "{msg}"
        );
        assert!(msg.contains("--features engine-gix"), "{msg}");
    }

    /// The same, the other way round, in the gix-only build.
    #[test]
    #[cfg(all(feature = "engine-gix", not(feature = "engine-znippy")))]
    fn znippy_is_refused_when_it_was_not_compiled_in() {
        let err = Engine::from_name("znippy").expect_err("znippy is not in this build");
        assert!(
            format!("{err:#}").contains("built without the `znippy` storage engine"),
            "{err:#}"
        );
    }
}