gunnar-server 3.0.0

//! **gunnar's head-to-head server binary, on whichever engine this build carries.**
//!
//! The twin of `edda/crates/storage-git-gix/examples/git-store-serve.rs`. Same
//! three modes, same argv, same `GUNNAR_BENCH_COPY_STATS` receipt, so `bench/`
//! needs no change at all: its whole seam is
//! `GUNNAR_BENCH_GUNNAR_SERVER=<path>`, and it knows a path, not a crate.
//!
//! ```text
//!   stock git  ──ext:: over a pipe──►  bench meter  ──►  THIS BINARY
//!                                                          │
//!                                            UploadPack / ReceivePack   (gunnar-wire)
//!                                                          │
//!                                             storage::Engine::open_repo
//!                                                          │
//!                                        znippy-plugin-git  ·OR·  storage-git-gix
//! ```
//!
//! # Why it lives in `gunnar-server` and not somewhere else
//!
//! It died with the old znippy engine crate, and znippy engine is not coming back. The three
//! things this binary needs are all here and nowhere else:
//!
//! 1. **the engine seam.** [`gunnar_server::storage::Engine`] is *the only place
//!    in gunnar that names an implementation of `GitOps`*. A bench server that
//!    opened `znippy_plugin_git::GitStore` directly would be a **second** place,
//!    and the two would drift — the exact defect `gunnar-znippy` was deleted for.
//!    Going through `Engine` also means this binary serves **either** engine
//!    (`GUNNAR_ENGINE=gix` on a build carrying both), from one wire.
//! 2. **`gunnar-wire`.** The edda arm cannot reach it (a path
//!    dependency on a sibling gunnar checkout is read at cargo *resolve* time),
//!    so it hand-rolls a v0 driver and says so. Here `UploadPack`/`ReceivePack`
//!    are the same state machines `transport.rs` runs, so this column measures
//!    *the server*, not a bench-shaped imitation of it.
//! 3. **the ingest.** [`Engine::ingest_push`] is the same call a real push lands
//!    in, `PushIngest` and all.
//!
//! A new crate would have had to name `znippy-plugin-git` by path a second time
//! for the sake of an example, which is the build-time cost paid twice for nothing.
//!
//! # The modes
//!
//! | argv | what it does |
//! |---|---|
//! | `prepare <repo>` | build the store for `<repo>`, **outside any timed region**, idempotent |
//! | `git-upload-pack <repo>` | serve a fetch/clone out of that store |
//! | `git-receive-pack <repo>` | receive a push into that store |
//! | `has <repo> <oid>` | a diagnostic: print the **error**, not the `false` it collapses to |
//!
//! `<repo>` is always the *filesystem* repository path the bench hands out, and
//! the store lives **beside** it at `<repo>.gunnar-<engine>` — a sibling, not a
//! child, because the bench deletes and re-copies repository directories between
//! rounds and a store inside one would be copied along as opaque bytes.
//!
//! **`prepare` is a separate mode on purpose.** Building the store is ingest
//! work, and ingest work inside `git-upload-pack` would be charged to the clone
//! it precedes. A serve that finds no store **fails** rather than quietly
//! building one.
//!
//! # The receipt
//!
//! `GUNNAR_BENCH_COPY_STATS` gets one `{"copied":N,"recompressed":M}` line per
//! served fetch, **byte-identical** to the gix arm's — that is a copy of its
//! `writeln!` and not a re-derivation of it, because two columns written by two
//! formatters are two columns that can disagree about what they mean.
//!
//! # What this binary deliberately does NOT implement
//!
//! `GUNNAR_BENCH_NO_PACK_COPY`, the bench's red switch for verbatim pack copy. There is no
//! `NoPackCopy` shim behind `GitServe::emit_pack`, so the binary **refuses to
//! start** when it is set rather than serving an unswitched server under a
//! switched label. A switch that is silently ignored is the hollow-guard shape
//! this tree has been burned by twice.

use std::io::{BufRead, BufReader, BufWriter, Write};
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::time::Instant;

/// The heap, the same one the appliance ships.
///
/// This binary is the `GUNNAR_BENCH_GUNNAR_SERVER` seam, so an allocator
/// difference here would show up as an engine difference in the forge columns.
/// See [`gunnar_server::ServerAllocator`].
#[global_allocator]
static ALLOC: gunnar_server::ServerAllocator = gunnar_server::SERVER_ALLOCATOR;

use gunnar_server::storage::Engine;
use gunnar_store::git_storage_trait::RefUpdate;
use gunnar_wire::receive_pack::AllowAll;
use gunnar_wire::{ProtocolVersion, ReceivePack, UploadPack};

type Res<T> = Result<T, Box<dyn std::error::Error>>;

/// The store that backs `repo`. A sibling, not a child — see the module doc.
///
/// The engine's name is in the path because a build carrying both must not have
/// its two arms fight over one directory: a znippy store and a gix store at the
/// same root is two engines' layouts interleaved, and whichever opened second
/// would either refuse or — worse — succeed.
fn store_root(repo: &Path, engine: Engine) -> PathBuf {
    let mut s = repo.as_os_str().to_os_string();
    s.push(format!(".gunnar-{}", engine.name()));
    PathBuf::from(s)
}

/// The repository's hash algorithm, read from its own config.
///
/// A repository served under the wrong hash is not a degraded repository, it is
/// a different one. Same discipline as the gix arm's `hash_kind`, and the same
/// discipline `RepoStore::hash_kind` is held to in the server proper: asked,
/// never defaulted.
fn hash_kind(repo: &Path) -> gix_hash::Kind {
    let config = std::fs::read_to_string(repo.join("config")).unwrap_or_default();
    if config.to_ascii_lowercase().contains("sha256") {
        gix_hash::Kind::Sha256
    } else {
        gix_hash::Kind::Sha1
    }
}

// ─────────────────────────────────────────────────────────────── prepare ────

/// Build `<repo>.gunnar-<engine>` from `<repo>`, or confirm the one that is there.
///
/// Idempotent: the clone scenario prepares one server repository once and the
/// push scenario prepares a fresh copy every round, so a mode that only worked
/// on an empty directory would need the caller to know which case it was in.
///
/// # What is ingested, and why it is the honest input
///
/// The repository's own `objects/pack/*.pack`, byte for byte, through
/// [`Engine::ingest_push`] — the same call a real push lands in. The bench
/// fixtures are `repack -adq && prune --expire=now`-ed, so those bytes are what
/// stock `git pack-objects` wrote at git's own window and depth, and are what a
/// forge would have received.
///
/// A repository that is part packed and part loose takes the **re-pack** route
/// instead, whole: ingesting only its packs would build a store quietly missing
/// objects, and a clone that omits objects still exits 0. Which path was taken
/// is printed, because a re-pack is a different input and a ratio taken across
/// two different inputs is not a ratio. (`git repack -adq` alone is **not** a
/// sufficient fixture — measured, it leaves loose objects behind, and this is
/// the branch that then diverges.)
fn prepare(repo: &Path, engine: Engine) -> Res<()> {
    let root = store_root(repo, engine);
    let kind = hash_kind(repo);
    let started = Instant::now();
    std::fs::create_dir_all(&root)?;

    // Already built? Ask the ref set, not the directory: `open_repo` creates the
    // directory itself, so its existence says nothing about ingest.
    let store = engine.open_repo(&root, kind)?;
    let held = store.serve().refs()?.len();
    if held > 0 {
        eprintln!(
            "{engine}-prepare: {} already holds {held} refs",
            root.display()
        );
        return Ok(());
    }

    // ── objects ─────────────────────────────────────────────────────────────
    let pack_dir = repo.join("objects").join("pack");
    let mut packs: Vec<PathBuf> = Vec::new();
    if pack_dir.is_dir() {
        for entry in std::fs::read_dir(&pack_dir)? {
            let path = entry?.path();
            if path.extension().is_some_and(|e| e == "pack") {
                packs.push(path);
            }
        }
    }
    packs.sort();
    let loose = loose_object_count(repo);

    let mut pack_bytes = 0u64;
    let mut announced = 0u64;
    let ingested_from;
    if !packs.is_empty() && loose == 0 {
        // The whole point of this backend: the bytes `git repack` produced, kept
        // as they are.
        ingested_from = format!("{} verbatim pack(s)", packs.len());
        for pack in &packs {
            let file = std::fs::File::open(pack)?;
            pack_bytes += file.metadata()?.len();
            let mut reader = BufReader::new(file);
            // `want_objects: false` — nothing here gates on the pushed set, and
            // asking would wait for the indexer drain for no reader's benefit.
            // It is also the one thing the gix arm cannot answer at all.
            let receipt = engine.ingest_push(
                &root,
                kind,
                &mut reader,
                gunnar_server::storage::ObjectSet::NotNeeded,
                // No console polls an offline import; the bytes are the figure
                // this tool already prints.
                None,
                // An offline import is no transaction of a serving box: nothing to frame.
                None,
            )?;
            announced += u64::from(receipt.num_objects);
        }
    } else {
        if packs.is_empty() && loose == 0 {
            return Err(format!("{} holds no objects at all", repo.display()).into());
        }
        let why = if packs.is_empty() {
            format!("{loose} loose object(s)")
        } else {
            format!("{} pack(s) beside {loose} loose object(s)", packs.len())
        };
        ingested_from = format!("{why}, re-packed into a stream with --delta-base-offset");
        let bytes = pack_objects(repo)?;
        pack_bytes += bytes.len() as u64;
        let mut reader: &[u8] = &bytes;
        let receipt = engine.ingest_push(
            &root,
            kind,
            &mut reader,
            gunnar_server::storage::ObjectSet::NotNeeded,
            None,
            None,
        )?;
        announced += u64::from(receipt.num_objects);
    }

    // ── refs, through `git for-each-ref` ────────────────────────────────────
    //
    // Not a walk of `refs/`: packed refs are the case a hand-rolled reader
    // misses first, and git's own reader is the definition of what this
    // repository contains. Identical to the gix arm's `read_refs`, deliberately.
    let edits = read_refs(repo)?;
    if edits.is_empty() {
        return Err(format!("{} has no references under refs/", repo.display()).into());
    }
    let count = edits.len();
    let updates: Vec<RefUpdate> = edits
        .iter()
        .map(|(name, oid)| RefUpdate::set(name.clone(), oid.clone()))
        .collect();
    store.serve().put_refs(&updates)?;

    // ── HEAD, through its own accessor ──────────────────────────────────────
    //
    // `HEAD` is **not** a row in `GitOps::refs` (the `GitServe` HEAD ruling, and
    // both arms now agree on that), so it goes through `GitServe::set_head`. A
    // clone needs it: without a symbolic `HEAD` git picks the default branch by
    // guesswork.
    store.serve().set_head(&head_target(repo)?)?;

    let after = store.serve().refs()?.len();
    if after == 0 {
        return Err("the store holds no refs after ingest".into());
    }
    eprintln!(
        "{engine}-prepare: {} <- {ingested_from}, {pack_bytes} bytes, {announced} announced, \
         {count} refs, {:.1} ms",
        root.display(),
        started.elapsed().as_secs_f64() * 1000.0
    );
    Ok(())
}

/// Every reference in `repo` as `(name, hex oid)`, symrefs resolved.
fn read_refs(repo: &Path) -> Res<Vec<(String, String)>> {
    let out = Command::new("git")
        .arg("-C")
        .arg(repo)
        .args(["for-each-ref", "--format=%(refname)%00%(objectname)"])
        .stderr(Stdio::inherit())
        .output()?;
    if !out.status.success() {
        return Err(format!("git for-each-ref in {} failed", repo.display()).into());
    }
    let mut refs = Vec::new();
    for line in String::from_utf8_lossy(&out.stdout).lines() {
        let mut parts = line.split('\0');
        if let (Some(name), Some(oid)) = (parts.next(), parts.next()) {
            if !name.is_empty() && !oid.is_empty() {
                refs.push((name.to_owned(), oid.to_owned()));
            }
        }
    }
    Ok(refs)
}

/// What `<repo>`'s `HEAD` points at, in the one `&str` `set_head` takes.
///
/// A `refs/…` name for the symbolic case, a hex oid for the detached one — the
/// two shapes `GitServe::set_head` tells apart, and it refuses anything else
/// rather than guessing. An empty repository still gets `refs/heads/main`,
/// because `git init` points `HEAD` at a branch before that branch exists.
fn head_target(repo: &Path) -> Res<String> {
    let sym = Command::new("git")
        .arg("-C")
        .arg(repo)
        .args(["symbolic-ref", "--quiet", "HEAD"])
        .stderr(Stdio::null())
        .output()?;
    if sym.status.success() {
        let name = String::from_utf8_lossy(&sym.stdout).trim().to_string();
        if !name.is_empty() {
            return Ok(name);
        }
    }
    let direct = Command::new("git")
        .arg("-C")
        .arg(repo)
        .args(["rev-parse", "--verify", "--quiet", "HEAD"])
        .stderr(Stdio::null())
        .output()?;
    if direct.status.success() {
        let oid = String::from_utf8_lossy(&direct.stdout).trim().to_string();
        if !oid.is_empty() {
            return Ok(oid);
        }
    }
    Ok("refs/heads/main".to_owned())
}

/// Every reachable object in `repo`, as one packfile.
///
/// `rev-list --objects --all` then `pack-objects --stdout --delta-base-offset`,
/// git's own two step. (`pack-objects --revs` with `--all` on stdin is **not**
/// the same thing and is rejected — "fatal: not a rev '--all'" — because
/// `--revs` reads revisions, not rev-list option words. Measured on git 2.53.0.)
fn pack_objects(repo: &Path) -> Res<Vec<u8>> {
    let listed = Command::new("git")
        .arg("-C")
        .arg(repo)
        .args(["rev-list", "--objects", "--all"])
        .stderr(Stdio::inherit())
        .output()?;
    if !listed.status.success() {
        return Err(format!("git rev-list in {} failed", repo.display()).into());
    }
    let mut oids = String::new();
    for line in String::from_utf8_lossy(&listed.stdout).lines() {
        if let Some(oid) = line.split_whitespace().next() {
            oids.push_str(oid);
            oids.push('\n');
        }
    }
    // An empty list and a repository with no objects produce the same packfile,
    // and only one of them is a thing to store.
    if oids.is_empty() {
        return Err(format!("git rev-list named no objects in {}", repo.display()).into());
    }
    let mut child = Command::new("git")
        .arg("-C")
        .arg(repo)
        .args(["pack-objects", "--stdout", "--delta-base-offset", "-q"])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::inherit())
        .spawn()?;
    child
        .stdin
        .take()
        .ok_or("git pack-objects has no stdin")?
        .write_all(oids.as_bytes())?;
    let out = child.wait_with_output()?;
    if !out.status.success() {
        return Err(format!(
            "git pack-objects in {} exited {:?}",
            repo.display(),
            out.status.code()
        )
        .into());
    }
    if out.stdout.len() < 32 {
        return Err(format!(
            "git pack-objects in {} produced {} bytes, which is not a packfile",
            repo.display(),
            out.stdout.len()
        )
        .into());
    }
    Ok(out.stdout)
}

/// Loose objects under `objects/`, i.e. everything that is not a pack.
fn loose_object_count(repo: &Path) -> usize {
    let dir = repo.join("objects");
    let Ok(entries) = std::fs::read_dir(&dir) else {
        return 0;
    };
    let mut n = 0;
    for entry in entries.flatten() {
        let name = entry.file_name();
        let name = name.to_string_lossy();
        if name.len() == 2 && name.chars().all(|c| c.is_ascii_hexdigit()) {
            if let Ok(inner) = std::fs::read_dir(entry.path()) {
                n += inner.flatten().count();
            }
        }
    }
    n
}

// ────────────────────────────────────────────────────────── the receipt ────

/// Append the `copied`/`recompressed` receipt, in the **byte-for-byte** shape
/// the gix arm writes and the bench meter reads.
///
/// A differing receipt would make the two columns incomparable, which is the
/// whole point of the exercise, so this is a copy of the other arm's `writeln!`
/// and not a re-derivation of it.
fn copy_stats(copied: usize, recompressed: usize) -> Res<()> {
    if let Ok(path) = std::env::var("GUNNAR_BENCH_COPY_STATS") {
        let mut f = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(path)?;
        writeln!(
            f,
            "{{\"copied\":{},\"recompressed\":{}}}",
            copied as f64, recompressed as f64
        )?;
    }
    Ok(())
}

// ───────────────────────────────────────────────────────────────── serve ────

/// Serve one request. `service` is git's `%S`.
///
/// Deliberately nothing bench-specific: this is the `UploadPack::serve` /
/// `ReceivePack::serve` `transport.rs` calls, over the engine `transport.rs`
/// opens. The only thing missing is the policy gate, and [`AllowAll`] is named
/// rather than implied.
fn serve(service: &str, repo: &Path, engine: Engine) -> Res<()> {
    let root = store_root(repo, engine);
    let kind = hash_kind(repo);
    let store = engine.open_repo(&root, kind)?;

    // A serve against an unprepared store is a red, not a lazy build. See the
    // module doc: setup inside the measured region is the defect, not the fix.
    if store.serve().refs()?.is_empty() {
        return Err(format!(
            "{} holds no refs; run `{} prepare {}` before measuring",
            root.display(),
            std::env::args().next().unwrap_or_default(),
            repo.display()
        )
        .into());
    }

    let stdin = std::io::stdin();
    let mut input = BufReader::with_capacity(gunnar_wire::TRANSPORT_BUFFER_BYTES, stdin.lock());
    let stdout = std::io::stdout();
    let mut out = BufWriter::with_capacity(gunnar_wire::TRANSPORT_BUFFER_BYTES, stdout.lock());
    let started = Instant::now();

    match service {
        "git-upload-pack" => {
            let version =
                ProtocolVersion::from_git_protocol(std::env::var("GIT_PROTOCOL").ok().as_deref());
            let summary =
                UploadPack::new(store.serve(), kind).serve(version, &mut input, &mut out)?;
            out.flush()?;
            eprintln!(
                "{engine}: served git-upload-pack v{} rounds={} common={} selected={} \
                 objects={} copied={} recompressed={} bitmap={} {:.1} ms",
                version.number(),
                summary.fetch_rounds,
                // Printed beside the rest because it is the one number that
                // says whether negotiation *worked*: a fetch that sent the
                // right pack while acknowledging nothing common is a fetch the
                // client warns about ("no common commits") and a round-trip
                // count nobody can compare.
                summary.common,
                // What `select` chose, printed BEFORE what went out, because
                // the pair is the reading and `objects` alone is the half that
                // misleads: measured 18 selected against 157 emitted on a
                // one-commit fetch out of a packed store (the 4-core laptop, 2026-08-10).
                summary.objects_selected,
                summary.objects_sent,
                summary.objects_copied_from_pack,
                summary.objects_recompressed,
                summary.answered_from_bitmap,
                started.elapsed().as_secs_f64() * 1000.0,
            );
            // The copy split, in the file and the JSON shape the bench meter
            // already reads off the gix arm. `copied`/`recompressed` is the one
            // applied output that says *why* a pack came out the size it did.
            copy_stats(
                summary.objects_copied_from_pack,
                summary.objects_recompressed,
            )?;
        }
        "git-receive-pack" => {
            let ingest = ExampleIngest {
                git_dir: root.clone(),
                hash_kind: kind,
                engine,
            };
            let outcome = ReceivePack::new(store.serve(), kind)
                .serve(&mut input, &mut out, &ingest, &AllowAll)?;
            out.flush()?;
            match outcome {
                Some(o) => eprintln!(
                    "{engine}: served git-receive-pack commands={} statuses={:?} {:.1} ms",
                    o.commands.len(),
                    o.statuses,
                    started.elapsed().as_secs_f64() * 1000.0,
                ),
                None => eprintln!("{engine}: served git-receive-pack, no commands"),
            }
        }
        other => return Err(format!("this store serves no {other:?}").into()),
    }
    Ok(())
}

/// The push ingest, which is [`Engine::ingest_push`] and a receipt.
///
/// The same three facts `transport.rs`'s `PushIngest` fills in, and for the same
/// reasons: the object count and version come off the stored pack's own header,
/// `pack_id` is its trailing checksum, and `index_id` is a **null** oid because
/// The storage contract replaced the `.idx` with the `objects` table — there is no index file to
/// have a checksum, and this tree's spelling for that is the null oid.
///
/// `want_objects: false`, and it is [`AllowAll`] above that makes it honest: no
/// gate here reads the pushed set, so `new_objects` is an `Err` naming the
/// reason rather than an empty set a gate would pass.
struct ExampleIngest {
    git_dir: PathBuf,
    hash_kind: gix_hash::Kind,
    engine: Engine,
}

impl gunnar_wire::PackIngest for ExampleIngest {
    fn ingest(&self, pack: &mut dyn BufRead) -> gunnar_wire::Result<gunnar_wire::PackReceipt> {
        let received = self
            .engine
            .ingest_push(
                &self.git_dir,
                self.hash_kind,
                pack,
                gunnar_server::storage::ObjectSet::NotNeeded,
                None,
                None,
            )
            .map_err(|err| {
                gunnar_wire::Error::Store(gunnar_store::Error::backend(format!(
                    "storing the pushed pack: {err:#}"
                )))
            })?;
        let pack_id = gix_hash::ObjectId::try_from(received.pack_id.as_slice()).map_err(|_| {
            gunnar_wire::Error::Store(gunnar_store::Error::backend(format!(
                "the stored pack's trailer is {} bytes, which is not a {} object id",
                received.pack_id.len(),
                self.hash_kind
            )))
        })?;
        Ok(gunnar_wire::PackReceipt::without_object_set(
            gunnar_wire::pack::IndexedPack {
                num_objects: received.num_objects,
                pack_id,
                index_id: gix_hash::ObjectId::null(self.hash_kind),
                pack_version: gunnar_wire::pack::PackVersion::from_u32(received.version),
                hash_kind: self.hash_kind,
                index_path: None,
                data_path: None,
                keep_path: None,
                // No `.pack` to stat, but the store counted what it appended —
                // the same answer the real server gives (`transport.rs`).
                stored_bytes: Some(received.pack_bytes),
            },
        ))
    }
}

/// Ask the store about one oid and print the **error**, not the `false` a
/// boolean is forced to collapse it into.
///
/// A diagnostic, never part of a measurement, and it earned its place: a push
/// refused with "missing necessary objects" and a swallowed lookup error look
/// identical from the client's end.
/// It also **times the open**, which is not decoration. On the znippy arm the
/// open is where the derived tables are rebuilt (`Absorber::refold`, driven from
/// `GitStore::open_with_arms`), and that rebuild is what makes a cleanly
/// restarted store able to answer `select` at all — an empty commit graph makes
/// `select` return `Ok(None)` and `emit_pack` refuse, both correctly. So "what
/// does a cold open cost" is the number that decides whether the rebuild can
/// stay on the open path, and this is where to read it.
fn has(repo: &Path, oid: &str, engine: Engine) -> Res<()> {
    let root = store_root(repo, engine);
    let kind = hash_kind(repo);
    let opened = Instant::now();
    let store = engine.open_repo(&root, kind)?;
    let open_ms = opened.elapsed().as_secs_f64() * 1000.0;
    let refs = store.serve().refs()?.len();
    if oid.is_empty() {
        println!("engine={engine} refs={refs} open={open_ms:.1} ms");
        return Ok(());
    }
    let raw = hex_oid(oid, kind)?;
    // `select` over the one tip, because that is the call a clone makes and the
    // one that answers `Ok(None)` on a store whose projection is empty. Printed
    // as the count or as the decline, never collapsed into a boolean.
    let selected = store
        .serve()
        .select(&[&raw], &[])
        .map(|r| r.map(|s| (s.objects.len(), s.commits.len())));
    // `extents` beside `has`, because the two are not the same question and the
    // proposed `contains_batch` turned on them being. See
    // `have_batch`'s header: the gix backend's `extents` reads its pack set
    // alone while its `has` also asks the loose object database, so a loose
    // object answers `None` here and `true` there. That divergence is the
    // reason the batch seam cannot be built out of `extents` alone, and it is
    // printed rather than described so it can be seen.
    println!(
        "engine={engine} refs={refs} open={open_ms:.1} ms has({oid})={:?} header={:?} \
         extents={:?} select(objects,commits)={selected:?}",
        store.serve().has(&raw),
        store.serve().header(&raw),
        store.serve().extents(&[&raw]).map(|v| v[0].is_some()),
    );
    Ok(())
}

// ────────────────────────────────────────────────────────── havebatch ────

/// One shape's have list: `want_hits` real oids out of `pool`, the rest a
/// client's own commits.
///
/// The absent ones are well-formed oids this server has never seen, with three
/// bytes flipped so that no one-bit near-miss in an index can answer them. The
/// generator is a fixed LCG, so two runs of this binary probe the same set and
/// a before/after is a before/after.
fn have_list(pool: &[Vec<u8>], n: usize, hit_frac: f64) -> Vec<Vec<u8>> {
    let want_hits = (n as f64 * hit_frac).round() as usize;
    let mut state: u64 = 0x5EED_1234 | 1;
    let mut next = move || {
        state = state
            .wrapping_mul(6364136223846793005)
            .wrapping_add(1442695040888963407);
        state ^ (state >> 31)
    };
    let mut out: Vec<Vec<u8>> = Vec::with_capacity(n);
    for i in 0..n {
        let mut o = pool[(next() % pool.len() as u64) as usize].clone();
        if i >= want_hits {
            o[0] ^= 0xff;
            o[1] ^= (i & 0xff) as u8;
            let last = o.len() - 1;
            o[last] ^= 0x5a;
        }
        out.push(o);
    }
    out
}

/// The shapes, straight off `bench/src/have_lookup.rs`'s 18-repository sweep of
/// real `git fetch`es: git's window is **16 and doubles**, and `HaveMemo`
/// collapses a negotiation to one membership call per *distinct* have. 256 is
/// `MAX_IN_VAIN`'s ceiling, was never observed, and says so in its name.
const HAVE_SHAPES: &[(&str, usize, f64)] = &[
    ("behind16", 16, 1.0),
    ("diverge16", 16, 0.375),
    ("diverge32", 32, 0.25),
    ("diverge64", 64, 0.20),
    ("diverge128", 128, 0.162),
    ("cap256_unobs", 256, 0.0),
];

/// **The measurement**: what a batch membership test is worth at gunnar's
/// own seam, against a real repository through the real engine.
///
/// The record shows `GitOps` grew batch primitives — `extents`, `sizes`,
/// `select`, `reachable` — while the one question negotiation asks a thousand
/// times, membership, stayed scalar: `GitOps::has` takes one oid. Every claim
/// about what fixing that is worth has so far been an inference from the *store
/// side* (znippy's own `lookup_batch`, 3.9x at batch 100, the reference desktop 2026-08-07), and
/// and the standing record says in as many words that **the gunnar seam is
/// unmeasured**. This mode is the caller-side number, and it is
/// built to be capable of coming back "worth nothing".
///
/// # Three columns, and the third is the only one that could ship
///
/// | column | what it does | correct on |
/// |---|---|---|
/// | `scalar` | `has()` per oid — **gunnar as it ships** | every backend |
/// | `extents` | one `extents()`, `is_some()` per position | znippy only |
/// | `hybrid` | one `extents()`, then `has()` for the `None`s | every backend |
///
/// The middle column is **not a candidate**, and measuring it anyway is the
/// point: it is the obvious implementation and it is WRONG. The gix backend's
/// `extents` reads its pack set alone
/// (`edda/crates/storage-git-gix/src/lib.rs:809-828`) while its `has` also asks
/// the loose object database (`:766-787`), so a loose object answers `None` to
/// `extents` and `true` to `has`. On a `have` that is only a wider pack; on a
/// `want` it is a refused fetch. `hybrid` is what survives that, because a
/// `Some(extent)` means the object is in a pack and is therefore authoritative
/// in the one direction that matters — only the `None`s have to be re-asked.
/// The disagreement is **printed rather than raised**, because refusing would
/// hide the very thing the middle column exists to expose.
///
/// # Cold and warm are both reported, and neither is the other
///
/// `cold_*` is the first touch on a **freshly opened** store, before this
/// process has faulted a single index page for those oids. It is one sample by
/// definition — there is only ever one first touch — so it is printed as a raw
/// nanosecond figure and never as a ratio of medians. The estimate was 4-8x "on
/// cold pages"; a warm-only number cannot speak to that and does not pretend to.
///
/// # The probe set means what it says
///
/// Every "present" have is checked present and every "absent" have is checked
/// absent, against `has()`, **before** the clock starts, and the columns are
/// asserted to agree on the count afterwards. A fast column that answers a
/// different question is the failure this tree keeps being burned by.
///
/// # The end-to-end table is the one that decides
///
/// `saved_pct` is the **ceiling** on a batch seam here: the whole scalar-minus-
/// hybrid difference, taken against a real negotiated fetch served through the
/// same `UploadPack::serve_v2` `transport.rs` runs. It assumes the saving is
/// free to collect, which it is not. If that ceiling is a rounding error, the
/// finding is closed by measurement rather than by argument.
fn have_batch(repo: &Path, engine: Engine) -> Res<()> {
    let root = store_root(repo, engine);
    let kind = hash_kind(repo);
    let opened = Instant::now();
    let store = engine.open_repo(&root, kind)?;
    let open_ms = opened.elapsed().as_secs_f64() * 1000.0;
    let serve = store.serve();

    let refs = serve.refs()?;
    let tip = refs
        .iter()
        .filter(|r| r.name.starts_with("refs/heads/"))
        .find_map(|r| r.oid.clone())
        .or_else(|| refs.iter().find_map(|r| r.oid.clone()))
        .ok_or_else(|| format!("{} advertises no oid-valued ref", root.display()))?;

    // The present pool is the tip's own closure — real objects the store really
    // holds, not oids invented for the probe. Commits when there are enough of
    // them, because a `have` is a commit.
    let reach = serve
        .select(&[&tip], &[])?
        .ok_or("the store declined to select over its own tip")?;
    let pool: Vec<Vec<u8>> = if reach.commits.len() >= 256 {
        reach.commits.clone()
    } else {
        reach.objects.clone()
    };
    if pool.len() < 256 {
        return Err(format!(
            "the tip's closure is {} object(s); a membership probe that small measures L1",
            pool.len()
        )
        .into());
    }

    let reps: usize = std::env::var("GUNNAR_HAVEBATCH_REPS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(2001);

    println!(
        "engine={engine} store={} open={open_ms:.1}ms pool={} reps={reps}",
        root.display(),
        pool.len()
    );
    println!(
        "{:<14} {:>5} {:>10} {:>11} {:>10} {:>8} {:>8} {:>12} {:>12}",
        "shape",
        "haves",
        "ns_scalar",
        "ns_extents",
        "ns_hybrid",
        "ext_x",
        "hyb_x",
        "cold_scalar",
        "cold_hybrid"
    );

    // The scalar and hybrid rates each shape measured, carried to the
    // end-to-end table below so the share it prints is this run's own number and
    // not one copied off another box.
    let mut rate: Vec<(f64, f64)> = Vec::with_capacity(HAVE_SHAPES.len());

    for (name, n, hit_frac) in HAVE_SHAPES.iter().copied() {
        let want_hits = (n as f64 * hit_frac).round() as usize;
        let haves = have_list(&pool, n, hit_frac);
        let slices: Vec<&[u8]> = haves.iter().map(|v| v.as_slice()).collect();

        // The probe set, checked rather than assumed.
        let mut present = 0usize;
        for (i, o) in slices.iter().enumerate() {
            let here = serve.has(o)?;
            if here {
                present += 1;
            }
            if (i < want_hits) != here {
                return Err(format!(
                    "{name}: have {i} is {}, the shape says {}",
                    if here { "present" } else { "absent" },
                    if i < want_hits { "present" } else { "absent" }
                )
                .into());
            }
        }

        // ── cold: the first touch, on a store opened moments ago ────────────
        let cold_store = engine.open_repo(&root, kind)?;
        let cold = cold_store.serve();
        let t = Instant::now();
        let mut acc = 0usize;
        for o in &slices {
            if cold.has(o)? {
                acc += 1;
            }
        }
        let cold_scalar = t.elapsed().as_nanos() as f64 / n as f64;

        let cold_store2 = engine.open_repo(&root, kind)?;
        let cold2 = cold_store2.serve();
        let t = Instant::now();
        let ext = cold2.extents(&slices)?;
        let mut acc2 = 0usize;
        for (i, e) in ext.iter().enumerate() {
            if e.is_some() || cold2.has(slices[i])? {
                acc2 += 1;
            }
        }
        let cold_hybrid = t.elapsed().as_nanos() as f64 / n as f64;
        if acc != present || acc2 != present {
            return Err(format!(
                "{name}: the cold columns found {acc}/{acc2} present, the checked set says \
                 {present}"
            )
            .into());
        }

        // ── warm ────────────────────────────────────────────────────────────
        let mut ns: [f64; 3] = [0.0; 3];
        let mut found: [usize; 3] = [0; 3];
        for (col, slot) in ns.iter_mut().enumerate() {
            let mut samples: Vec<f64> = Vec::with_capacity(9);
            for rep in 0..9 {
                let t = Instant::now();
                let mut hits = 0usize;
                for _ in 0..reps {
                    match col {
                        0 => {
                            for o in &slices {
                                if serve.has(o)? {
                                    hits += 1;
                                }
                            }
                        }
                        1 => {
                            for e in serve.extents(&slices)? {
                                if e.is_some() {
                                    hits += 1;
                                }
                            }
                        }
                        _ => {
                            for (i, e) in serve.extents(&slices)?.iter().enumerate() {
                                if e.is_some() || serve.has(slices[i])? {
                                    hits += 1;
                                }
                            }
                        }
                    }
                }
                samples.push(t.elapsed().as_secs_f64() * 1e9 / (reps * n) as f64);
                if rep == 0 {
                    found[col] = hits / reps;
                }
            }
            samples.sort_by(|a, b| a.partial_cmp(b).unwrap());
            *slot = samples[0];
        }
        if found[0] != present || found[2] != present {
            return Err(format!(
                "{name}: scalar found {} and hybrid found {}, the checked set says {present}",
                found[0], found[2]
            )
            .into());
        }
        let ext_note = if found[1] == present {
            ""
        } else {
            " EXTENTS-DISAGREES"
        };

        rate.push((ns[0], ns[2]));
        println!(
            "{name:<14} {n:>5} {:>10.1} {:>11.1} {:>10.1} {:>8.3} {:>8.3} {cold_scalar:>12.1} \
             {cold_hybrid:>12.1}{ext_note}",
            ns[0],
            ns[1],
            ns[2],
            ns[0] / ns[1],
            ns[0] / ns[2],
        );
    }

    // ── end to end ──────────────────────────────────────────────────────────
    println!();
    println!(
        "{:<14} {:>5} {:>10} {:>10} {:>10} {:>9}",
        "e2e_shape", "haves", "fetch_ms", "probe_us", "saved_pct", "objects"
    );
    let want_hex = tip.iter().map(|b| format!("{b:02x}")).collect::<String>();
    let format = if kind == gix_hash::Kind::Sha256 {
        "sha256"
    } else {
        "sha1"
    };
    for (idx, (name, n, hit_frac)) in HAVE_SHAPES.iter().copied().enumerate() {
        let haves = have_list(&pool, n, hit_frac);
        // The pkt-line shape stock git sends, the same one
        // `bench/src/stages.rs::fetch_request` builds. `0001` is the delimiter
        // between the command and its arguments and is a raw four-byte marker,
        // NOT a pkt-line carrying a payload.
        let pkt = |s: &str| -> Vec<u8> {
            let mut v = format!("{:04x}", s.len() + 4).into_bytes();
            v.extend_from_slice(s.as_bytes());
            v
        };
        let mut req: Vec<u8> = Vec::new();
        req.extend(pkt("command=fetch\n"));
        req.extend(pkt(&format!("object-format={format}\n")));
        req.extend_from_slice(b"0001");
        req.extend(pkt("thin-pack\n"));
        req.extend(pkt("no-progress\n"));
        req.extend(pkt("ofs-delta\n"));
        req.extend(pkt(&format!("want {want_hex}\n")));
        for h in &haves {
            let hex = h.iter().map(|b| format!("{b:02x}")).collect::<String>();
            req.extend(pkt(&format!("have {hex}\n")));
        }
        req.extend(pkt("done\n"));
        req.extend_from_slice(b"0000");

        let mut samples: Vec<f64> = Vec::new();
        let mut objects = 0u32;
        let mut digest: u64 = 0;
        for _ in 0..5 {
            let mut input = BufReader::new(std::io::Cursor::new(req.clone()));
            let mut out: Vec<u8> = Vec::new();
            let t = Instant::now();
            let summary = UploadPack::new(serve, kind).serve_v2(&mut input, &mut out)?;
            samples.push(t.elapsed().as_secs_f64() * 1000.0);
            objects = summary.objects_sent;
            let mut d: u64 = 0xcbf2_9ce4_8422_2325;
            for byte in &out {
                d ^= u64::from(*byte);
                d = d.wrapping_mul(0x0000_0100_0000_01b3);
            }
            digest = d;
        }
        samples.sort_by(|a, b| a.partial_cmp(b).unwrap());
        let fetch_ms = samples[samples.len() / 2];
        // The CEILING on a batch seam here: the whole scalar-minus-hybrid
        // difference, taken against the fetch it would be collected inside.
        let (ns_scalar, ns_hybrid) = rate[idx];
        let probe_us = ns_scalar * n as f64 / 1e3;
        let saved_us = (ns_scalar - ns_hybrid) * n as f64 / 1e3;
        println!(
            "{name:<14} {n:>5} {fetch_ms:>10.3} {probe_us:>10.3} {:>10.4} {objects:>9}  \
             digest={digest:016x}",
            100.0 * saved_us / (fetch_ms * 1e3),
        );
    }
    Ok(())
}

fn hex_oid(oid: &str, kind: gix_hash::Kind) -> Res<Vec<u8>> {
    let raw = (0..oid.len())
        .step_by(2)
        .map(|i| u8::from_str_radix(&oid[i..i + 2], 16))
        .collect::<Result<Vec<u8>, _>>()?;
    if raw.len() != kind.len_in_bytes() {
        return Err(format!(
            "{oid} is {} bytes, and this repository's hash is {kind}",
            raw.len()
        )
        .into());
    }
    Ok(raw)
}

/// Print one op's timing as a JSON line: median/min/max in microseconds.
fn ops_emit(engine: &str, fixture: &str, op: &str, batch: usize, mut ns: Vec<u128>) {
    if ns.is_empty() {
        return;
    }
    ns.sort_unstable();
    let us = |x: u128| x as f64 / 1000.0;
    let med = us(ns[ns.len() / 2]);
    let mn = us(ns[0]);
    let mx = us(ns[ns.len() - 1]);
    println!(
        "{{\"engine\":\"{engine}\",\"fixture\":\"{fixture}\",\"op\":\"{op}\",\"reps\":{},\
         \"batch\":{batch},\"median_us\":{med:.3},\"min_us\":{mn:.3},\"max_us\":{mx:.3}}}",
        ns.len()
    );
}

/// **Per-storage-op isolation timer, on whichever engine `GUNNAR_ENGINE` selects.**
///
/// Times the `GitOps`/`GitServe` calls the wire is built out of — the batch have
/// path (`has`/`size`/`header`/`extents`), the object read (`read` = full bytes
/// vs `header` = type+size only), the ref namespace ops (`refs`/`head`/`set_head`/
/// `update_ref` single-CAS/`put_refs_cas` batch-CAS) and reachability
/// (`reachable`) — each in its own cold loop of `GUNNAR_OPS_REPS` reps, against
/// the already-`prepare`d `<repo>.gunnar-<engine>` store. One JSON line per op on
/// stdout; the engine seam is the SAME `Engine::open_repo` the server and the
/// clone matrix use, so a number here and a number there are the same engine.
///
/// Measure, never assert: every timed op asserts its result inside the loop — `head` is `Some`,
/// `has(tip)` is true, `read`'s byte length equals `header`'s size, `set_head`
/// reads back what was set, and a `update_ref` with a deliberately wrong `old`
/// is REJECTED (a silent accept is the bug this proves absent).
fn opsprobe(repo: &Path, engine: Engine) -> Res<()> {
    let root = store_root(repo, engine);
    let kind = hash_kind(repo);
    let store = engine.open_repo(&root, kind)?;
    let serve = store.serve();
    let base = serve.refs()?;
    if base.is_empty() {
        return Err(format!("{} holds no refs; run `prepare` first", root.display()).into());
    }
    let reps: usize = std::env::var("GUNNAR_OPS_REPS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(5);
    let batch: usize = std::env::var("GUNNAR_OPS_BATCH")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(128);
    let eng = engine.name();
    // A per-run nonce so the scratch refs this probe creates never collide with a
    // previous run's (create-CAS demands the name not already exist).
    let nonce = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let fixture = repo
        .file_name()
        .map(|s| s.to_string_lossy().into_owned())
        .unwrap_or_default();
    // The tip to read/probe: refs/heads/main, else the first ref carrying an oid.
    let tip: Vec<u8> = base
        .iter()
        .find(|r| r.name == "refs/heads/main")
        .and_then(|r| r.oid.clone())
        .or_else(|| base.iter().find_map(|r| r.oid.clone()))
        .ok_or("no ref carries an oid to probe")?;
    let head_target = base
        .iter()
        .find(|r| r.name == "refs/heads/main")
        .map(|_| "refs/heads/main".to_string())
        .unwrap_or_else(|| base[0].name.clone());

    // ── read ops (idempotent) ────────────────────────────────────────────────
    // refs
    let mut ns = Vec::with_capacity(reps);
    for _ in 0..reps {
        let t = Instant::now();
        let r = serve.refs()?;
        ns.push(t.elapsed().as_nanos());
        if r.len() != base.len() {
            return Err("refs count drifted mid-probe".into());
        }
    }
    ops_emit(eng, &fixture, "refs", base.len(), ns);
    // head
    let mut ns = Vec::with_capacity(reps);
    for _ in 0..reps {
        let t = Instant::now();
        let h = serve.head()?;
        ns.push(t.elapsed().as_nanos());
        if h.is_none() {
            return Err("head() returned None on a prepared store".into());
        }
    }
    ops_emit(eng, &fixture, "head", 0, ns);
    // has
    let mut ns = Vec::with_capacity(reps);
    for _ in 0..reps {
        let t = Instant::now();
        let ok = serve.has(&tip)?;
        ns.push(t.elapsed().as_nanos());
        if !ok {
            return Err("has(tip) was false for a ref this store holds".into());
        }
    }
    ops_emit(eng, &fixture, "has", 0, ns);
    // size
    let mut ns = Vec::with_capacity(reps);
    let mut sz_from_size = None;
    for _ in 0..reps {
        let t = Instant::now();
        let s = serve.size(&tip)?;
        ns.push(t.elapsed().as_nanos());
        sz_from_size = s;
        if s.is_none() {
            return Err("size(tip) was None".into());
        }
    }
    ops_emit(eng, &fixture, "size", 0, ns);
    // header (type + size, no bytes)
    let mut ns = Vec::with_capacity(reps);
    let mut sz_from_header = None;
    for _ in 0..reps {
        let t = Instant::now();
        let h = serve.header(&tip)?;
        ns.push(t.elapsed().as_nanos());
        match h {
            Some((_, sz)) => sz_from_header = Some(sz),
            None => return Err("header(tip) was None".into()),
        }
    }
    ops_emit(eng, &fixture, "header", 0, ns);
    // read (full bytes) — and cross-check its length against header's size (measured, not asserted)
    let mut ns = Vec::with_capacity(reps);
    for _ in 0..reps {
        let t = Instant::now();
        let r = serve.read(&tip)?;
        ns.push(t.elapsed().as_nanos());
        match r {
            Some((_, bytes)) => {
                if Some(bytes.len() as u64) != sz_from_header {
                    return Err(format!(
                        "read() gave {} bytes but header() said {:?}",
                        bytes.len(),
                        sz_from_header
                    )
                    .into());
                }
            }
            None => return Err("read(tip) was None".into()),
        }
    }
    ops_emit(eng, &fixture, "read", 0, ns);
    // sanity: size() and header() must agree
    if sz_from_size != sz_from_header {
        return Err(
            format!("size()={sz_from_size:?} disagrees with header()={sz_from_header:?}").into(),
        );
    }
    // extents (pack-extent lookup; may be None on gix for a loose object — tolerated)
    let mut ns = Vec::with_capacity(reps);
    for _ in 0..reps {
        let t = Instant::now();
        let _ = serve.extents(&[&tip])?;
        ns.push(t.elapsed().as_nanos());
    }
    ops_emit(eng, &fixture, "extents", 0, ns);
    // reachable (the frontier a clone of one tip walks) — tolerate an engine that
    // declines it, recording the decline rather than aborting the whole probe.
    let mut ns = Vec::with_capacity(reps);
    let mut reach_err = None;
    for _ in 0..reps {
        let t = Instant::now();
        match serve.reachable(&[&tip], &[]) {
            Ok(v) => {
                ns.push(t.elapsed().as_nanos());
                if v.is_empty() {
                    reach_err = Some("reachable(tip,[]) returned an empty frontier".to_string());
                    break;
                }
            }
            Err(e) => {
                reach_err = Some(e.to_string());
                break;
            }
        }
    }
    match reach_err {
        None => ops_emit(eng, &fixture, "reachable", 0, ns),
        Some(e) => println!(
            "{{\"engine\":\"{eng}\",\"fixture\":\"{fixture}\",\"op\":\"reachable\",\"error\":{:?}}}",
            e
        ),
    }

    // ── t/perf-style BULK arms (git `cat-file --batch` / `--batch-check`) ──────
    // git's de-facto perf standard (`t/perf/p*.sh`) times bulk object streaming,
    // not one object at a time. The single `read`/`header`/`size` arms above miss
    // that: they touch ONE oid. These two arms stream the whole reachable frontier
    // — the object set a clone of `tip` must serve — the way `cat-file --batch`
    // (full bytes) and `--batch-check` (type+size only) walk a repository. The
    // frontier is resolved ONCE, outside every timed region, so the arms measure
    // bulk access and not the reachability walk (which `reachable` already owns).
    let frontier: Vec<Vec<u8>> = serve.reachable(&[&tip], &[]).unwrap_or_default();
    if !frontier.is_empty() {
        let oids: Vec<&[u8]> = frontier.iter().map(|o| o.as_slice()).collect();
        // batch_check — bulk type+size over the whole frontier via the trait's
        // BULK `sizes()` path, which no single-op arm here exercises (the analog of
        // `git cat-file --batch-check`). Measure, never assert: the result vector must be the same
        // length as the input, carry no `None` (every reachable object must resolve
        // a size), and — the guard that catches a hollow bulk path — its total must
        // equal the sum of the per-object `size()` calls, i.e. bulk agrees with
        // singular. A `sizes()` that returned zeros or a short vector fails here.
        let mut single_total: u64 = 0;
        for o in &oids {
            single_total += serve.size(o)?.ok_or("size() None for a reachable object")?;
        }
        let mut ns = Vec::with_capacity(reps);
        for _ in 0..reps {
            let t = Instant::now();
            let szs = serve.sizes(&oids)?;
            ns.push(t.elapsed().as_nanos());
            if szs.len() != oids.len() {
                return Err(format!(
                    "sizes() returned {} entries for {} oids",
                    szs.len(),
                    oids.len()
                )
                .into());
            }
            let bulk_total: u64 = szs
                .iter()
                .map(|s| s.ok_or("sizes() had a None for a reachable object"))
                .sum::<std::result::Result<u64, _>>()?;
            if bulk_total != single_total {
                return Err(format!(
                    "bulk sizes() total {bulk_total} disagrees with per-object size() total {single_total}"
                )
                .into());
            }
        }
        ops_emit(eng, &fixture, "batch_check", frontier.len(), ns);
        // cat_file_batch — bulk FULL read of every frontier object (the analog of
        // `git cat-file --batch`). Measure, never assert: every read is `Some`, the summed byte
        // total is non-zero, and it is IDENTICAL across reps and equal to the
        // `batch_check` size total — three ways a store that silently dropped or
        // truncated an object would be caught, none of which a single-oid `read`
        // can see.
        let mut ns = Vec::with_capacity(reps);
        for _ in 0..reps {
            let mut total: u64 = 0;
            let t = Instant::now();
            for o in &oids {
                match serve.read(o)? {
                    Some((_, b)) => total += b.len() as u64,
                    None => return Err("cat_file_batch: read() None for a reachable object".into()),
                }
            }
            ns.push(t.elapsed().as_nanos());
            if total == 0 {
                return Err("cat_file_batch read 0 bytes over a non-empty frontier".into());
            }
            if total != single_total {
                return Err(format!(
                    "cat_file_batch byte total {total} disagrees with sizes() total {single_total}"
                )
                .into());
            }
        }
        ops_emit(eng, &fixture, "cat_file_batch", frontier.len(), ns);
    }

    // ── write ops (each rep uses a UNIQUE scratch ref name so CAS never fights) ─
    // set_head: write, then read back and assert (timing is the set_head only).
    let mut ns = Vec::with_capacity(reps);
    for _ in 0..reps {
        let t = Instant::now();
        serve.set_head(&head_target)?;
        ns.push(t.elapsed().as_nanos());
        let h = serve.head()?.ok_or("head() None after set_head")?;
        let ok = h.symref_target.as_deref() == Some(head_target.as_str()) || h.oid.is_some();
        if !ok {
            return Err("set_head did not read back".into());
        }
    }
    ops_emit(eng, &fixture, "set_head", 0, ns);
    // update_ref: single create-CAS (old = None = must-not-exist), unique name/rep.
    let mut ns = Vec::with_capacity(reps);
    for i in 0..reps {
        let name = format!("refs/heads/opsprobe/single/{eng}/{nonce}/{i}");
        let t = Instant::now();
        serve.update_ref(&name, None, Some(&tip))?;
        ns.push(t.elapsed().as_nanos());
    }
    ops_emit(eng, &fixture, "update_ref", 0, ns);
    // Negative CAS (measured, not asserted): a wrong `old` on an existing ref MUST be rejected.
    let mut wrong = tip.clone();
    wrong[0] ^= 0xff;
    let neg = serve.update_ref(&head_target, Some(&wrong), Some(&tip));
    if neg.is_ok() {
        return Err("update_ref accepted a compare-and-swap with a wrong old value".into());
    }
    // put_refs_cas: a batch of `batch` create-CAS in one call, unique names/rep.
    let mut ns = Vec::with_capacity(reps);
    for i in 0..reps {
        let names: Vec<String> = (0..batch)
            .map(|j| format!("refs/heads/opsprobe/batch/{eng}/{nonce}/{i}/{j}"))
            .collect();
        let edits: Vec<gunnar_store::git_storage_trait::RefCas> = names
            .iter()
            .map(|n| gunnar_store::git_storage_trait::RefCas {
                name: n.clone(),
                old: None,
                new: Some(&tip),
                peeled: None,
            })
            .collect();
        let before = serve.refs()?.len();
        let t = Instant::now();
        serve.put_refs_cas(&edits)?;
        ns.push(t.elapsed().as_nanos());
        let after = serve.refs()?.len();
        if after != before + batch {
            return Err(format!(
                "put_refs_cas of {batch} refs moved the count by {}",
                after as i64 - before as i64
            )
            .into());
        }
    }
    ops_emit(eng, &fixture, "put_refs_cas", batch, ns);
    Ok(())
}

fn main() {
    let args: Vec<String> = std::env::args().collect();
    let mode = args.get(1).cloned().unwrap_or_default();
    let repo = PathBuf::from(args.get(2).cloned().unwrap_or_default());

    // See the module doc: a red switch this binary cannot honour must stop it,
    // not be ignored by it.
    if std::env::var_os("GUNNAR_BENCH_NO_PACK_COPY").is_some() && mode != "prepare" {
        eprintln!(
            "git-store-serve: GUNNAR_BENCH_NO_PACK_COPY is set, and this server has no \
             NoPackCopy shim to honour it with. Serving anyway would report a switched run's \
             label over an unswitched server."
        );
        std::process::exit(1);
    }

    let engine = match Engine::selected() {
        Ok(e) => e,
        Err(err) => {
            eprintln!("git-store-serve: {err:#}");
            std::process::exit(1);
        }
    };

    let result = match mode.as_str() {
        "prepare" => prepare(&repo, engine),
        "opsprobe" => opsprobe(&repo, engine),
        "havebatch" => have_batch(&repo, engine),
        "has" => has(
            &repo,
            args.get(3).map(String::as_str).unwrap_or_default(),
            engine,
        ),
        "git-upload-pack" | "git-receive-pack" => serve(&mode, &repo, engine),
        other => Err(format!(
            "usage: {} prepare|opsprobe|has|havebatch|git-upload-pack|git-receive-pack <repo>  \
             (got {other:?}); \
             this binary carries {}",
            args.first()
                .map(String::as_str)
                .unwrap_or("git-store-serve"),
            Engine::compiled_list(),
        )
        .into()),
    };
    // **The third exit shape, and the one that made this a measurement.**
    //
    // The two shipped binaries reach `gunnar_server::run`, which closes the
    // stores itself; this one calls the engine directly, so it closes them
    // itself too. On the error path as well: `process::exit` runs no destructor,
    // and a bench server that leaves the database dirty poisons the *next* run
    // rather than its own — MEASURED 2026-08-10, +427 MB RSS and +107 k faults
    // on a subsequent clone that emits a byte-identical pack. See
    // `gunnar_server::storage::shutdown`.
    gunnar_server::storage::shutdown();
    if let Err(err) = result {
        eprintln!("git-store-serve: {err}");
        std::process::exit(1);
    }
}