gunnar_wire/

graph.rs

//! Walking the object graph: what a pack must contain, and whether a push
//! closes.
//!
//! One walker serves both directions (one implementation). `upload-pack` asks
//! *which objects are reachable from these wants and not from these haves* and
//! packs the answer; `receive-pack` asks *is everything reachable from these
//! new tips actually here* and throws the answer away. The difference is which
//! set is the frontier and whether a missing object is an error, so they are
//! the same function with two arguments, not two functions that must be kept
//! agreeing.
//!
//! # Honest limits
//!
//! The exclusion set is the **full** object closure of the haves, computed by
//! walking. Correct, since the client demonstrably has every object reachable
//! from a commit it has, but it costs `O(client history)` per fetch where git
//! with reachability bitmaps does `O(1)` lookups. The `git` znippy package's
//! `__gunnar_reach__` module is where that cost goes away; nothing here claims
//! it has. Sending *fewer* objects than this walk selects would be a bug;
//! sending more is only slow.
//!
//! The *memory* is the same shape and used to be worse than the time: that
//! closure is `O(repository)` object ids, and [`select_objects`] used to
//! materialise it **three times over**, once as the exclusion set, once as the
//! walk's own `seen` set, once more as a `Vec` of the same ids. Then it cloned
//! the result for step 6. See [`OidSet`].

//! # What this walker reads through, and why it is not `GitOps`
//!
//! [`Graph::load`] is the single funnel under [`Graph::peel`],
//! [`Graph::commit_links`], [`Graph::commit_facts_into`] and
//! [`Graph::tree_closure`], and every one of them needs the object
//! **inflated and delta-resolved**. `GitOps::get` deliberately does not give
//! one: `Stored::bytes` is documented as *"the pack entry, header and all,
//! byte for byte as the client sent them"* and for the ordinary push that is a
//! delta, while `Stored::obj_type` is *"the entry's type in the pack"* and may
//! be `OfsDelta`/`RefDelta`. Handing a delta to `CommitRefIter` would not
//! error; it would decode garbage.
//!
//! So the walk reads through [`GitServe`] — the **reading** contract
//! (`locked_on_target.md`'s `GitServe` reading contract) — and `GitServe::read`'s kind is the resolved
//! one. `GitOps` stays the storage contract, narrow, and no adapter in gunnar
//! re-derives an object model on top of it.

use std::collections::{BinaryHeap, HashMap, HashSet, VecDeque};

use gix_hash::{Kind, ObjectId};
use gunnar_store::git_storage_trait::{GitServe, ObjType};
use gunnar_store::object;

use crate::filter::Filter;
use crate::limits::Limits;
use crate::{Error, Result};

/// `GitServe`'s resolved object type, as the decoder names it.
///
/// **This mapping exists once, here.** `ObjType` is the contract's vocabulary
/// (six codes, four resolved and two delta) and [`object::Kind`] is the
/// decoder's (four). The two delta codes cannot come out of `GitServe::read` or
/// `GitServe::header`, which are defined to answer the *resolved* type — so a
/// delta code here is a backend that broke its own contract, and it is reported
/// as a corrupt object rather than quietly mapped onto something.
///
/// The refusal itself lives in [`object::Kind::resolved`], next to the four
/// names; what stays here is the *sentence* said about it, because only this
/// crate knows which oid was being asked for.
fn resolved_kind(id: &ObjectId, kind: ObjType) -> Result<object::Kind> {
    object::Kind::resolved(kind).ok_or_else(|| Error::CorruptObject {
        oid: *id,
        reason: format!(
            "the store answered a resolved read with {}, which is a delta entry \
             and not an object type",
            kind.as_str()
        ),
    })
}

/// **The one conversion from a storage fault into a wire error.**
///
/// `GitServe` returns `anyhow::Result` — the contract crate depends on
/// `anyhow` and nothing else so that neither backend has to resolve the
/// other's dependencies — and everything this crate can say about such a fault
/// is [`gunnar_store::Error::Backend`]. One function rather than a `map_err`
/// at each site (one implementation): the day `RefCas`/`RefLocked` (the typed-CAS question)
/// becomes a typed rejection there is exactly one place that learns about it.
/// **Nothing may tell two storage faults apart by matching on this string.**
///
/// `{:#}` and not `{}`: `anyhow`'s alternate form prints the whole cause chain.
pub(crate) fn backend(err: impl std::fmt::Display) -> Error {
    Error::Store(gunnar_store::Error::backend(format!("{err:#}")))
}

/// One object id out of the contract's vocabulary.
///
/// `git-storage-trait` speaks raw bytes — `Oid<'a> = &'a [u8]`, `Vec<u8>` in a
/// returned row — because it must name no hash library; this crate speaks
/// `ObjectId`. A length that is not a hash gunnar knows is a **backend fault,
/// not an object that happens to be absent**, so it is an error rather than a
/// skip: a selection or an advertisement silently short by the rows it could
/// not parse is the under-send that exits zero.
pub(crate) fn oid(bytes: &[u8]) -> Result<ObjectId> {
    ObjectId::try_from(bytes).map_err(|err| {
        backend(format_args!(
            "the store answered with a {}-byte object id: {err}",
            bytes.len()
        ))
    })
}

/// [`Graph::contains_have`] without a [`Graph`], for the two negotiation sites
/// that hold only the store.
///
/// One implementation, two entry points, because the *reason* a storage fault
/// reads as `false` here is subtle and must not be written down twice: a
/// client's `have` naming an object this server does not hold is the normal,
/// majority outcome, and every consumer of a negative degrades in exactly one
/// direction — the `have` drops out of `common`, so the pack excludes less and
/// is **wider and still valid**; it drops out of `Selection::client_has`, so no
/// thin delta names it as a base and the object is sent whole; `ready` may be
/// withheld a round longer. A false *positive* would be fatal — a thin pack
/// whose bases the client lacks is a failed fetch, not a smaller one — and this
/// cannot produce one.
///
/// So a backend fault is not swallowed into a *wrong* answer, only into a
/// *pessimistic* one. It is still logged, because "the object store is failing"
/// is not something a server should only be able to infer from a wide pack.
pub(crate) fn has_for_negotiation(store: &dyn GitServe, id: &gix_hash::oid) -> bool {
    match store.has(id.as_bytes()) {
        Ok(present) => present,
        Err(err) => {
            tracing::warn!(
                target: crate::receive_pack::GIT_TARGET,
                event = "git.have.store_error",
                oid = %id,
                error = %format_args!("{err:#}"),
                "the object store failed answering a `have`; treating it as absent, which \
                 widens the pack and cannot corrupt it"
            );
            false
        }
    }
}

/// How far a tag chain is followed before it is called a loop.
const MAX_TAG_DEPTH: usize = 16;

/// The default ceiling on object ids one traversal on the **serve** side may
/// hold at once.
///
/// **This is no longer what bounds gunnar's memory.** [`Limits::walk_bytes`] is
/// — in bytes, which is the unit a machine is sized in and the only unit that
/// can be pooled across requests. This number survives as an operator's
/// explicit cap on the size of a clone, reachable as `--fetch-budget`, and it
/// is charged through the same [`Graph::charge_walk`] so the two cannot drift.
///
/// # Three things the comment justifying this number said, and all three were wrong
///
/// It said *"objects visited"*. It was charged against objects **emitted** —
/// `out.len()` on the [`Emit`] each walker happened to be holding.
///
/// It said *"well past any real repository (the linux kernel is around ten
/// million objects total)"*. The kernel is **13 816 503** objects, so it sat
/// 28% *below* the repository it was chosen against. MEASURED 2026-08-05,
/// `bench/examples/walkbudget-probe.rs --scale` against a purpose-built
/// 14 042 000-object repository with no `.bitmap`: **the clone is refused**,
/// `WalkBudget`, and the same machine served it in 5 460 816 kB of peak RSS and
/// 64 seconds once the ceiling let it. That is measured now, not extrapolated.
///
/// And it was implied to be *one* ceiling. It was compared at four sites
/// against four different accumulators, so a single fetch could hold two full
/// budgets and be refused by neither, while the one repository-sized structure
/// that outlived them both — step 6's `seen` set — was compared against
/// nothing. [`Graph::charge_walk`] is the single check that replaced them.
///
/// # Where it binds, MEASURED, before and after
///
/// `bench/examples/walkbudget-probe.rs`, 2026-08-05, one process per
/// repository, binary-searching the smallest budget at which a clone still
/// completes rather than assuming the edge is the object count — the probe that
/// assumed that failed on its first repository, where a budget of 82 served
/// bifrost's 83 objects.
///
/// | corpus | objects | bound at, per-accumulator | bound at, held ids |
/// |---|---|---|---|
/// | `h2h-linear-sha1-32c` | 190 | 0.837 | 1.174 |
/// | `forge-year-8a-8b-12c` | 1 544 | 0.933 | 1.069 |
/// | `forge-year-8a-240b-12c` | 28 893 | 0.892 | 1.108 |
/// | `h2h-linear-sha1-2048c` | 36 172 | 0.943 | 1.057 |
/// | `h2h-linear-sha256-2048c` | 36 172 | 0.943 | 1.057 |
/// | `h2h-linear-sha1-20480c` | 352 396 | 0.942 | **1.058** |
///
/// The old column is the defect in one number: `budget = N` admitted between
/// 0.84N and 0.94N objects, and **which** was decided by the repository's shape
/// rather than by anything an operator could see. The new column is the same
/// ceiling meaning the same thing everywhere — 1.057, 1.057, 1.058 on the three
/// largest — and it is above 1.0 because a clone holds its peeled tips and its
/// tags as well as the pack. The two small rows are further out for the same
/// reason: a fixed handful of extra ids is a large fraction of 190.
///
/// **It used to bite only without a bitmap**, and that was the defect rather
/// than a property: a reachability index answers the clone without entering
/// this walk at all, which the same probe established by serving a whole clone
/// at a budget of ONE. Since the storage contract landed the store that answers that way — the znippy
/// cold tier — is the one the appliance runs, so an operator's `--fetch-budget`
/// was accepted and ignored on the shipping path. [`Graph::charge_answer`] is
/// where the same ceiling is now applied to an answer that arrived whole, and
/// it is the same comparison ([`Graph::charge_budget`]) so the two cannot
/// drift.
///
/// # Why the unit had to change
///
/// The thing a ceiling exists to bound is memory, and objects do not convert to
/// memory at a fixed rate. MEASURED across the same corpus, `VmHWM` per emitted
/// object: **1 572.9 B** on `forge-year-8a-240b-12c` (28 893 objects, 848 bytes
/// of content each) against **393.8 B** on the 14M-object corpus (43 bytes of
/// content each). Four times, at the same order of object count, decided by
/// what is *in* the repository. See [`WALK_BYTES_PER_OBJECT`] for the
/// two-term fit that does predict it and for what an id-counted ceiling can
/// honestly claim.
///
/// [`Limits::walk_bytes`]: crate::Limits::walk_bytes
/// [`WALK_BYTES_PER_OBJECT`]: crate::WALK_BYTES_PER_OBJECT
pub const DEFAULT_BUDGET: usize = 10_000_000;

/// The ceiling on objects visited by **`receive-pack`'s connectivity check**.
///
/// # Why this is not [`DEFAULT_BUDGET`], which would be one constant
///
/// Because they bound two different quantities, and one implementation is
/// about not writing the same thing twice rather than about giving two things
/// one name. A fetch's budget stands in front of everything the server is
/// about to *materialise*: several repository-sized id vectors and then the
/// pack itself. This walk materialises nothing — since the one-walk fix it holds
/// exactly one repository-sized [`OidSet`], the `seen` set, because
/// [`verify_connectivity`] tallies its output instead of collecting it — and
/// it runs over objects the server has **already received and indexed**, whose
/// size `gunnar_pack::IndexOptions` bounded on the way in. A push has
/// therefore already paid its admission by the time this runs; a fetch has
/// not.
///
/// Sixty-four million is 4.6x the kernel's 13 816 503 objects. At sha1 the one
/// set costs ~25 bytes per id held, so a walk that genuinely reached this
/// ceiling would hold ~1.6 GB — a bound, and an honest one for a machine that
/// is asked to accept a kernel-sized mirror push at all.
pub const CONNECTIVITY_BUDGET: usize = 64_000_000;

/// A set of object ids that a traversal grows, and that is therefore
/// `O(repository)`.
///
/// It exists for two reasons, and both are the one-set rule.
///
/// **It does not implement `Clone`, deliberately.** It exists to delete one
/// line, `let mut seen = excluded_objects.clone();`. One character of syntax
/// that doubled the peak memory of every in-flight fetch, at a size (ten
/// million ids, ~450 MB) where doubling is an out-of-memory kill rather than a
/// slowdown. A guard watching for that line is a guard somebody has to keep
/// running; a type that cannot be cloned is a compile error the day it comes
/// back. Fix by construction, one implementation.
///
/// **It is metered.** Under `cfg(test)` every id held in one of these is
/// counted, so a test can assert what the *peak* was. See [`meter`] and
/// `select_objects_holds_the_exclusion_closure_once`. Sets that are bounded by
/// what the client sent (a shallow frontier, a `have` list) are deliberately
/// left as plain [`HashSet`]s: they are not the ones that scale with the
/// repository, and metering them would blur the number that matters.
///
/// The API is the narrow slice the walkers use. Everything read-only comes from
/// `Deref`, so a `&OidSet` still passes where a `&HashSet<ObjectId>` is wanted.
#[derive(Debug, Default)]
pub(crate) struct OidSet {
    ids: HashSet<ObjectId>,
}

impl OidSet {
    pub(crate) fn new() -> Self {
        OidSet::default()
    }

    /// Add `id`, reporting whether it was new. Same contract as
    /// [`HashSet::insert`].
    pub(crate) fn insert(&mut self, id: ObjectId) -> bool {
        let fresh = self.ids.insert(id);
        if fresh {
            meter::acquire(1);
        }
        fresh
    }

    /// Hand the ids over as a plain [`HashSet`], releasing the meter.
    ///
    /// The one legitimate way out of this type. It is **not** a `Clone` in
    /// disguise: the set is moved, so nothing is duplicated and the peak does
    /// not move — which is the entire property this type exists to keep.
    pub(crate) fn into_inner(mut self) -> HashSet<ObjectId> {
        let ids = std::mem::take(&mut self.ids);
        // `Drop` releases `self.ids.len()`, which is now zero, so release the
        // real count here. Getting this wrong would leave the meter permanently
        // high and quietly break every test that reads a peak.
        meter::release(ids.len());
        ids
    }

    pub(crate) fn extend(&mut self, ids: impl IntoIterator<Item = ObjectId>) {
        let before = self.ids.len();
        self.ids.extend(ids);
        meter::acquire(self.ids.len() - before);
    }

    /// Take `id` back out, reporting whether it was there.
    ///
    /// The one caller is [`select_objects`]'s step 6, which uses the same set
    /// as its don't-send set and as the client's-closure voucher and has to
    /// separate the two again at the end. Removing is what keeps that *one*
    /// set: the alternative was a second, repository-sized copy taken before
    /// the tree walk, which is exactly the line this type was invented to make
    /// impossible.
    pub(crate) fn remove(&mut self, id: &ObjectId) -> bool {
        let gone = self.ids.remove(id);
        if gone {
            meter::release(1);
        }
        gone
    }
}

impl std::ops::Deref for OidSet {
    type Target = HashSet<ObjectId>;
    fn deref(&self) -> &HashSet<ObjectId> {
        &self.ids
    }
}

impl Drop for OidSet {
    fn drop(&mut self) {
        meter::release(self.ids.len());
    }
}

impl FromIterator<ObjectId> for OidSet {
    fn from_iter<I: IntoIterator<Item = ObjectId>>(ids: I) -> Self {
        let mut set = OidSet::new();
        set.extend(ids);
        set
    }
}

/// How many object ids the walkers are holding, and the most they ever held.
///
/// # It used to be `#[cfg(test)]`, and that is why the ceiling was wrong
///
/// This was a test instrument: in a shipping build every function was an empty
/// inline body and [`OidSet`] was a `HashSet` with a newtype around it. So the
/// one place in gunnar that knew what a traversal was **holding** could not be
/// consulted by the thing whose job is to bound it, and
/// [`DEFAULT_BUDGET`] instead charged the nearest accumulator to hand —
/// a different one at each of its four check sites, none of them the total.
/// See [`Graph::charge_walk`].
///
/// It costs a non-atomic `Cell` read-modify-write per id, on a path that has
/// just done a hash insert. Thread-local rather than atomic because a
/// traversal is single-threaded — `gunnar-cli` gives each transfer one blocking
/// thread for its whole duration — so the thread *is* the request, which also
/// gives libtest's per-test threads a per-test peak for free.
pub(crate) mod meter {
    use std::cell::Cell;

    thread_local! {
        static LIVE: Cell<usize> = const { Cell::new(0) };
        static PEAK: Cell<usize> = const { Cell::new(0) };
    }

    pub(crate) fn acquire(count: usize) {
        if count == 0 {
            return;
        }
        let live = LIVE.with(|live| {
            let now = live.get() + count;
            live.set(now);
            now
        });
        PEAK.with(|peak| {
            if live > peak.get() {
                peak.set(live);
            }
        });
    }

    pub(crate) fn release(count: usize) {
        LIVE.with(|live| live.set(live.get().saturating_sub(count)));
    }

    /// Forget the peak. Live ids are left alone; a set that outlives the reset
    /// is still held, and pretending otherwise would let a test start from a
    /// lie.
    pub(crate) fn reset_peak() {
        PEAK.with(|peak| peak.set(LIVE.with(|live| live.get())));
    }

    /// The most ids held at once since the last [`reset_peak`].
    ///
    /// Test-only, and the only part of this module that still is: the peak is
    /// what a test asserts against, while the shipping ceiling reads
    /// [`live`] — a request is refused for what it is holding now, not for what
    /// it held before something was freed.
    #[cfg(test)]
    pub(crate) fn peak() -> usize {
        PEAK.with(|peak| peak.get())
    }

    /// Ids held right now.
    pub(crate) fn live() -> usize {
        LIVE.with(|live| live.get())
    }

    /// Put `live` back to `to`, which is what a request's scope guard does when
    /// it ends.
    ///
    /// An assignment and not a subtraction, on purpose. A [`super::Selection`]
    /// outlives the walk that produced it and may be dropped on another thread
    /// — the pack writer's — and that drop releases against a counter that is
    /// not this one. Subtracting would leave every subsequent request on this
    /// thread starting from a number that drifts one way forever; assigning
    /// makes each request's charge exactly its own.
    pub(crate) fn restore(to: usize) {
        LIVE.with(|live| live.set(to));
    }
}

/// Where a tree walk puts what it finds.
///
/// Two callers want two different things out of the same walk. The fetch path
/// needs the ids, in order, because it is about to pack them. The exclusion
/// walk needs only that they were *visited*; their ids are already in its
/// `seen` set, and building a second repository-sized `Vec` of them is half of
/// One walker, two sinks (one implementation).
///
/// **Neither arm is metered, and that is deliberate rather than an omission.**
///
/// `Tally` holds nothing by construction. `Keep` holds a `Vec`, and a `Vec` has
/// no hook to release on — charging its pushes would acquire against
/// [`meter`] with nothing on the other side, so a `Graph` that served two
/// selections would bill the second for the first. That is not hypothetical:
/// it is what `a_fetch_holds_the_clients_object_closure_once` reported the
/// first time this was tried, 610 ids still charged after the set that held
/// them was gone.
///
/// It costs nothing to leave out. Every id a `Keep` walk emits was put into
/// that walk's `seen` set on the way — [`Graph::tree_closure`] pushes only what
/// `seen.insert` just accepted — and `seen` is an [`OidSet`], which is metered
/// exactly because it has the `Drop` that keeps acquire and release balanced.
/// So the emitted vector is a second copy of a subset of a set that is already
/// charged, and [`WALK_BYTES_PER_OBJECT`](crate::WALK_BYTES_PER_OBJECT) is a
/// whole-process rate measured with that vector, the writer's copy of it and
/// gix's own lists all resident.
pub(crate) enum Emit<'v> {
    /// Keep them.
    Keep(&'v mut Vec<ObjectId>),
    /// Count them and throw them away.
    Tally(usize),
}

impl Emit<'_> {
    fn push(&mut self, id: ObjectId) {
        match self {
            Emit::Keep(out) => out.push(id),
            Emit::Tally(count) => *count += 1,
        }
    }

    fn len(&self) -> usize {
        match self {
            Emit::Keep(out) => out.len(),
            Emit::Tally(count) => *count,
        }
    }
}

/// What a tree walk remembers, so that each object is visited exactly once.
///
/// # Why this is a trait and not an [`OidSet`]
///
/// `Graph::tree_closure` is the **one** tree walk in gunnar (one
/// implementation), and it now has two callers that record what it found in
/// different shapes. The walk path records object ids, because that is what it
/// hands back. The bitmap path records *bits*, because its whole point is that
/// the client's history never becomes a repository-sized set of ids — and
/// because a bit already set by some commit's stored bitmap is exactly the
/// signal to stop descending, which an id set cannot give.
///
/// Generic rather than `&mut dyn`: this is called once per object in the
/// transfer, and a virtual call per object is a cost with nothing behind it.
pub(crate) trait Seen {
    /// Record `id`, and say whether the walk should descend into it.
    ///
    /// `false` means "already accounted for" — either seen in this walk or
    /// known to be on the far side of the cut.
    fn insert(&mut self, id: &ObjectId) -> bool;
}

impl Seen for OidSet {
    fn insert(&mut self, id: &ObjectId) -> bool {
        OidSet::insert(self, *id)
    }
}

/// The don't-send set, keeping a note of every id **this transfer** added to
/// it.
///
/// [`select_objects`] uses one set for two questions that are not the same
/// question: *has the walk already accounted for this id?* and *does the client
/// hold it?* Step 6 answers the first by inserting everything it visits, which
/// makes the second one wrong for exactly the ids it inserted — and vouching
/// for one of those is the thin-pack defect [`Selection::client_has`] documents.
///
/// So the ids step 6 adds are listed as they go and taken back out at the end.
/// The list is bounded by what the transfer *touches*, not by the repository,
/// which is what keeps the one repository-sized set at one.
struct Carried<'a> {
    seen: &'a mut OidSet,
    carried: &'a mut Vec<ObjectId>,
}

impl Seen for Carried<'_> {
    fn insert(&mut self, id: &ObjectId) -> bool {
        let fresh = self.seen.insert(*id);
        if fresh {
            self.carried.push(*id);
        }
        fresh
    }
}

/// What to do about an object that is referenced but absent.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub(crate) enum Missing {
    /// Fail. Used for everything the server is about to *send*, and for the
    /// connectivity check on a push.
    #[default]
    Fail,
    /// Skip it. Used for the client's `have`s: a shallow or damaged client may
    /// name a commit whose ancestry it does not have, and that is its business,
    /// not a server error.
    Skip,
}

/// Read-side view of the object database, with the parsing gunnar needs.
///
/// **One of these is one request's traversal scope.** Constructing it snapshots
/// the thread's live-id count and resets the peak; dropping it puts the count
/// back and hands any [`WalkPool`] reservation back. See [`Graph::charge_walk`].
pub(crate) struct Graph<'a> {
    store: &'a dyn GitServe,
    hash_kind: Kind,
    /// An operator's explicit cap on objects held, when one was set.
    budget: Option<usize>,
    /// Bytes of bookkeeping this traversal may hold, from
    /// [`Limits::walk_bytes`].
    walk_bytes: Option<usize>,
    /// The server-wide purse, when the server gave this connection one.
    pool: Option<std::sync::Arc<crate::WalkPool>>,
    /// What this traversal currently holds from [`Self::pool`].
    reserved: std::cell::Cell<usize>,
    /// Ids the thread was already holding when this scope opened.
    entered_at: usize,
}

/// The traversal scope ends here, and both halves of it are released.
///
/// The [`WalkPool`](crate::WalkPool) reservation must come back or the pool
/// closes for good after enough requests, and `meter::live` must go back to
/// what it was or the *next* request on this thread starts already charged. The
/// second is the one that is easy to get wrong: a [`Selection`] outlives its
/// `Graph` and is dropped by the pack writer, possibly on another thread, so
/// this cannot be a subtraction of what the walk acquired.
impl Drop for Graph<'_> {
    fn drop(&mut self) {
        if let Some(pool) = &self.pool {
            pool.release(self.reserved.get());
        }
        meter::restore(self.entered_at);
    }
}

impl<'a> Graph<'a> {
    /// A traversal scope over `store`, for a repository whose ids are
    /// `hash_kind`.
    ///
    /// # Why the hash kind is a parameter now
    ///
    /// It was `store.hash_kind()`. [`GitServe`] has no such method: the frozen
    /// signature leaves
    /// *"whether `hash_kind()` belongs on `GitOps`"* open, because it needs a
    /// `HashKind` type in a crate that depends on `anyhow` and nothing else.
    /// Threading it in from the caller settles the question for this crate
    /// without pre-empting it for the contract — `RepoStore::hash_kind()`
    /// already exists and the server already holds one per repository, so the
    /// value is not being invented, only carried the last hop. The parsers here
    /// need it to size a parent oid, and one wrong answer silently decodes a
    /// sha256 commit as sha1, so it is a required argument rather than a
    /// defaulted one.
    pub(crate) fn new(store: &'a dyn GitServe, hash_kind: Kind) -> Self {
        let entered_at = meter::live();
        meter::reset_peak();
        Graph {
            hash_kind,
            store,
            budget: None,
            walk_bytes: None,
            pool: None,
            reserved: std::cell::Cell::new(0),
            entered_at,
        }
    }

    /// An operator's object cap, when `gunnar serve` was given one.
    ///
    /// `None` is the shipping default and is **not** an unbounded walk:
    /// [`Limits::walk_bytes`](crate::Limits::walk_bytes) is what bounds it, in
    /// the unit a machine is sized in. Leaving this at
    /// [`DEFAULT_BUDGET`] would have kept the kernel-sized clone refused
    /// however generous the byte ceiling was, which is the whole thing being
    /// fixed.
    pub(crate) fn with_budget(mut self, budget: Option<usize>) -> Self {
        self.budget = budget;
        self
    }

    /// The bytes ceiling and the shared purse this traversal serves under.
    ///
    /// Separate from [`Self::with_budget`] because they are different
    /// quantities: `budget` is an operator's explicit cap on how many objects a
    /// request may touch, and this is what the machine can hold. The push
    /// side's connectivity walk sets only the first (see
    /// [`CONNECTIVITY_BUDGET`]), because a push has already paid its admission
    /// through `gunnar_pack::IndexOptions` by the time it runs.
    pub(crate) fn with_walk_limit(
        mut self,
        walk_bytes: Option<usize>,
        pool: Option<std::sync::Arc<crate::WalkPool>>,
    ) -> Self {
        self.walk_bytes = walk_bytes;
        self.pool = pool;
        self
    }

    /// **The one place a traversal's size is judged.**
    ///
    /// # The four checks this replaces, and why four was three too many
    ///
    /// [`DEFAULT_BUDGET`] was tested at four sites and each one compared a
    /// *different* accumulator against the same number:
    /// [`Graph::commit_closure`] its own commit list,
    /// [`Graph::limited_closure`] its own, [`Graph::tree_closure`] the
    /// [`Emit`] it was handed, [`Reachability`] a visit counter. A single fetch
    /// runs the first (or second) and then the third, so it could hold two full
    /// budgets and be refused by neither — and step 5's `seen` set, which is
    /// repository-sized and outlives both, was compared against nothing at all.
    /// Under a `blob:none` filter that set grows with every blob in the
    /// repository while the `Emit` being charged grows with none of them.
    ///
    /// So the check is against [`meter::live`], the ids the thread is actually
    /// holding, minus what it was already holding when this scope opened. One
    /// number, every accumulator in it by construction (one implementation): a new
    /// repository-sized [`OidSet`] is charged the day it is written, without
    /// anyone remembering to add a fifth comparison.
    ///
    /// # Nothing is inflated to charge it
    ///
    /// The bytes come from the id count and [`WALK_BYTES_PER_OBJECT`], a
    /// measured rate. A budget that had to decompress an object to learn what to
    /// charge for it would be re-creating the defect that made `gix-odb` inflate
    /// every object just to read its compressed length, which cost 51.3% of
    /// clone CPU.
    ///
    /// [`WALK_BYTES_PER_OBJECT`]: crate::WALK_BYTES_PER_OBJECT
    fn charge_walk(&self) -> Result<()> {
        let held = meter::live().saturating_sub(self.entered_at);
        self.charge_budget(held)?;
        let bytes = held.saturating_mul(crate::WALK_BYTES_PER_OBJECT);
        if let Some(limit) = self.walk_bytes {
            if bytes > limit {
                return Err(Error::WalkTooLarge { held: bytes, limit });
            }
        }
        // The shared purse, topped up in chunks. `reserved` only ever grows
        // within one traversal: handing bytes back as a walk's `seen` set
        // shrinks would let a second transfer take the room this one is about
        // to need again, which is the shape that deadlocks a pool rather than
        // bounding it. It all comes back at `Drop`.
        if let Some(pool) = &self.pool {
            if bytes >= crate::WalkPool::FLOOR_BYTES && bytes > self.reserved.get() {
                let want = bytes.next_multiple_of(crate::WalkPool::CHUNK_BYTES);
                pool.try_reserve(want - self.reserved.get())?;
                self.reserved.set(want);
            }
        }
        Ok(())
    }

    /// `--fetch-budget`, and the **one** comparison that applies it.
    ///
    /// Written once because it now has two callers that count two different
    /// things — [`Self::charge_walk`] the ids a walker is holding,
    /// [`Self::charge_answer`] the size of an answer nobody walked for — and an
    /// operator's ceiling that meant `>=` on one path and `>` on the other
    /// would be a flag whose meaning depended on which store was underneath.
    fn charge_budget(&self, held: usize) -> Result<()> {
        if let Some(limit) = self.budget {
            if held >= limit {
                return Err(Error::WalkBudget { limit });
            }
        }
        Ok(())
    }

    /// The operator's object cap, charged against an answer that arrived
    /// **whole**.
    ///
    /// # `--fetch-budget` was accepted and ignored by every store gunnar ships
    ///
    /// [`Self::charge_walk`] reads [`meter::live`], which counts ids a *walker*
    /// is holding. On the path [`crate::reachable::select`] takes there is no
    /// walker: the store does the set algebra over its own reachability index
    /// and hands back the finished closure, so `meter::live()` is still zero
    /// when `select_objects` returns at step 0 and the budget is never
    /// compared against anything. That step is not a rare shape — since the storage
    /// contract landed the
    /// znippy cold tier is the storage engine the appliance runs, and it
    /// answers every unfiltered, non-shallow clone this way. So an operator
    /// setting `--fetch-budget 1` got a whole clone and exit 0.
    ///
    /// [`crate::reachable`] already had to make exactly this argument once, for
    /// [`crate::Limits::object_bytes`]: a ceiling that quietly stops applying
    /// on the fast path is worse than no ceiling, because every test of it
    /// stays green. This is the second one, found the same way.
    ///
    /// # Why this charges the budget and not [`crate::Limits::walk_bytes`]
    ///
    /// They are ceilings on two different things and only one of them is an
    /// operator's stated intent. `budget` is *"refuse a clone bigger than N
    /// objects"*, and N objects is exactly what the store just named — the
    /// count is exact here, where the walk can only ever approximate it.
    /// `walk_bytes` and the [`crate::WalkPool`] bound what a *traversal* holds,
    /// and this path runs no traversal; charging them here would start refusing
    /// clones the appliance serves today, on a default that is on. That is a
    /// shipping decision and it is not this function's to take — but the answer
    /// path does hold a repository-sized `Vec<ObjectId>` that nothing charges,
    /// and that gap is real and unowned.
    pub(crate) fn charge_answer(&self, objects: usize) -> Result<()> {
        self.charge_budget(objects)
    }

    /// Is this object here?
    ///
    /// # It returns a `Result` now, and that is not tidiness
    ///
    /// `ObjectStore::contains` was infallible `bool`; `GitOps::has` is
    /// `Result<bool>`. The tempting adaptation is `.unwrap_or(false)`, and at
    /// three of this method's call sites it would be a **silent under-send**:
    /// `Missing::Skip` drops an object it is told is absent, so a backend fault
    /// would quietly narrow a pack and exit zero. That is the shape a measured
    /// assertion is about. The fault propagates instead, and the one place
    /// where a negative really is the answer keeps its own method —
    /// [`Graph::contains_have`].
    pub(crate) fn contains(&self, id: &gix_hash::oid) -> Result<bool> {
        self.store.has(id.as_bytes()).map_err(backend)
    }

    /// [`Graph::contains`] for a **`have`**, where the client not having been
    /// heard of is the answer rather than a reason to go and look at the disk
    /// again.
    ///
    /// Only negotiation may call this. A `want`, a delta base and the walk's own
    /// `Missing::Skip` filter all keep [`Graph::contains`], because for them a
    /// stale `false` is a refused fetch rather than a wider pack.
    ///
    /// # What was lost, and it can only cost syscalls
    ///
    /// `ObjectStore` had two questions here, `contains` and
    /// `contains_without_rescan`, and the difference was whether a miss was
    /// allowed to trigger a re-scan of the backing store first. git draws the
    /// same line (`OBJECT_INFO_QUICK` against `HAS_OBJECT_RECHECK_PACKED`).
    /// `GitOps` has one `has`, so the distinction is gone and every question
    /// gets whatever the backend does. The **answers** are unchanged; only the
    /// syscalls a negotiation may spend re-deriving a `false` it already had
    /// are, and nothing in this crate can observe that. It is recorded here
    /// rather than dropped because it is a measured optimisation that a future
    /// `GitServe` method could restore, not a rule that stopped being true.
    pub(crate) fn contains_have(&self, id: &gix_hash::oid) -> bool {
        has_for_negotiation(self.store, id)
    }

    /// The object database underneath, for the callers that need to ask it
    /// something a graph walk cannot — a bulk `sizes`, a `select`.
    pub(crate) fn store(&self) -> &'a dyn GitServe {
        self.store
    }

    /// Load an object's content into `buf`, returning its kind.
    ///
    /// # 🔴 The serial closure walk lives through this function. Read before changing it.
    ///
    /// Every commit in the repository used to be read and decoded **twice** —
    /// once for its parents and once for the root tree the first decode had
    /// already parsed and discarded. Removing that measured **1.295x on
    /// `walk_1m`** and **1.025x on `clone_large`**, instructions retired, both
    /// sides, and the guard that holds it is `commit_reads_per_commit` in
    /// `nornir-bench`: it must read **1.000**, never 2.000. What keeps it there
    /// is that [`Graph::commit_links`] makes exactly **one** call to this
    /// function and returns the tree *and* the parents out of that one decode.
    /// Retargeting the store underneath is exactly where a second call gets
    /// added by accident.
    ///
    /// # The buffer, and what the frozen signature costs it
    ///
    /// `buf` belongs to the caller so that a walk of a million commits reuses
    /// one allocation; the allocator was **14.2%** of `gunnar.walk_1m`'s
    /// profile on 2026-08-03, which is why it is a parameter at all.
    /// `GitServe::read`, as frozen, returns an owned `Vec<u8>`, so the
    /// allocation happens inside the backend and the caller's is dropped. The
    /// move below (`*buf = bytes`) is the least-bad adaptation — it is a move,
    /// not a copy, so no object's bytes are duplicated — but it is one malloc
    /// and one free per object where there used to be neither. **This is a
    /// signature question, not a bug here**: a defaulted `read_into` on
    /// `GitServe`, costing an implementer nothing, restores the reuse. It is
    /// reported rather than invented, because the contract is frozen and
    /// changing it is a two-implementation decision.
    ///
    /// The magnitude of that regression is an **unmeasured hypothesis**: the
    /// bench target does not build and nothing has been run.
    fn load(&self, id: &ObjectId, what: &str, buf: &mut Vec<u8>) -> Result<object::Kind> {
        match self.store.read(id.as_bytes()).map_err(backend)? {
            Some((kind, bytes)) => {
                *buf = bytes;
                resolved_kind(id, kind)
            }
            None => Err(Error::missing(*id, what.to_owned())),
        }
    }

    fn corrupt(&self, id: &ObjectId, err: impl std::fmt::Display) -> Error {
        Error::CorruptObject {
            oid: *id,
            reason: err.to_string(),
        }
    }

    /// The object's kind, or `None` if it is not here.
    ///
    /// A header read, never an object read: `GitServe::header` answers kind and
    /// post-resolution size *without* the payload, which is the whole reason it
    /// is a separate method from `read`.
    pub(crate) fn kind(&self, id: &ObjectId) -> Result<Option<object::Kind>> {
        match self.store.header(id.as_bytes()).map_err(backend)? {
            Some((kind, _size)) => Ok(Some(resolved_kind(id, kind)?)),
            None => Ok(None),
        }
    }

    /// Follow a tag chain to the first non-tag object.
    ///
    /// The intermediate tag objects come back too: a `want` naming an
    /// annotated tag has to put the tag itself in the pack, not only what it
    /// points at.
    pub(crate) fn peel(&self, id: &ObjectId) -> Result<Peeled> {
        let mut tags = Vec::new();
        let mut current = *id;
        let mut buf = Vec::new();
        for depth in 0..MAX_TAG_DEPTH {
            // **The label is the first thing an operator reads.** Every read
            // here used to be called `"peeling a tag"`, including the first
            // one — so a push whose tip was simply absent came back as
            // `missing necessary objects: <oid> is not in this repository
            // (peeling a tag)` with no tag anywhere in the push, the
            // repository or the pack. bench `bare_lifecycle` refused a
            // force-push rewind with exactly that sentence and it was read as
            // a tag-peeling defect in the connectivity check; the protocol
            // half is clean under every tag shape a frontier can hold
            // (`tests/force_update.rs`), and what the sentence was actually
            // reporting was an object the store did not have.
            //
            // Only a read that follows a tag's target is tag peeling. The
            // first read is whatever the caller handed in.
            let what = if depth == 0 {
                "resolving an object"
            } else {
                "peeling a tag"
            };
            // **Kind is a HEADER question; the inflated read is only for a
            // tag's BODY.** `header` answers kind off the index
            // (`GitServe::header`) without pulling the object through the
            // exploded table. The exploded read below is genuinely needed only
            // to parse a tag's *target* out of its body — so for the terminal
            // object of a peel (a commit, in the overwhelming common case of
            // one annotated tag over a commit) `read` was pure waste: a full
            // inflated pull, and on a many-ref repository a 64 MiB Arrow batch
            // decode that thrashes `ExplodedTable::CACHE`.
            // `advertise_from` peels every annotated tag on every advertisement
            // (`ls-refs`, clone, fetch, replication), so this halves the
            // exploded reads that whole path costs — reads per advertised tag
            // fall from two (the tag *and* the commit it names) to one (the tag
            // alone). The before/after µs/ref were measured.
            let Some(kind) = self.kind(&current)? else {
                return Err(Error::missing(current, what.to_owned()));
            };
            if kind != object::Kind::Tag {
                return Ok(Peeled {
                    id: current,
                    kind,
                    tags,
                });
            }
            // It IS a tag: the body is now genuinely needed, to read the target
            // it names. `load` re-confirms the kind it just derived — the two
            // must agree (`serve.rs::header_resolves_every_delta_to_its_real_kind`).
            let loaded = self.load(&current, "peeling a tag", &mut buf)?;
            debug_assert_eq!(loaded, object::Kind::Tag);
            let target = object::tag_target(&buf, self.hash_kind)
                .map_err(|err| self.corrupt(&current, err))?;
            tags.push(current);
            current = target;
        }
        Err(self.corrupt(id, format!("tag chain deeper than {MAX_TAG_DEPTH}")))
    }

    /// A commit's tree and parents, and nothing else.
    ///
    /// [`Graph::commit`] answers a superset of this by way of
    /// `CommitRef::from_bytes`, which decodes the **whole** object: both
    /// signatures, both dates, the encoding, the extra headers and the message.
    /// The closure walks want two fields off the front of the object and never
    /// look at the rest, and `committed_at` — the one field of [`CommitFacts`]
    /// that is not here — is read only by negotiation, which asks about a
    /// handful of commits rather than about every commit in the repository.
    ///
    /// `CommitRefIter` decodes only as far as it is driven, and git's own
    /// commit grammar puts `tree` first and every `parent` next, so stopping at
    /// the first non-parent token leaves the committer, the encoding, the extra
    /// headers and the message unparsed. It does **not** make the walk
    /// signature-free: the iterator reports "no more parents" by decoding the
    /// *author*, so one signature and one date parse per commit remain. Halving
    /// that work is what is on offer, not removing it — `gix_actor` and
    /// `gix_date` together were 2.8% of `gunnar.walk_1m`'s profile on
    /// 2026-08-03.
    ///
    /// Both buffers belong to the caller and are reused across commits. This
    /// runs once per commit in the repository, and a `Vec::new()` per call is a
    /// malloc, a grow and a free per commit; the allocator was 14.2% of that
    /// same profile.
    pub(crate) fn commit_links(
        &self,
        id: &ObjectId,
        buf: &mut Vec<u8>,
        parents: &mut Vec<ObjectId>,
    ) -> Result<ObjectId> {
        let kind = self.load(id, "reading a commit", buf)?;
        if kind != object::Kind::Commit {
            return Err(self.corrupt(id, format!("expected a commit, found a {kind}")));
        }
        // `tree`, then `parent`*, then `author`: `commit_links` stops at the
        // first line that is not a parent, so the committer, the encoding, the
        // extra headers and the message are never looked at.
        object::commit_links(buf, self.hash_kind, parents).map_err(|err| self.corrupt(id, err))
    }

    /// A commit's parents, tree and committer timestamp.
    pub(crate) fn commit(&self, id: &ObjectId) -> Result<CommitFacts> {
        let mut buf = Vec::new();
        self.commit_facts(id, &mut buf)
    }

    /// [`Graph::commit`] with the caller's buffer, for a walk that reads
    /// thousands of commits.
    ///
    /// **One pass of `CommitRefIter`, stopping at the committer.** This used to
    /// be `CommitRef::from_bytes`, which decodes the *whole* object — both
    /// signatures, both dates, the encoding, every extra header and the entire
    /// message — to hand back three fields. `commit_links` above is this same
    /// reader with the date left off, and it already had to decode the author
    /// to learn that the parents were done; the committer is the very next
    /// token after that, so the date is nearly free and the message never gets
    /// looked at. One decoder rather than two (one implementation).
    ///
    /// The date is `SignatureRef::seconds`, which parses the leading integer
    /// and ignores the timezone, where the old reader ran the strict
    /// `gix_date` header parser and fell back to `0`. A well-formed commit
    /// gives the identical number either way; a malformed one now yields its
    /// seconds instead of the epoch, which is git's own leniency
    /// (`parse_commit_buffer` does the same) and is strictly better for a walk
    /// that orders by that number.
    fn commit_facts(&self, id: &ObjectId, buf: &mut Vec<u8>) -> Result<CommitFacts> {
        let mut parents = Vec::new();
        let (tree, committed_at) = self.commit_facts_into(id, buf, &mut parents)?;
        Ok(CommitFacts {
            tree,
            parents,
            committed_at,
        })
    }

    /// [`Graph::commit_facts`] into the **caller's** buffers, returning
    /// `(tree, committed_at)`.
    ///
    /// Same decoder, same single pass; the only difference is who owns the two
    /// `Vec`s. That matters wherever this is called once per commit over a
    /// list, for the reason [`Graph::commit_links`]'s own doc gives — a
    /// `Vec::new()` per call is a malloc, a grow and a free per commit, and the
    /// allocator was 14.2% of `gunnar.walk_1m`'s profile on 2026-08-03.
    ///
    /// # The measurement that put this here, **and what it did not explain**
    ///
    /// `negotiate::HaveMemo::resolve` used to call [`Graph::commit`] and read
    /// only `committed_at`, so `CommitFacts::parents` was dead and the compiler
    /// was free to delete the vector and the loop that fills it. The moment
    /// negotiation began *using* the parents (git's `THEY_HAVE` marking, see
    /// [`crate::negotiate::Negotiation::ack`]) that stopped, and the difference
    /// showed up as **+7.6 ms of server CPU** on a fetch across a 500-commit
    /// divergence with 1 008 `have`s — MEASURED 2026-08-04,
    /// `bench/examples/fetchbytes-probe.rs`, 29.0 → 36.6 ms median of five,
    /// isolated by keeping the parent read and removing only its consumer.
    ///
    /// **This entry point did not recover it, and that is recorded rather than
    /// quietly dropped.** The obvious reading of that 7.6 ms was the allocation
    /// [`Graph::commit_links`]'s doc names — a malloc, a grow and a free per
    /// commit. Hoisting both buffers out of the loop, which is exactly what
    /// this method exists to allow, moved the number by **nothing** (36.6 →
    /// 37.4 ms, inside the band). So the cost is the parent *decode*, not the
    /// allocation behind it, and what remains unexplained is why decoding
    /// ~1 000 parent lines costs 7.6 ms. It is kept because one buffer pair per
    /// negotiation is right regardless of what else is slow, not because it was
    /// measured to be worth 7.6 ms — it was measured not to be.
    ///
    /// The cost is confined to the **walk**: on the bitmap path, which is what
    /// every repacked repository serves from, the same change measured 0.0231 →
    /// 0.0217 s, i.e. no regression at all.
    fn commit_facts_into(
        &self,
        id: &ObjectId,
        buf: &mut Vec<u8>,
        parents: &mut Vec<ObjectId>,
    ) -> Result<(ObjectId, i64)> {
        let kind = self.load(id, "reading a commit", buf)?;
        if kind != object::Kind::Commit {
            return Err(self.corrupt(id, format!("expected a commit, found a {kind}")));
        }
        // `tree`, then `parent`*, then `author`, then `committer`. Everything
        // after that is the encoding, the extra headers and the message, and
        // none of it is ours to decode.
        //
        // A commit with an unparsable committer date is not a reason to fail a
        // clone; it only costs the walk its ordering for that one commit, and
        // `SLOP` is what absorbs exactly that. `commit_facts` answers `0` there
        // rather than an error, which is the same thing the
        // `gix_object::CommitRefIter` this replaced did by never yielding a
        // `Committer` token.
        object::commit_facts(buf, self.hash_kind, parents).map_err(|err| self.corrupt(id, err))
    }

    /// [`Graph::commit_facts_into`], for callers outside this module.
    pub(crate) fn commit_into(
        &self,
        id: &ObjectId,
        buf: &mut Vec<u8>,
        parents: &mut Vec<ObjectId>,
    ) -> Result<i64> {
        self.commit_facts_into(id, buf, parents).map(|(_, at)| at)
    }

    /// Commits reachable from `tips`, in breadth-first order.
    ///
    /// * `stop`: commits that are neither included nor traversed. The client's
    ///   side of the graph.
    /// * `boundary`: commits that *are* included but whose parents are not
    ///   traversed at all. A client's declared shallow frontier.
    /// * `restrict`: when set, only parents inside this set are followed.
    ///
    /// The last two are not the same thing, and conflating them is a bug worth
    /// naming: `deepen-not refs/heads/side` cuts a **merge** off from *one* of
    /// its parents, and treating that merge as a boundary drops the other
    /// parent's whole line as well. A restriction set says which parents
    /// survive; a boundary says none do.
    pub(crate) fn commit_closure(
        &self,
        tips: &[ObjectId],
        stop: &HashSet<ObjectId>,
        boundary: &HashSet<ObjectId>,
        restrict: Option<&HashSet<ObjectId>>,
        missing: Missing,
    ) -> Result<CommitClosure> {
        let mut seen = OidSet::new();
        let mut out = CommitClosure::default();
        let mut queue: VecDeque<ObjectId> = VecDeque::new();
        // Bounded by the number of distinct stopped-at parents, which is at
        // most the frontier between what the client has and what it wants —
        // not by the repository. A plain `HashSet` on purpose: `OidSet` meters
        // the repository-sized sets the rule is about, and metering this one
        // would blur the number that matters.
        let mut edges: HashSet<ObjectId> = HashSet::new();
        // Both live for the whole walk: see `commit_links`.
        let mut buf = Vec::new();
        let mut parents = Vec::new();

        for tip in tips {
            if !stop.contains(tip) && seen.insert(*tip) {
                queue.push_back(*tip);
            }
        }

        while let Some(id) = queue.pop_front() {
            self.charge_walk()?;
            if missing == Missing::Skip && !self.contains(&id)? {
                continue;
            }
            let tree = match self.commit_links(&id, &mut buf, &mut parents) {
                Ok(tree) => tree,
                Err(Error::ObjectMissing { .. }) if missing == Missing::Skip => continue,
                Err(err) => return Err(err),
            };
            // Pushed together, and nowhere else, which is what makes
            // `CommitClosure`'s pairing true by construction rather than by
            // agreement between two loops.
            out.ids.push(id);
            out.trees.push(tree);
            if boundary.contains(&id) {
                continue;
            }
            for parent in parents.drain(..) {
                if stop.contains(&parent) {
                    // An **edge**: a commit the client holds, sitting directly
                    // behind one it does not. `edges` is deduplicated because
                    // a widely-merged parent is stopped at once per child, and
                    // step 5 would otherwise walk its tree that many times.
                    if edges.insert(parent) {
                        out.edges.push(parent);
                    }
                    continue;
                }
                if restrict.is_some_and(|allowed| !allowed.contains(&parent)) {
                    continue;
                }
                if seen.insert(parent) {
                    queue.push_back(parent);
                }
            }
        }
        Ok(out)
    }

    /// The commits to send, worked out **without** first enumerating the
    /// client's whole history.
    ///
    /// This is git's `limit_list()` (`revision.c`), and it replaces the pair of
    /// [`Graph::commit_closure`] calls that [`select_objects`] used to open
    /// with. Those two are correct and they are also `O(client history)`: the
    /// first walks every commit the client holds purely to build a stop set for
    /// the second, so a client one day behind on a hundred-thousand-commit
    /// repository costs a hundred thousand commit reads to serve twenty.
    ///
    /// # How one walk does what two did
    ///
    /// Both sides go into **one queue ordered by committer date**, newest
    /// first, each commit carrying git's `UNINTERESTING` bit. Popping newest
    /// first means the two histories are drained together rather than one after
    /// the other, and the moment nothing interesting is left in the queue the
    /// walk is done: everything still queued is reachable from a `have`, so
    /// dropping it cannot drop a commit that belongs in the pack. That
    /// termination is [`still_interesting`](Limiter::still_interesting), and
    /// [`SLOP`] is the slack it gives the ordering.
    ///
    /// # Why it cannot send less
    ///
    /// The walk stops only at a point where `everybody_uninteresting` has just
    /// returned true, so every id it abandons is marked `UNINTERESTING`. A
    /// commit is marked `UNINTERESTING` only by propagation from a `have`, and
    /// a commit reachable from a `have` is one the client demonstrably holds.
    /// So an abandoned commit is a held commit, and its ancestors are held too
    /// — which is why abandoning them unread is safe. In the other direction
    /// skew beyond `SLOP` can leave a held commit *un*marked and send it again:
    /// a superset, the same trade [`Exclusion::Edges`] already makes, and the
    /// same one git makes.
    ///
    /// `haves` are taken as the peeled `have` commits, not their closure —
    /// computing that closure is the cost this exists to remove.
    pub(crate) fn limited_closure(
        &self,
        tips: &[ObjectId],
        haves: &[ObjectId],
        boundary: &HashSet<ObjectId>,
        have_boundary: &HashSet<ObjectId>,
        restrict: Option<&HashSet<ObjectId>>,
        missing: Missing,
    ) -> Result<LimitedClosure> {
        let mut walk = Limiter::default();
        let mut buf = Vec::new();

        // The `have`s first, so that a commit which is both a `want` and a
        // `have` is already uninteresting when the want reaches it — the same
        // answer the two-pass walk gives by testing every tip against its stop
        // set before queueing it.
        for id in haves {
            walk.admit(self, *id, true, Missing::Skip, &mut buf)?;
        }
        for id in tips {
            walk.admit(self, *id, false, missing, &mut buf)?;
        }

        let mut out = LimitedClosure::default();
        let mut slop = SLOP;
        // git's `date`, starting at `TIME_MAX`: the committer date of the last
        // commit this walk kept.
        let mut date = i64::MAX;
        let mut cache: Option<ObjectId> = None;

        while let Some(Pending { id, .. }) = walk.queue.pop() {
            if cache == Some(id) {
                cache = None;
            }
            self.charge_walk()?;
            let uninteresting = walk.is_uninteresting(&id);
            let facts = walk
                .parsed
                .get(&id)
                .expect("only a parsed commit is queued");
            let (tree, commit_date) = (facts.tree, facts.committed_at);
            // Cloned because `admit` takes `&mut walk`. Bounded by one commit's
            // parent count, not by the repository.
            let parents = facts.parents.clone();

            if uninteresting {
                out.uninteresting.push(id);
                // A shallow client's graft: it holds this commit and says so,
                // but not what is behind it. Walking past it here counts the
                // whole repository as already-delivered and packs nothing.
                if !have_boundary.contains(&id) {
                    for parent in &parents {
                        walk.admit(self, *parent, true, Missing::Skip, &mut buf)?;
                        if have_boundary.contains(parent) {
                            continue;
                        }
                        // git's `process_parents`: having just marked the
                        // parent, push the mark on down through whatever this
                        // walk has already read. It reads nothing itself.
                        walk.mark_parents_uninteresting(parent, have_boundary);
                    }
                }
                slop = walk.still_interesting(date, slop, &mut cache);
                if slop > 0 {
                    continue;
                }
                break;
            }

            if !boundary.contains(&id) {
                for parent in &parents {
                    if restrict.is_some_and(|allowed| !allowed.contains(parent)) {
                        continue;
                    }
                    walk.admit(self, *parent, false, missing, &mut buf)?;
                }
            }
            date = commit_date;
            out.ids.push(id);
            out.trees.push(tree);
        }

        // git's `mark_edges_uninteresting` (`list-objects.c`): the commits the
        // client holds that sit directly behind one it is being sent.
        //
        // Read off the finished walk rather than recorded inside it, because a
        // parent is not always *known* to be uninteresting at the moment its
        // child is popped — the mark can arrive later, from another branch of
        // the same walk — and an edge set collected too early would miss
        // exactly the merge parents whose trees carry the shared content. It
        // costs no reads: every parent list is already in hand.
        let mut edges: HashSet<ObjectId> = HashSet::new();
        for id in &out.ids {
            if boundary.contains(id) {
                continue;
            }
            let Some(facts) = walk.parsed.get(id) else {
                continue;
            };
            for parent in &facts.parents {
                if !walk.is_uninteresting(parent) {
                    continue;
                }
                if edges.insert(*parent) {
                    out.edges.push(*parent);
                }
            }
        }
        Ok(out)
    }

    /// Add everything under `root` to `out`, skipping anything already in
    /// `seen` or `exclude`, and honouring `filter`.
    ///
    /// `depth` starts at 1 for a commit's own root tree, which is how git
    /// numbers `tree:<depth>`: `tree:0` yields no trees at all and `tree:1`
    /// yields the root tree only.
    ///
    /// `max_blob_bytes` is the fetch path's object ceiling
    /// ([`crate::Limits::object_bytes`]) and is `None` on every walk that is not
    /// about to *send* what it finds. Only blobs are checked, and that is not a
    /// shortcut: a commit and a tree are bounded by their own grammar, a blob is
    /// bounded by nothing, and the walk already knows an entry is a blob from
    /// its tree entry's mode without reading anything.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn tree_closure<S: Seen + ?Sized>(
        &self,
        root: ObjectId,
        seen: &mut S,
        exclude: &HashSet<ObjectId>,
        filter: Option<&Filter>,
        missing: Missing,
        max_blob_bytes: Option<u64>,
        out: &mut Emit<'_>,
    ) -> Result<()> {
        let mut stack = vec![(root, 1usize)];
        let mut buf = Vec::new();
        // Hoisted out of the loop for the same reason `commit_links` takes its
        // buffers from the caller: these are two mallocs and two frees **per
        // tree in the repository**, and the allocator was 14.2% of
        // `gunnar.walk_1m`'s profile on 2026-08-03. They are cleared, not
        // reallocated, on every entry.
        let mut children: Vec<(ObjectId, usize)> = Vec::new();
        let mut blobs: Vec<ObjectId> = Vec::new();
        while let Some((id, depth)) = stack.pop() {
            self.charge_walk()?;
            if exclude.contains(&id) || !seen.insert(&id) {
                continue;
            }
            if missing == Missing::Skip && !self.contains(&id)? {
                continue;
            }
            let keep_tree = filter.is_none_or(|f| f.keeps_tree(depth));
            let kind = match self.load(&id, "reading a tree", &mut buf) {
                Ok(kind) => kind,
                Err(Error::ObjectMissing { .. }) if missing == Missing::Skip => continue,
                Err(err) => return Err(err),
            };
            if kind != object::Kind::Tree {
                return Err(self.corrupt(&id, format!("expected a tree, found a {kind}")));
            }
            if keep_tree {
                out.push(id);
            }
            let entries = object::TreeIter::from_bytes(&buf, self.hash_kind);
            children.clear();
            blobs.clear();
            for entry in entries {
                let entry = entry.map_err(|err| self.corrupt(&id, err))?;
                match entry.kind {
                    object::EntryKind::Tree => children.push((entry.oid, depth + 1)),
                    // A gitlink names a commit in *another* repository. It is
                    // not ours to have, to send, or to demand on a push.
                    object::EntryKind::Commit => {}
                    _ => blobs.push(entry.oid),
                }
            }
            for blob in blobs.drain(..) {
                // Inside the blob loop and not only at the top of the tree
                // loop: one tree's entries are bounded by nothing gunnar chose,
                // so charging per tree lets a single wide tree overshoot the
                // ceiling by its whole width. It costs a `Cell` read against a
                // header read.
                self.charge_walk()?;
                if exclude.contains(&blob) || !seen.insert(&blob) {
                    continue;
                }
                if !self.keeps_blob(&blob, filter, missing, max_blob_bytes)? {
                    continue;
                }
                out.push(blob);
            }
            stack.append(&mut children);
        }
        Ok(())
    }

    /// Does this blob go in the pack — and is gunnar willing to hold it?
    ///
    /// Two questions, **one header read**, because they are asked about the
    /// same object at the same moment and reading its header twice would be
    /// the second copy one implementation is about. A `filter` that already
    /// declines every blob short-circuits before the read: an object that is
    /// not going in the pack is not one gunnar will materialise, so its size
    /// is nobody's problem.
    ///
    /// # The ceiling costs a `header` where a `contains` used to do
    ///
    /// With `max_blob_bytes` set this asks the store for a header rather than
    /// for existence. It is a **replacement**, not an addition — the same index
    /// probe, plus the entry-header decode gix needs to answer the size — and it
    /// buys the only thing that makes the ceiling a guard rather than a report:
    /// the size is known *before* any content is read. A ceiling checked after
    /// `Find::try_find` has filled a buffer cannot stop the allocation it exists
    /// to stop.
    fn keeps_blob(
        &self,
        id: &ObjectId,
        filter: Option<&Filter>,
        missing: Missing,
        max_blob_bytes: Option<u64>,
    ) -> Result<bool> {
        // What the filter decides without looking at the object at all.
        let filter_limit = match filter {
            None => None,
            Some(filter) => match filter.blob_limit() {
                Some(limit) => Some(limit),
                None if filter.keeps_any_blob() => None,
                None => return Ok(false),
            },
        };

        if filter_limit.is_none() && max_blob_bytes.is_none() {
            if missing == Missing::Fail && !self.contains(id)? {
                return Err(Error::missing(*id, "packing a blob"));
            }
            return Ok(true);
        }

        let Some((_kind, size)) = self.store.header(id.as_bytes()).map_err(backend)? else {
            return match missing {
                Missing::Skip => Ok(false),
                Missing::Fail => Err(Error::missing(*id, "sizing a blob")),
            };
        };
        if let Some(limit) = max_blob_bytes {
            if size > limit {
                return Err(Error::ObjectTooLarge {
                    oid: *id,
                    size,
                    limit,
                });
            }
        }
        // `<`, not `<=`, because that is what git does and a partial clone is
        // judged against git.
        //
        // `list-objects-filter.c`'s `filter_blobs_limit` keeps a blob when
        // `object_length < filter_data->max_bytes`, so `blob:limit=1024` omits a
        // blob of **exactly** 1024 bytes. gunnar kept it, and the disagreement
        // is observable with two commands.
        //
        // MEASURED 2026-08-05 — one repository, three blobs of 1023, 1024
        // and 1025 bytes, `--filter=blob:limit=1024`, the same stock git 2.53.0
        // as the client on both arms:
        //
        // | server | blobs delivered |
        // |---|---|
        // | stock `git-upload-pack` | 1023 |
        // | gunnar, before | 1023, **1024** |
        // | gunnar, after | 1023 |
        //
        // One blob, so it looks small; what it costs is the invariant. A
        // partial-clone client sizes its lazy-fetch expectations from git's
        // rule, and `tests/real_git_surface.rs` had written gunnar's `<=` down
        // as the expected answer, so the suite agreed with the defect instead
        // of catching it.
        Ok(filter_limit.is_none_or(|limit| size < limit))
    }
}

/// What [`Graph::peel`] found at the end of a tag chain.
#[derive(Debug, Clone)]
pub(crate) struct Peeled {
    /// The first non-tag object.
    pub id: ObjectId,
    /// Its kind.
    pub kind: object::Kind,
    /// Every tag object walked through, outermost first.
    pub tags: Vec<ObjectId>,
}

/// The parts of a commit gunnar's traversal needs.
#[derive(Debug, Clone)]
pub(crate) struct CommitFacts {
    pub tree: ObjectId,
    pub parents: Vec<ObjectId>,
    pub committed_at: i64,
}

/// What [`Graph::commit_closure`] found: the commits, and each one's root tree.
///
/// The trees ride along because **every** caller of `commit_closure` walks each
/// commit's tree the moment it gets the list back, and asking the store for the
/// tree oid again means reading and decoding that commit a *second* time. On
/// the `fs` backend that second read is a pack-index probe, a zlib inflate and
/// a delta chain — the odb handles are built with `refresh_never()` and no
/// object cache, so nothing memoises it. `commit_links` already has the oid in
/// hand; dropping it and re-reading the object for it is the twin one
/// implementation is about, one field wide.
///
/// The price is one commit-sized `Vec<ObjectId>` beside the one that was
/// already being returned — 20 bytes per commit at sha1, 32 at sha256. It is
/// **not** metered by [`meter`], which watches the repository-sized *object*
/// sets the one-set rule is about, and it is not one of those: it is bounded by the
/// commit count, not the object count.
#[derive(Debug, Default)]
pub(crate) struct CommitClosure {
    /// The commits, in breadth-first order.
    pub ids: Vec<ObjectId>,
    /// `trees[i]` is `ids[i]`'s root tree. Same length, same order, and true by
    /// construction: the one loop that pushes either pushes both.
    pub trees: Vec<ObjectId>,
    /// Commits in `stop` that this walk ran straight into: a parent of an
    /// included commit that the client already holds.
    ///
    /// This is git's `mark_edges_uninteresting` (`list-objects.c`) edge set,
    /// and it falls out of a walk that was happening anyway — the `stop` test
    /// is already there, this only records which id failed it. See
    /// [`Exclusion::Edges`] for what it buys and what it costs.
    ///
    /// Empty unless the walk was given a non-empty `stop`, so a clone (no
    /// `have`s) never allocates.
    pub edges: Vec<ObjectId>,
}

/// What [`Graph::limited_closure`] found.
///
/// [`Self::ids`] and [`Self::trees`] are [`CommitClosure`]'s, with the same
/// pairing property and for the same reason; [`Self::edges`] is the same edge
/// set. The one field that is new is [`Self::uninteresting`], and what it is
/// *not* is the point: the commits the walk actually looked at on the client's
/// side, which is a frontier, where the two-pass walk's equivalent was the
/// client's entire history.
#[derive(Debug, Default)]
pub(crate) struct LimitedClosure {
    /// The commits to send, newest committer date first.
    pub ids: Vec<ObjectId>,
    /// `trees[i]` is `ids[i]`'s root tree.
    pub trees: Vec<ObjectId>,
    /// Client-held commits sitting directly behind a commit being sent.
    pub edges: Vec<ObjectId>,
    /// Client-held commits this walk popped. A frontier, not a history.
    pub uninteresting: Vec<ObjectId>,
}

/// git's `SLOP` (`revision.c`): how many further commits the limiting walk
/// looks at once everything left in its queue is uninteresting.
///
/// Five, git's number, for git's reason. Committer dates are wall-clock
/// readings taken on other people's machines, so a queue ordered by them can
/// hand back a commit that is *older* than its own child. This is the slack
/// that ordering gets before the walk calls the frontier closed. Beyond it the
/// walk sends objects the client already had, which is slow; it can never send
/// fewer, which is what would be broken. See [`Graph::limited_closure`].
const SLOP: usize = 5;

/// The bits git keeps on `struct object` while it limits a walk.
///
/// git's `ADDED` is deliberately absent: it guards against processing one
/// commit's parents twice, and here [`flag::SEEN`] already admits an id to the
/// queue at most once, so it could never fire.
mod flag {
    /// Queued, and therefore already read.
    pub const SEEN: u8 = 1 << 0;
    /// Reachable from something the client already has.
    pub const UNINTERESTING: u8 = 1 << 1;
    /// Referenced, absent, and tolerated. git re-reads such a commit and leans
    /// on `parse_commit_gently`'s per-object memo; gunnar's store has none, so
    /// the "we looked, it is not here" answer is remembered here instead.
    pub const ABSENT: u8 = 1 << 2;
}

/// One entry of the limiting walk's date-ordered queue.
///
/// `Ord` is the committer date first, so [`BinaryHeap::pop`] hands back the
/// newest commit — the same end git's `commit_list_insert_by_date` list is read
/// from. The insertion sequence breaks ties, reversed so the *earlier*
/// insertion wins, because git's list puts an equal date after the ones already
/// there and a heap would otherwise settle ties by whatever the object ids
/// happen to compare as. That would make the order commits are emitted in, and
/// therefore the order of objects in the pack, depend on the hash of their
/// contents.
#[derive(Debug, PartialEq, Eq)]
struct Pending {
    date: i64,
    seq: std::cmp::Reverse<u64>,
    id: ObjectId,
}

impl Ord for Pending {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        // `seq` is unique per entry, so this is a total order and never reports
        // `Equal` for two different commits.
        self.date
            .cmp(&other.date)
            .then_with(|| self.seq.cmp(&other.seq))
    }
}

impl PartialOrd for Pending {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

/// The state of one limiting walk: the fields git keeps on `struct commit` and
/// `struct object`, which gunnar has nowhere to hang.
///
/// All three maps are bounded by the commits the walk **reads**, which is the
/// frontier between the two histories rather than either of them. They are
/// plain [`HashMap`]s rather than [`OidSet`]s for the reason `commit_closure`'s
/// `edges` set is: [`meter`] watches the repository-sized sets the one-set rule is
/// about, and metering a frontier-sized one would blur the number that matters.
#[derive(Default)]
struct Limiter {
    flags: HashMap<ObjectId, u8>,
    parsed: HashMap<ObjectId, CommitFacts>,
    queue: BinaryHeap<Pending>,
    seq: u64,
}

impl Limiter {
    /// git's `repo_parse_commit_gently` followed by
    /// `commit_list_insert_by_date`: mark `id`, read it if this is the first
    /// time it has been named, and queue it.
    ///
    /// The mark is applied **before** the read and before the already-seen
    /// test, which is what lets a commit already queued as interesting be
    /// turned uninteresting by a later arrival from the client's side.
    fn admit(
        &mut self,
        graph: &Graph<'_>,
        id: ObjectId,
        uninteresting: bool,
        missing: Missing,
        buf: &mut Vec<u8>,
    ) -> Result<()> {
        {
            let flags = self.flags.entry(id).or_insert(0);
            if uninteresting {
                *flags |= flag::UNINTERESTING;
            }
            if *flags & (flag::SEEN | flag::ABSENT) != 0 {
                return Ok(());
            }
        }
        let facts = match graph.commit_facts(&id, buf) {
            Ok(facts) => facts,
            Err(Error::ObjectMissing { .. }) if missing == Missing::Skip => {
                *self.flags.entry(id).or_insert(0) |= flag::ABSENT;
                return Ok(());
            }
            Err(err) => return Err(err),
        };
        *self.flags.entry(id).or_insert(0) |= flag::SEEN;
        self.queue.push(Pending {
            date: facts.committed_at,
            seq: std::cmp::Reverse(self.seq),
            id,
        });
        self.seq += 1;
        self.parsed.insert(id, facts);
        Ok(())
    }

    /// git's `mark_parents_uninteresting` (`revision.c`).
    ///
    /// **It reads nothing.** It descends only through commits this walk has
    /// already parsed and stops at the first one it has not, which is what
    /// keeps it bounded by the frontier rather than by the history behind it —
    /// and is why the walk it belongs to can be cheap at all. It exists for the
    /// case where a commit was queued as interesting, so its parents are in
    /// hand and marked interesting too, and only afterwards turns out to be
    /// reachable from a `have`.
    fn mark_parents_uninteresting(&mut self, start: &ObjectId, have_boundary: &HashSet<ObjectId>) {
        let Some(facts) = self.parsed.get(start) else {
            return;
        };
        let mut stack: Vec<ObjectId> = facts.parents.clone();
        while let Some(mut id) = stack.pop() {
            loop {
                let flags = self.flags.entry(id).or_insert(0);
                if *flags & flag::UNINTERESTING != 0 {
                    break;
                }
                *flags |= flag::UNINTERESTING;
                if have_boundary.contains(&id) {
                    break;
                }
                let Some(facts) = self.parsed.get(&id) else {
                    break;
                };
                let Some((first, rest)) = facts.parents.split_first() else {
                    break;
                };
                stack.extend_from_slice(rest);
                id = *first;
            }
        }
    }

    fn is_uninteresting(&self, id: &ObjectId) -> bool {
        self.flags
            .get(id)
            .is_some_and(|flags| flags & flag::UNINTERESTING != 0)
    }

    /// git's `everybody_uninteresting`, with git's `interesting_cache` memo.
    ///
    /// The scan is over the whole queue. Remembering the one interesting entry
    /// it found makes every later call O(1) until that entry is popped, which
    /// matters because the caller runs on every uninteresting commit.
    fn everybody_uninteresting(&self, cache: &mut Option<ObjectId>) -> bool {
        if let Some(id) = *cache {
            if !self.is_uninteresting(&id) {
                return false;
            }
            *cache = None;
        }
        for pending in &self.queue {
            if self.is_uninteresting(&pending.id) {
                continue;
            }
            *cache = Some(pending.id);
            return false;
        }
        true
    }

    /// git's `still_interesting`. The slop left; zero means stop.
    ///
    /// `date` is the committer date of the last commit the walk kept.
    fn still_interesting(&self, date: i64, slop: usize, cache: &mut Option<ObjectId>) -> usize {
        let Some(head) = self.queue.peek() else {
            return 0;
        };
        // Something queued is at least as new as the last commit kept, so the
        // frontier is not behind the walk yet.
        if date <= head.date {
            return SLOP;
        }
        if !self.everybody_uninteresting(cache) {
            return SLOP;
        }
        slop.saturating_sub(1)
    }
}

/// The result of choosing what goes in a pack.
#[derive(Debug, Default)]
pub(crate) struct Selection {
    /// Every object to send, tags first, then commits, then trees and blobs.
    pub objects: Vec<ObjectId>,
    /// The commits included, in traversal order.
    pub commits: Vec<ObjectId>,
    /// Every object this walk proved the **client already holds**. Nothing
    /// else.
    ///
    /// This is step 6's `seen` set with the objects this transfer carries taken
    /// back out again — see [`Carried`]. the rule that there is one
    /// repository-sized set and no second copy still holds: the set is moved
    /// out rather than cloned, and the subtraction shrinks it.
    ///
    /// # The paragraph that used to be here, and the object that refuted it
    ///
    /// It read: a base `write_pack_copy_thin` cannot re-point is *either*
    /// outside the transfer — and then the client holds it — *or* inside the
    /// transfer in a different source pack, "and then naming it is an
    /// `OBJ_REF_DELTA` against an object in the same pack, which
    /// `tests/ofs_delta_off.rs` already holds stock git to. Both are sound."
    ///
    /// Stock git is sound on it. **gitoxide is not**, and gitoxide is the
    /// receiver `gunnar-client` resolves a fetch with — and every other
    /// gitoxide-based client, and gunnar's own `receive-pack`.
    /// `gix_pack::data::input::LookupRefDeltaObjectsIter` reads every
    /// `OBJ_REF_DELTA` in a thin pack as naming an object the *receiver already
    /// has*: it looks in the local object database and in the bases it has
    /// already spliced in, and **nowhere else** — never inside the pack it is
    /// reading. A base that is in the transfer and has not been spliced in is
    /// `input::Error::NotFound`, and the fetch dies with
    /// `The object <base> could not be decoded or wasn't found`.
    ///
    /// Reproduced on `gunnar.year`, 2026-08-05, and then offline from the
    /// run's own directories: tree
    /// `1d062a8c1d97fd2d04222bb9a0d5c309fde57796` went into the pack as a delta
    /// on a tree the client held; the root tree of the *next* commit went in as
    /// a delta on `1d062a8c`; gitoxide refused the whole pack while
    /// `git index-pack --strict --fix-thin` completed the same bytes. The
    /// server holds `1d062a8c` **twice** — whole in the pack whose push
    /// introduced its dependant, and as a delta in the pack whose push
    /// introduced it — so `gix_pack`'s copier cannot re-point the second delta
    /// and names its base by object id. That double storage is what
    /// `index-pack --fix-thin` leaves behind on any repository pushed to more
    /// than once and not `gc`'d, which is exactly the repository with no
    /// `.bitmap`: this path and no other.
    ///
    /// So the rule is the narrow one. **A thin delta may name a base only if
    /// the receiver already holds it.** An object this transfer is carrying is
    /// not held until the pack has been resolved, and the receiver resolves
    /// thin bases *before* it resolves the pack.
    ///
    /// # What it is **not**
    ///
    /// It is not a promise about a shallow client. A `have` from one names a
    /// commit whose ancestry that client may not hold, and steps 2 and 5
    /// deliberately tolerate those gaps rather than fail the fetch. The caller
    /// is what must not offer this as a voucher in that case; see
    /// `upload_pack::send_pack`.
    ///
    /// # Two spellings, one question
    ///
    /// The walk hands over its exclusion closure. The bitmap path has no such
    /// set and must not build one — enumerating the client's history as object
    /// ids is the `O(client history)` cost the bitmap exists to remove — so it
    /// hands over the `have` bitmap and the positions that name it. Both answer
    /// [`ClientHas::contains`] in constant time, which is the only thing the
    /// one consumer asks, and since 2026-08-05 both answer the **same**
    /// question: does the receiver hold this already.
    pub client_has: ClientHas,
    /// Whether a `.bitmap` answered this, rather than the walk below it.
    ///
    /// # Why this is a field and not a log line
    ///
    /// Because a guard needs an instrument. The bitmap path declines request
    /// shapes it cannot answer exactly, and the list is long enough that
    /// getting one entry wrong turns the whole feature off **silently**: every
    /// test still passes, because the walk still answers and the walk is
    /// correct. That is exactly what happened on 2026-08-04 — `max_blob_bytes`
    /// is `Some` by default (`Limits::object_bytes`), the bitmap path declined
    /// on it, and four end-to-end cases against stock git went green while
    /// measuring the walk. A deliberately broken reader was the only thing that
    /// showed it, and only because it left the unit tests red and the
    /// end-to-end ones green.
    pub answered_from_bitmap: bool,
    /// How many commits the bitmap path had to **open**, because they had no
    /// stored bitmap of their own.
    ///
    /// The second thing about this path no other guard can see. ORing a stored
    /// bitmap in and then walking its parents anyway is *semantically
    /// identical* — a bitmap is closed under reachability, so the extra work
    /// changes no answer — and it turns a bounded walk into a full one. Proven
    /// 2026-08-04: with the termination deliberately left out, every one of the
    /// 9 set-equality tests and all 25 end-to-end cases against stock git
    /// stayed green. Nothing but a count can fail on it.
    ///
    /// Zero for a fetch whose every want already has a bitmap, which is every
    /// fetch against a freshly repacked repository.
    pub bitmap_commits_walked: usize,
}

/// What the receiver is known **already to hold**, in whichever shape the
/// selection produced.
///
/// One question — *may a thin delta name this base?* — asked of the exclusion
/// closure the selection produced. It used to have a second arm carrying bits
/// over a git `.bitmap`'s positions; that path is gone with the format that
/// needed it, and reachability is answered by `GitServe::select` in object
/// ids. The answer is `true` only
/// for an object the receiver had **before** this transfer: see
/// [`Selection::client_has`] for the object that proved "and the ones this
/// transfer carries" is a different, wrong answer.
#[derive(Debug)]
pub(crate) enum ClientHas {
    /// The walk's exclusion closure, moved out rather than dropped.
    Ids(OidSet),
}

impl Default for ClientHas {
    fn default() -> Self {
        ClientHas::Ids(OidSet::new())
    }
}

impl ClientHas {
    /// **`contains` is gone, and it is not an oversight.**
    ///
    /// It answered *"may a thin delta name `id` as its base?"* on this side of
    /// the seam, for a pack writer that lived here. The pack writer is the
    /// engine's now, so the question is asked where the answer is used:
    /// `GitServe::emit_pack` takes the negotiated common **tips** and derives
    /// the voucher from its own reachability. Keeping a second, caller-side
    /// answer would be two authorities on which bases are safe, and the unsafe
    /// direction is a *failed* fetch rather than a larger one.
    ///
    /// **As of 2026-08-10 the znippy arm does not read `have` at all** — it
    /// closes every delta base into the pack rather than offering any as
    /// external — so the voucher is currently derived by nobody and costs bytes
    /// rather than correctness. That is recorded in that engine's source as an
    /// honest unused parameter, and it does not change the argument above:
    /// when the narrowing pass is written, `have` is where it reads from, and
    /// it must stay one authority rather than two.
    ///
    /// What survives is [`ClientHas::len`], because the server logs how much
    /// this transfer was allowed to point a delta at, and a regression in
    /// either selection path shows up as that number moving while
    /// `recompressed` climbs beside it.
    ///
    /// How many objects this vouches for.
    pub(crate) fn len(&self) -> usize {
        match self {
            ClientHas::Ids(ids) => ids.len(),
        }
    }
}

/// Which of the client's commits step 5 opens up to find objects it already
/// holds.
///
/// # The measurement this exists for
///
/// Taken 2026-08-04, one fetch of 20 commits out of 1500 against the
/// `scale-deep-1500c-256k-20d` fixture (a 377.2 MiB single pack, one
/// incompressible 256 KiB blob per commit), load average 7.70:
///
/// ```text
///                     commit reads   tree reads   pack faulted in
///   Full                    1 504        3 000     377.23 MiB  (99.96%)
///   stock git 2.53.0            —            —      10.17 MiB  ( 2.69%)
/// ```
///
/// The 0.78 MiB those 4 504 reads actually inflate is not the cost. `git
/// repack` writes a pack in *commit* order — each commit, then its trees, then
/// its blobs — so the 3 000 trees are spread evenly across all 377 MiB, 0.2
/// MiB of contiguous commits at the front and everything else interleaved with
/// a quarter-megabyte blob. Every one of those touches is an mmap fault, and
/// the desktop's `md1` carries `read_ahead_kb = 2048`, so each fault pulls in 2 MiB.
/// 3 000 touches over 189 two-megabyte windows hits every window: the whole
/// pack. The amplification is **readahead over a scattered walk**, not
/// decompression.
///
/// [`Exclusion::Full`] does what `git rev-list --objects --not` does with
/// `--objects-edge-aggressive`; [`Exclusion::Edges`] does what `pack-objects`
/// does by default.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Exclusion {
    /// Walk the trees of **every** commit the client holds.
    ///
    /// Exact: an object the client has is never sent twice. Also
    /// `O(client history)` tree reads, which is the row above.
    ///
    /// Kept, and reachable, because it is the red switch this optimisation is
    /// checked against: `GUNNAR_EXCLUSION=full` restores it, and a test that
    /// cannot restore the old behaviour cannot show that the new one changed
    /// anything.
    Full,
    /// Walk the trees only of the client's own tips and of the **edge** — the
    /// commits it holds that sit directly behind a commit it is being sent.
    ///
    /// This is `mark_edges_uninteresting()` in git's `list-objects.c`, and it
    /// is what `git upload-pack` uses for every fetch unless
    /// `--objects-edge-aggressive` is asked for.
    ///
    /// **It is a superset, not an equality, and that is the trade.** An object
    /// reachable only from deep in the client's history and *also* reachable
    /// from a commit being sent is no longer recognised as already-held, so it
    /// goes in the pack again. Sending an object the client has is slow;
    /// sending one it lacks is a broken fetch, and nothing here can do that —
    /// the *commit* stop set is still the client's full closure, so no commit,
    /// tree or blob that belongs in the pack is ever dropped. Git made the same
    /// call for the same reason.
    ///
    /// On the fixture above the superset is empty: 76 objects either way.
    #[default]
    Edges,
}

impl Exclusion {
    /// The serving default, with `GUNNAR_EXCLUSION=full` as the way back.
    ///
    /// Read every time rather than cached in a `OnceLock`, because a switch
    /// that only takes effect if it was set before the first fetch is a switch
    /// that reads green on a warm process and red on a cold one. It is one
    /// `getenv` per fetch against a walk that reads thousands of objects.
    ///
    /// Anything other than `full` is `Edges`, including a typo. That is
    /// deliberate: this is a rollback lever, and the failure mode of an
    /// unrecognised value must be *the shipping behaviour*, not a third one.
    pub(crate) fn from_env() -> Self {
        match std::env::var("GUNNAR_EXCLUSION").as_deref() {
            Ok("full") => Exclusion::Full,
            _ => Exclusion::Edges,
        }
    }
}

/// How [`select_objects`] works out which commits to send.
///
/// # The measurement this exists for
///
/// `gunnar.fetch_readside` , 2026-08-04, one fetch of 20 commits out of
/// 1500 on `scale-deep-1500c-256k-20d`. [`Exclusion::Edges`] had just taken the
/// tree reads from 3 000 to 40, and this is what was left:
///
/// ```text
///   store reads   1 546
///     of which    1 506  commits — step 3, the client's whole history
///                    40  trees   — the frontier
/// ```
///
/// 97.4% of every read a fetch still made was one walk enumerating history the
/// client had already told the server it holds, purely to build a stop set. On
/// a fixture with 1500 commits that is 1 506 reads; on a hundred-thousand-commit
/// repository it is a hundred thousand, for the same twenty commits of work.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Limiting {
    /// Two walks: the client's whole commit history, then the commits to send
    /// against that history as a stop set.
    ///
    /// Exact, `O(client history)`, and **required** by [`Exclusion::Full`],
    /// which needs the root tree of every commit the client holds and can only
    /// get it from a walk that visited every one of them. Also the rollback
    /// lever: `GUNNAR_LIMIT=two-pass`.
    TwoPass,
    /// One walk with both sides in it, ordered by committer date, stopping when
    /// the frontier is behind it. git's `limit_list`, and
    /// [`Graph::limited_closure`].
    #[default]
    Merged,
}

impl Limiting {
    /// The serving default, with `GUNNAR_LIMIT=two-pass` as the way back.
    ///
    /// Read every time, and any unrecognised value is the shipping behaviour,
    /// for the two reasons [`Exclusion::from_env`] gives.
    pub(crate) fn from_env() -> Self {
        match std::env::var("GUNNAR_LIMIT").as_deref() {
            Ok("two-pass") => Limiting::TwoPass,
            _ => Limiting::Merged,
        }
    }
}

/// Whether the serve path may answer from a `.bitmap`.
///
/// The rollback lever, in the same shape as [`Exclusion`] and [`Limiting`]:
/// `GUNNAR_BITMAP=off` puts every request back on the walk, which is what every
/// request did before 2026-08-04. It is also how the tests get both answers out
/// of one server to compare, so the lever is exercised by the suite rather than
/// only by an operator having a bad day.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Bitmaps {
    /// Read the `.bitmap` when the store has one and the request shape allows.
    #[default]
    Read,
    /// Walk, always.
    Off,
}

impl Bitmaps {
    /// The serving default, with `GUNNAR_BITMAP=off` as the way back.
    ///
    /// Read every time, and any unrecognised value is the shipping behaviour,
    /// for the two reasons [`Exclusion::from_env`] gives.
    pub(crate) fn from_env() -> Self {
        match std::env::var("GUNNAR_BITMAP").as_deref() {
            Ok("off") => Bitmaps::Off,
            _ => Bitmaps::Read,
        }
    }
}

/// Options for [`select_objects`].
#[derive(Debug, Default, Clone)]
pub(crate) struct SelectOptions<'f> {
    /// The client's shallow frontier: included, but not traversed past.
    pub boundary: HashSet<ObjectId>,
    /// Where the client's *own* history stops, bounding step 3's walk over the
    /// `excludes`.
    ///
    /// Equal to `boundary` for every request that is not a deepening, and
    /// deliberately not folded into it for the ones that are: a deepening
    /// walks past the client's cut on purpose, while the walk that decides
    /// what the client already holds never may. See
    /// [`crate::shallow::ShallowOutcome::client_boundary`].
    pub have_boundary: HashSet<ObjectId>,
    /// A partial-clone filter, if the client sent one.
    pub filter: Option<&'f Filter>,
    /// When set, the commit walk may only visit commits in this set, which is
    /// the answer a deepening request computed.
    pub restrict: Option<&'f HashSet<ObjectId>>,
    /// Whether a referenced-but-absent object is an error.
    pub missing: Missing,
    /// How much of the client's history step 5 opens up. See [`Exclusion`].
    pub exclusion: Exclusion,
    /// How much of the client's history steps 3 and 4 walk. See [`Limiting`].
    ///
    /// [`Exclusion::Full`] forces [`Limiting::TwoPass`] whatever this says,
    /// because the full exclusion needs the root tree of every commit the
    /// client holds and only the two-pass walk ever visits them. The two knobs
    /// are separate all the same: they name different halves of the same
    /// request, and folding them into one would make the exactness lever and
    /// the traversal lever impossible to roll back independently.
    pub limiting: Limiting,
    /// The ceiling on a single blob this walk may select — the fetch path's
    /// [`crate::Limits::object_bytes`].
    ///
    /// **`None` by default, and that is not the exception to "a guard that is
    /// off by default is not a guard".** This struct is crate-private and has
    /// three callers. Only one of them is about to *materialise* what it
    /// selects, and only that one may set the field; the exclusion walk and
    /// `WantPolicy`'s reachability closure decide membership and read no
    /// content, so a ceiling there would refuse fetches over an object nobody
    /// was going to hold. The default that is on lives where the server is
    /// configured — [`crate::upload_pack::Config::limits`].
    pub max_blob_bytes: Option<u64>,
    /// Whether this selection may be answered from a `.bitmap`. See
    /// [`Bitmaps`], and [`crate::bitmap::select`] for what it does and for the
    /// request shapes it declines.
    pub bitmaps: Bitmaps,
}

impl SelectOptions<'_> {
    /// The options a **fetch is actually served with**, before the
    /// request-specific fields are filled in.
    ///
    /// One definition, used by `upload_pack::send_pack` and by the tests (one
    /// implementation). Not a convenience: a test that builds these by hand is
    /// a test of a configuration the server never runs in, and the ceiling
    /// below is the proof — `Limits::object_bytes` is `Some` by default,
    /// `..Default::default()` makes it `None`, and every bitmap test written
    /// that way was measuring
    /// the walk.
    pub(crate) fn for_serving(limits: &Limits) -> Self {
        SelectOptions {
            missing: Missing::Fail,
            exclusion: Exclusion::from_env(),
            limiting: Limiting::from_env(),
            bitmaps: Bitmaps::from_env(),
            max_blob_bytes: limits.object_bytes.map(|n| n as u64),
            ..Default::default()
        }
    }

    /// The options a **mirror push** is built with.
    ///
    /// [`Self::for_serving`] with the client-facing blob ceiling left off, and
    /// for one reason: a mirror is not a client. `Limits::object_bytes` refuses
    /// to *hand out* an object bigger than an operator allows a fetch to pull;
    /// applying it here would mean a repository holding one large blob could
    /// never obtain a second copy at all, which is the opposite of what the
    /// ceiling is for.
    ///
    /// The three rollback levers — `GUNNAR_EXCLUSION`, `GUNNAR_LIMIT`,
    /// `GUNNAR_BITMAP` — are read exactly as the serve path reads them, so an
    /// operator putting the server back on the walk puts replication back on it
    /// too rather than leaving one of the two paths on a setting nobody chose.
    pub(crate) fn for_mirroring() -> Self {
        SelectOptions {
            missing: Missing::Fail,
            exclusion: Exclusion::from_env(),
            limiting: Limiting::from_env(),
            bitmaps: Bitmaps::from_env(),
            max_blob_bytes: None,
            ..Default::default()
        }
    }
}

/// **The objects one push must carry**: everything reachable from `want` that
/// is not reachable from `have`, in this store's ids.
///
/// # Why replication comes through here rather than calling the store
///
/// 🔴 MEASURED on the reference appliance 2026-08-31. A resync of 25 repositories left
/// six failing, every one of them on an annotated tag, every one refused by the
/// standby's own `receive-pack` with
/// `missing necessary objects: <oid> is not in this repository (peeling a
/// tag)` — where `<oid>` is the commit the tag names and the primary holds.
///
/// The mirror built its object list from
/// [`GitOps::reachable`](gunnar_store::git_storage_trait::GitOps::reachable),
/// handing it the ref tips **unpeeled**. For an annotated tag that tip is the
/// tag *object*, and the shipped storage engine contributes a non-commit tip's
/// own id and stops: the tag went into the pack and the commit under it did
/// not. It only showed where the tagged commit was on no branch — `skade`'s
/// `v0.9.10`, `edda`'s twenty-five tags on `2c7f9919…` — because otherwise a
/// branch tip in the same push carried it by accident.
///
/// A clone of those same repositories has always worked, and this function is
/// why: [`select_objects`] peels its tips (step 1) and puts the tag objects it
/// walked through into the answer, and so does the store-answered fast path in
/// [`crate::reachable::select`]. Replication asking the store directly was a
/// second definition of *"what does this ref tip reach"*, and the two
/// definitions disagreed about exactly one edge. There is one now (one implementation).
///
/// `have` is tolerant by construction: a tip the far end advertised and this
/// store does not hold contributes nothing rather than failing, which is what a
/// standby whose history diverged looks like. A `want` this store does not hold
/// is an error, because that pack cannot be built.
///
/// # No object budget and no byte ceiling, deliberately
///
/// [`Graph::with_budget`] and [`Graph::with_walk_limit`] are what an operator
/// aims at a *client*: a fetch that would cost more than this machine will
/// spend is refused, and the client is told to ask for less. A mirror cannot
/// ask for less — there is no smaller request that still leaves the standby
/// holding the repository — so a ceiling here would not bound the work, it
/// would only make one repository permanently unreplicable. That is what
/// `GitOps::reachable` did before this function existed, over the same objects,
/// so this changes nothing about what the replicator may consume.
pub fn objects_to_send(
    store: &dyn GitServe,
    hash_kind: Kind,
    want: &[&[u8]],
    have: &[&[u8]],
) -> Result<Vec<Vec<u8>>> {
    let tips: Vec<ObjectId> = want.iter().map(|raw| oid(raw)).collect::<Result<_>>()?;
    let excludes: Vec<ObjectId> = have.iter().map(|raw| oid(raw)).collect::<Result<_>>()?;
    let graph = Graph::new(store, hash_kind);
    let selection = select_objects(&graph, &tips, &excludes, &SelectOptions::for_mirroring())?;
    Ok(selection
        .objects
        .into_iter()
        .map(|id| id.as_slice().to_vec())
        .collect())
}

/// Everything reachable from `tips` that is not reachable from `excludes`.
///
/// This is `git rev-list --objects <tips> --not <excludes>`. Three callers use
/// it: to build a fetch's pack, to check that a push's new tips close, and,
/// with an empty `excludes`, to enumerate a whole repository for a clone.
pub(crate) fn select_objects(
    graph: &Graph<'_>,
    tips: &[ObjectId],
    excludes: &[ObjectId],
    options: &SelectOptions<'_>,
) -> Result<Selection> {
    // ── 0. the bitmap, when there is one and it can answer this ─────────────
    //
    //    Before anything else, because everything after it is the walk this
    //    replaces. It declines rather than approximates: a request shape the
    //    bitmap cannot answer exactly — a filter, a shallow boundary, a
    //    deepening, an object ceiling — falls through to the code below and is
    //    served exactly as it was before this existed. See [`crate::bitmap`].
    if options.bitmaps == Bitmaps::Read {
        let _s = crate::stage::enter("select.bitmap");
        // ── the escape hatch moved, and it is still the whole design ────────
        //
        // This used to be two questions: `ObjectStore::reachable()`, an
        // `Option<Arc<dyn ReachIndex>>` answering *"do you have a reachability
        // index at all"*, and then `ReachIndex::select`, answering *"can you
        // answer THIS request exactly"*. `GitServe` has one method and it is
        // the second one — `select` is required, and `Ok(None)` is how a
        // backend declines. A backend with no index answers `Ok(None)` always.
        //
        // **`Ok(None)` must stay reachable.** The Arrow arm returns it when its
        // reachability projection is empty, which a clean restart produces
        // today; without it a backend whose index does not cover a tip would
        // contribute only the tip and the clone would be short by everything
        // beneath it — while exiting zero. Losing the option type is how that
        // becomes unrepresentable.
        if let Some(selection) = crate::reachable::select(graph, tips, excludes, options)? {
            // ── the budget applies to the fast path too ────────────────────
            //
            // It did not, and that made `--fetch-budget` a control that does
            // nothing on the shipped engine: the znippy arm answers `select`
            // out of its reachability index, so the walk below — the only place
            // `charge_walk` runs — never happens, and a budget of **1** served
            // a whole clone. `tests/wire_config.rs` saw it as *"a budget of 1
            // served a clone; the traversal budget never reached the walk"*.
            //
            // "Objects visited" is the walk's quantity; the index's is
            // "objects selected", and it is the honest analogue — it is what
            // the server is about to materialise, which is what the budget's
            // own doc says it stands in front of. So the ceiling binds on both
            // paths and means the same thing on both: **this fetch is bigger
            // than you allowed.**
            //
            // Deliberately `>` and not `>=`: a budget of N must serve a fetch
            // of exactly N objects. The walk's check is `>=` because it fires
            // *before* visiting the next object rather than after counting the
            // last one; both refuse the same N+1 request.
            if let Some(limit) = graph.budget {
                if selection.objects.len() > limit {
                    return Err(Error::WalkBudget { limit });
                }
            }
            return Ok(selection);
        }
    }

    let mut selection = Selection::default();
    let mut emitted = OidSet::new();
    let _peel = crate::stage::enter("select.peel");

    // 1. Peel the tips. A `want` may name a tag, a tree, or even a blob.
    let mut commit_tips = Vec::new();
    let mut direct: Vec<(ObjectId, object::Kind)> = Vec::new();
    for tip in tips {
        let peeled = graph.peel(tip)?;
        for tag in peeled.tags {
            if emitted.insert(tag) {
                selection.objects.push(tag);
            }
        }
        match peeled.kind {
            object::Kind::Commit => commit_tips.push(peeled.id),
            other => direct.push((peeled.id, other)),
        }
    }

    // 2. The same for the excludes, tolerating gaps: a `have` from a shallow
    //    client can name a commit whose ancestry that client does not hold.
    let mut exclude_commits = Vec::new();
    for id in excludes {
        if !graph.contains(id)? {
            continue;
        }
        match graph.peel(id) {
            Ok(peeled) if peeled.kind == object::Kind::Commit => exclude_commits.push(peeled.id),
            Ok(_) => {}
            Err(Error::ObjectMissing { .. }) => {}
            Err(err) => return Err(err),
        }
    }

    // 3 & 4. What the client already has, and what it is about to be sent.
    //
    //    Two questions, and [`Limiting`] decides whether they take two walks or
    //    one. Both run before step 5 rather than after it, which is the order
    //    this used to be in, so that step 5 can *take* `excluded_commits`
    //    instead of copying it. Nothing here reads the object closure, only the
    //    commit graph, so the steps commute.
    //
    //    `Exclusion::Full` needs the root tree of every commit the client
    //    holds, and only the two-pass walk ever visits them, so it pins the
    //    choice. `excluded_trees` is that list, and it is empty under the
    //    merged walk because nothing there wants it.
    let limiting = match options.exclusion {
        Exclusion::Full => Limiting::TwoPass,
        Exclusion::Edges => options.limiting,
    };
    let mut excluded_trees: Vec<ObjectId> = Vec::new();
    crate::stage::end(_peel);
    let _walk = crate::stage::enter("select.commit_walk");
    let (commits, excluded_commits) = match limiting {
        Limiting::TwoPass => {
            let held = graph.commit_closure(
                &exclude_commits,
                &HashSet::new(),
                // Bounded by where the client said *its* history stops, not by
                // what this request decided to send. A shallow client offers
                // its grafted tip as a `have`; walking past that graft here
                // counts the whole repository as already-delivered and packs
                // nothing.
                &options.have_boundary,
                // The client's own history is not limited by what this request
                // decided to deepen to.
                None,
                Missing::Skip,
            )?;
            let excluded: OidSet = held.ids.iter().copied().collect();
            excluded_trees = held.trees;
            let commits = graph.commit_closure(
                &commit_tips,
                &excluded,
                &options.boundary,
                options.restrict,
                options.missing,
            )?;
            (commits, excluded)
        }
        Limiting::Merged => {
            let limited = graph.limited_closure(
                &commit_tips,
                &exclude_commits,
                &options.boundary,
                &options.have_boundary,
                options.restrict,
                options.missing,
            )?;
            // A frontier, where `TwoPass` puts the client's whole history here.
            // It seeds step 6's `seen`, where the only ids it can decide are
            // commits, and `tree_closure` never walks into one.
            let excluded: OidSet = limited.uninteresting.iter().copied().collect();
            let commits = CommitClosure {
                ids: limited.ids,
                trees: limited.trees,
                edges: limited.edges,
            };
            (commits, excluded)
        }
    };
    let CommitClosure {
        ids: commit_ids,
        trees: commit_trees,
        // Where the client's history meets what it is being sent. Free either
        // way: both walks already had to decide which parents the client holds.
        edges: commit_edges,
    } = commits;
    for commit in &commit_ids {
        if emitted.insert(*commit) {
            selection.objects.push(*commit);
        }
    }
    selection.commits = commit_ids;
    crate::stage::end(_walk);

    // 5. ...and as a set of every object under those commits. This is the
    //    expensive half, and the one the `git` package's reachability bitmaps replace.
    //
    //    **One set, and only one**. It starts as the excluded commits,
    //    moved rather than copied; the tree walk's `seen` set *is* this set, so
    //    a visited id is recorded once; and the walk tallies what it found
    //    rather than handing back a `Vec` of ids this function already holds.
    //    All three of those used to be separate repository-sized allocations.
    //    **Which** of the client's commits get opened up is [`Exclusion`], and
    //    it is the difference between reading 3 000 trees and reading 42.
    let mut excluded_objects = excluded_commits;
    {
        let _s = crate::stage::enter("select.exclude_trees");
        let mut tally = Emit::Tally(0);
        let mut buf = Vec::new();
        let mut parents = Vec::new();
        // `Full` already has every root tree in hand — the walk read them off
        // the commits it decoded; see `CommitClosure`. `Edges` names a handful
        // of commits instead and pays one `commit_links` each, which is a read
        // of an object sitting in the contiguous commit run at the front of the
        // pack, not another scattered fault in the middle of it.
        let roots: Vec<ObjectId> = match options.exclusion {
            Exclusion::Full => excluded_trees,
            Exclusion::Edges => {
                // The client's own tips, then the edge. Git's
                // `mark_edges_uninteresting` walks both: the tips because they
                // are what the client actually said it holds, the edge because
                // that is where the two histories meet and where the shared
                // content is.
                let mut roots = Vec::with_capacity(exclude_commits.len() + commit_edges.len());
                for id in exclude_commits.iter().chain(commit_edges.iter()) {
                    match graph.commit_links(id, &mut buf, &mut parents) {
                        Ok(tree) => roots.push(tree),
                        // The same tolerance step 2 already applies to a
                        // `have`: a shallow client can name a commit whose
                        // ancestry it does not hold, and refusing the fetch
                        // over it would be a regression on a path that used to
                        // skip it silently.
                        Err(Error::ObjectMissing { .. }) => {}
                        Err(err) => return Err(err),
                    }
                }
                roots
            }
        };
        for tree in &roots {
            graph.tree_closure(
                *tree,
                &mut excluded_objects,
                &HashSet::new(),
                None,
                Missing::Skip,
                // The client already has these; gunnar is not about to hold one.
                None,
                &mut tally,
            )?;
        }
    }

    // 6. Their trees and blobs, plus anything a want pointed at directly.
    //
    //    The exclusion set is handed straight over to be step 6's `seen` set.
    //    It used to be cloned here, and that clone was the defect. Merging the two
    //    is the same predicate written once, not a shortcut: `tree_closure`
    //    skips an id that is in `exclude` **or** already in `seen`, so a `seen`
    //    that starts out holding everything `exclude` held decides every id
    //    identically. The `exclude` argument is left for `verify_connectivity`,
    //    the one caller that must keep the two apart.
    let _tree = crate::stage::enter("select.tree_closure");
    let mut seen = excluded_objects;
    // ── a NAMED want under a filter is never "already held" (FORGE-SHALLOW) ──
    //
    // The exclusion above closes over the client's `have` tips, and for a
    // partial clone that closure is a lie in the dangerous direction: a
    // `blob:none` client's tip reaches every blob under it and it was sent none
    // of them. Its lazy fetch (`git fetch --filter=blob:none origin <blob>`)
    // names exactly such a blob, and with the blob in `seen` the arm below
    // skipped it and the answer was an EMPTY pack (`did not send all necessary
    // objects`; git.gunnar.rs, edda, 2026-10-28). Stock `git upload-pack`
    // sends it. Without a filter the client's tips do vouch for everything
    // under them, so this is the filtered case only.
    if options.filter.is_some() {
        for (id, _) in &direct {
            seen.remove(id);
        }
    }
    // Every id this step puts into `seen`, so the two questions that set is
    // answering can be separated again at the end. See [`Carried`], and
    // `Selection::client_has` for what goes wrong when they are not.
    let mut carried: Vec<ObjectId> = Vec::with_capacity(emitted.len());
    let mut payload = Vec::new();
    let mut keep = Emit::Keep(&mut payload);
    {
        let mut visited = Carried {
            seen: &mut seen,
            carried: &mut carried,
        };
        for id in emitted.iter() {
            visited.insert(id);
        }
        // The named wants FIRST. The commit-tree walk below inserts every object
        // it visits into `seen`, including the ones a filter then declines to
        // send, so a want naming a blob under one of the wanted commits would
        // be skipped as "visited" if it ran second. What a want names is sent.
        for (id, kind) in direct {
            match kind {
                object::Kind::Tree => graph.tree_closure(
                    id,
                    &mut visited,
                    &HashSet::new(),
                    options.filter,
                    options.missing,
                    options.max_blob_bytes,
                    &mut keep,
                )?,
                // A `want` naming a blob outright — a partial clone's lazy fetch
                // does exactly this. It never reaches `tree_closure`, so the
                // ceiling has to be applied here as well or the one shape that
                // asks for a single enormous object by name is the one shape that
                // is not checked.
                _ => {
                    if let (Some(limit), object::Kind::Blob) = (options.max_blob_bytes, kind) {
                        if let Some((_kind, size)) =
                            graph.store.header(id.as_bytes()).map_err(backend)?
                        {
                            if size > limit {
                                return Err(Error::ObjectTooLarge {
                                    oid: id,
                                    size,
                                    limit,
                                });
                            }
                        }
                    }
                    if visited.insert(&id) {
                        keep.push(id);
                    }
                }
            }
        }
        for tree in &commit_trees {
            graph.tree_closure(
                *tree,
                &mut visited,
                &HashSet::new(),
                options.filter,
                options.missing,
                options.max_blob_bytes,
                &mut keep,
            )?;
        }
    }
    selection.objects.extend(payload);
    // ── the voucher is what the client HOLDS ────────────────────────────────
    //
    // `seen` is now the client's closure **plus** everything this transfer
    // visited, and only the first half may vouch for a thin delta's base. So
    // the second half goes back out. It is `O(objects in this transfer)` and
    // frees memory rather than taking any, which is why this is a subtraction
    // rather than the snapshot-before-step-6 that the one-set rule forbids.
    //
    // `carried` is the exact list, not `selection.objects`: a walk under a
    // partial-clone filter *visits* trees and blobs it then declines to send,
    // and those are neither held nor carried. Using the sent list would leave
    // them vouched for.
    for id in &carried {
        seen.remove(id);
    }
    selection.client_has = ClientHas::Ids(seen);
    crate::stage::end(_tree);

    Ok(selection)
}

/// Every object reachable from `tips`, as **one** set and no vector.
///
/// This answers `upload-pack`'s [`WantPolicy::ReachableFromRefs`] question: *is
/// this `want` reachable from some reference?* It used to be answered by
/// `select_objects(graph, tips, &[], …).objects.into_iter().collect()`, and
/// that phrasing holds the whole repository **twice** at its peak — a `Vec` of
/// every id, alive while the `HashSet` built out of it fills up. At ten million
/// sha256 ids that is a 320 MB vector plus a ~500 MB set, ~820 MB of peak for a
/// question that never wanted the order and never wanted a second copy. It is
/// The same shape, one caller further out.
///
/// Here the tree walk's `seen` set **is** the answer: it tallies rather than
/// emits, so nothing is built beside it. The set that comes back is the only
/// repository-sized allocation the call ever makes.
///
/// A tip that does not resolve is **skipped**, not fatal. A dangling reference
/// is a problem for `gc`; refusing every non-tip `want` in the repository
/// because one reference is broken is not this function's call to make — the
/// same judgement `Frontier::new` already makes about the same values.
///
/// [`WantPolicy::ReachableFromRefs`]: crate::WantPolicy::ReachableFromRefs
pub(crate) fn reachable_closure(graph: &Graph<'_>, tips: &[ObjectId]) -> Result<OidSet> {
    let mut seen = OidSet::new();
    let mut commit_tips = Vec::new();
    let mut direct_trees = Vec::new();

    for tip in tips {
        if !graph.contains(tip)? {
            continue;
        }
        let peeled = match graph.peel(tip) {
            Ok(peeled) => peeled,
            Err(Error::ObjectMissing { .. }) => continue,
            Err(err) => return Err(err),
        };
        for tag in peeled.tags {
            seen.insert(tag);
        }
        match peeled.kind {
            object::Kind::Commit => commit_tips.push(peeled.id),
            object::Kind::Tree => direct_trees.push(peeled.id),
            _ => {
                seen.insert(peeled.id);
            }
        }
    }

    let commits = graph.commit_closure(
        &commit_tips,
        &HashSet::new(),
        &HashSet::new(),
        None,
        // A repository with a gap in it is a problem for `gc`, not a reason to
        // refuse every fetch.
        Missing::Skip,
    )?;
    let mut tally = Emit::Tally(0);
    for (commit, tree) in commits.ids.iter().zip(&commits.trees) {
        seen.insert(*commit);
        graph.tree_closure(
            *tree,
            &mut seen,
            &HashSet::new(),
            None,
            Missing::Skip,
            None,
            &mut tally,
        )?;
    }
    for tree in direct_trees {
        graph.tree_closure(
            tree,
            &mut seen,
            &HashSet::new(),
            None,
            Missing::Skip,
            None,
            &mut tally,
        )?;
    }
    Ok(seen)
}

/// What the repository already vouched for, before a push.
///
/// Built **once per push**, not once per reference command: a client pushing a
/// thousand refs in one go used to rebuild this a thousand times.
pub(crate) struct Frontier {
    /// Reference values that peel to a commit. The connectivity walk stops
    /// here: a commit a reference already points at passed this same check
    /// before that reference was allowed to move, so its closure is complete by
    /// induction.
    commits: HashSet<ObjectId>,
    /// Root trees of those commits, plus any reference value that peels to a
    /// tree or a blob. Anything in here is present and complete, so the tree
    /// walk does not descend into it.
    present: HashSet<ObjectId>,
}

impl Frontier {
    /// `values` is every reference target the repository held before the push.
    ///
    /// Absent and unreadable values are skipped rather than fatal: a repository
    /// with a dangling reference is broken, but refusing every push until
    /// someone fixes it is not this function's call to make.
    pub(crate) fn new(graph: &Graph<'_>, values: &[ObjectId]) -> Result<Self> {
        let mut commits = HashSet::new();
        let mut present = HashSet::new();
        for id in values {
            if !graph.contains(id)? {
                continue;
            }
            match graph.peel(id) {
                Ok(peeled) if peeled.kind == object::Kind::Commit => {
                    commits.insert(peeled.id);
                }
                Ok(peeled) => {
                    present.insert(peeled.id);
                }
                Err(Error::ObjectMissing { .. }) => {}
                Err(err) => return Err(err),
            }
        }
        for commit in &commits {
            present.insert(graph.commit(commit)?.tree);
        }
        Ok(Frontier { commits, present })
    }
}

/// Assert that every object reachable from `tips` is present, stopping at the
/// frontier.
///
/// This is receive-pack's connectivity check, and it walks **from the new tips
/// down**. It never computes the frontier's object closure, which is the whole
/// point: the frontier is every reference the repository already had, so its
/// closure is the entire repository, and paying for it once per pushed
/// reference made push latency a function of repository size.
///
/// # What this actually costs (the previous claim here was wrong)
///
/// The commit walk stops at any commit a reference already pointed at, so the
/// ordinary case — a push onto an existing branch — visits exactly the commits
/// being pushed. Their trees and blobs then get walked, with the frontier's
/// root trees excluded and every subtree visited at most once across the whole
/// push. So the honest bound is:
///
/// > `O(new commits + the tree closure of those commits)`
///
/// which is one snapshot's worth of trees, not the repository's history. It is
/// **not** `O(newly pushed history)` alone, and this is the part the old
/// comment claimed and never had:
///
/// * a blob is existence-checked, not inflated, so the tree side is cheap per
///   object but is still proportional to the number of paths in the snapshot;
/// * a push that bases a new branch on an old commit that is not itself a
///   reference value has nothing to stop the commit walk before the root, so it
///   degenerates to `O(history below the base)`.
///
/// Both remainders need reachability data gunnar does not have: the `git`
/// package's
/// `__gunnar_reach__` bitmaps, or the set of object ids the incoming pack
/// actually introduced, either of which turns "is this already here and
/// complete" into a lookup. Until one exists, this comment states the cost the
/// code has rather than the cost the design wants.
pub(crate) fn verify_connectivity(
    graph: &Graph<'_>,
    tips: &[ObjectId],
    frontier: &Frontier,
) -> Result<usize> {
    // 1. Peel the new tips. `peel` loads every object it walks through, so a
    //    tag chain is proven present by getting this far.
    //
    //    `tags` is bounded by `tips.len() * MAX_TAG_DEPTH` — by what the client
    //    sent, not by the repository — so it is a plain `HashSet` for the same
    //    reason `commit_closure`'s `edges` is. It is the only set here that
    //    exists purely for the returned count.
    let mut tags: HashSet<ObjectId> = HashSet::new();
    let mut commit_tips = Vec::new();
    let mut direct: Vec<(ObjectId, object::Kind)> = Vec::new();
    for tip in tips {
        let peeled = graph.peel(tip)?;
        tags.extend(peeled.tags);
        match peeled.kind {
            object::Kind::Commit => commit_tips.push(peeled.id),
            other => direct.push((peeled.id, other)),
        }
    }

    // 2. The commits being pushed. `Missing::Fail` is what makes this a check:
    //    a commit named by a parent and absent from the store stops the push.
    let commits = graph.commit_closure(
        &commit_tips,
        &frontier.commits,
        &HashSet::new(),
        None,
        Missing::Fail,
    )?;

    // 3. Their trees and blobs. One `seen` set for the whole push, so a subtree
    //    shared by ten pushed commits is walked once.
    //
    //    **`Tally`, not `Keep`.** The only caller — `receive_pack::apply` —
    //    throws the returned count's *identity* away and logs the number, so
    //    materialising a `Vec<ObjectId>` of every object in the push bought
    //    nothing and cost a second repository-sized allocation beside `seen`
    //    (the one-set rule, on the push side). There used to be a third: a `counted`
    //    `OidSet` deduplicating the three groups against each other, which
    //    cannot fire — a tag, a commit and a tree are different objects with
    //    different ids, `commit_closure` deduplicates its own output, and
    //    `seen` is shared across every `tree_closure` call below, so each group
    //    is already distinct and the groups are disjoint by kind. Three
    //    repository-sized structures on this path are now one.
    let mut seen = OidSet::new();
    let mut keep = Emit::Tally(0);
    for tree in &commits.trees {
        graph.tree_closure(
            *tree,
            &mut seen,
            &frontier.present,
            None,
            Missing::Fail,
            // A push is checked for completeness, not packed. Nothing here is
            // materialised, so the serve path's object ceiling does not apply —
            // the *receive* path's ceiling is `gunnar_pack::IndexOptions`, and
            // it has already run by the time this does.
            None,
            &mut keep,
        )?;
    }
    for (id, kind) in direct {
        match kind {
            object::Kind::Tree => graph.tree_closure(
                id,
                &mut seen,
                &frontier.present,
                None,
                Missing::Fail,
                None,
                &mut keep,
            )?,
            // A reference pointing straight at a blob. `peel` already read it.
            _ => {
                if seen.insert(id) {
                    keep.push(id);
                }
            }
        }
    }

    Ok(tags.len() + commits.ids.len() + keep.len())
}

/// Memoised "does this commit have an ancestor the client said it has".
///
/// This is git's `can_all_from_reach_with_flag`, which is what decides whether
/// the server may stop negotiating. Memoising across wants is what keeps it
/// `O(V+E)` for the whole request instead of `O(V+E)` per want. On a
/// merge-heavy repository that is milliseconds against minutes.
pub(crate) struct Reachability<'g, 'a> {
    graph: &'g Graph<'a>,
    common: OidSet,
    known: HashMap<ObjectId, bool>,
    cutoff: i64,
    visited: usize,
    budget: Option<usize>,
}

impl<'g, 'a> Reachability<'g, 'a> {
    pub(crate) fn new(graph: &'g Graph<'a>, common: &[ObjectId], cutoff: i64) -> Self {
        Reachability {
            graph,
            common: common.iter().copied().collect(),
            known: HashMap::new(),
            cutoff,
            visited: 0,
            budget: graph.budget,
        }
    }

    /// Can `start` reach any commit the client claimed?
    ///
    /// Commits older than the cutoff are answered `false` without being walked
    /// through: the client's oldest `have` bounds how far back a useful cut
    /// point can be, and without that bound a divergent history walks to the
    /// root of the repository on every round.
    ///
    /// # One read per commit, and why the obvious shape is two
    ///
    /// This used to be an iterative **post-order**: a commit was left on the
    /// stack under its unresolved parents and looked at again once they were
    /// known. That is a correct way to fold a DAG, and it reads every commit on
    /// the path **twice**: once going down, once on the way back up, because
    /// the second look has to re-parse the commit to get at its parents again.
    /// On a divergent negotiation that doubling was a third of the entire
    /// suite's most expensive arm.
    ///
    /// It is not needed, because the fold being computed is an *or*. A plain
    /// downward walk answers `true` the instant it touches a common commit, and
    /// if it exhausts without touching one then **every commit it visited** is
    /// proven not to reach a common. One sweep, one read each, and a stronger
    /// memo than the post-order produced. The `true` case only memoises `start`
    /// itself: the walk stops early, so nothing else on the frontier has been
    /// settled and claiming otherwise would be a lie the next `want` acts on.
    ///
    /// One consequence worth stating rather than discovering: `visited` now
    /// counts each commit **once**, where the post-order counted the ones it
    /// read twice twice. So the budget is reached after roughly twice as many
    /// commits as before. That is the count being *correct* rather than the
    /// ceiling being raised; the budget bounds objects visited, and it was
    /// previously charging some of them double. It does mean the effective
    /// ceiling on a divergent walk moved, so judge `DEFAULT_BUDGET` against the
    /// honest number.
    pub(crate) fn reaches_common(&mut self, start: ObjectId) -> Result<bool> {
        if self.common.is_empty() {
            return Ok(false);
        }
        if let Some(&answer) = self.known.get(&start) {
            return Ok(answer);
        }

        // Breadth-first rather than depth-first so that a common commit a few
        // steps away is found in a few steps, whatever the depth of the branch
        // it happens to hang off.
        let mut queue: VecDeque<ObjectId> = VecDeque::new();
        let mut queued = OidSet::new();
        // Visited, and unless the walk stops early, proven `false`.
        let mut exhausted: Vec<ObjectId> = Vec::new();
        queue.push_back(start);
        queued.insert(start);

        let mut found = false;
        while let Some(id) = queue.pop_front() {
            if self.common.contains(&id) {
                found = true;
                break;
            }
            match self.known.get(&id) {
                Some(true) => {
                    found = true;
                    break;
                }
                // Already proven not to reach a common: its ancestry adds
                // nothing, and re-walking it is what the memo exists to stop.
                Some(false) => continue,
                None => {}
            }
            self.visited += 1;
            if let Some(limit) = self.budget {
                if self.visited > limit {
                    return Err(Error::WalkBudget { limit });
                }
            }
            // `visited` is cumulative and `queued` is live, so they are two
            // different quantities and both need saying. This one is the shared
            // charge every other walk answers to, which is what puts a
            // negotiation's `queued` set under the same bytes ceiling as a
            // clone's `seen` set instead of under none.
            self.graph.charge_walk()?;
            if !self.graph.contains(&id)? {
                self.known.insert(id, false);
                continue;
            }
            let facts = self.graph.commit(&id)?;
            if facts.committed_at < self.cutoff {
                self.known.insert(id, false);
                continue;
            }
            exhausted.push(id);
            for parent in facts.parents {
                if queued.insert(parent) {
                    queue.push_back(parent);
                }
            }
        }

        if found {
            self.known.insert(start, true);
        } else {
            // The queue drained, so every commit on this frontier had its whole
            // in-cutoff ancestry looked at and none of it was common.
            for id in exhausted {
                self.known.insert(id, false);
            }
        }
        Ok(found)
    }
}

#[cfg(test)]
mod tests {

    /// [`SelectOptions`] with the **bitmap short-circuit off**.
    ///
    /// # Why every walk test in this module needs it, as of 2026-08-10
    ///
    /// `SelectOptions::default()` leaves [`Bitmaps::Read`] on, and step 0 then
    /// asks [`GitServe::select`] before any walk happens. The old fixture store
    /// could not answer — `ObjectStore::reachable()` handed back `None`,
    /// because `MemObjectStore` had no reachability index — so the question was
    /// free and every test below fell through to the walk it was written about.
    ///
    /// [`MemGitServe::select`] **does** answer: it is a genuine reachability
    /// walk over an in-memory store, and it declines only for an empty store or
    /// an unknown `want`. So with the default options these tests measure the
    /// short-circuit and not the traversal — seen exactly that way when they
    /// first compiled again: *"the two-pass walk touched the store 3 times over
    /// a client history of 300 commits"*, and a fetch whose peak was 0 ids.
    ///
    /// `Bitmaps::Off` is the documented *"walk, always"*, the same switch
    /// `GUNNAR_BITMAP=off` sets, so this forces the subject of the test rather
    /// than working around anything. What step 0 itself does is
    /// `crate::reachable`'s to assert, and it does.
    fn walked() -> SelectOptions<'static> {
        SelectOptions {
            bitmaps: Bitmaps::Off,
            ..SelectOptions::default()
        }
    }
    use super::*;
    use gix_hash::Kind;

    // The byte-level suites' fixture builder, reused rather than re-written
    // (one implementation). Declared in `lib.rs`; see the `#[path]` there.
    use crate::common::Fixture;

    /// A fetch holds the client's object closure **once**.
    ///
    /// # What is measured, and why it is not a wall clock or an RSS reading
    ///
    /// The bug this guards is invisible to every other kind of assertion. It
    /// changes no byte on the wire, no object in the pack and no exit status;
    /// it only makes the server hold a second copy of a repository-sized set of
    /// object ids. So the instrument is a count of the ids the walkers hold at
    /// once ([`meter`]), which is the applied output of the thing that went
    /// wrong. Peak resident set is the same fact one layer down, but a
    /// `MemObjectStore` fixture *is* memory, with the objects outweighing the
    /// sets about four to one, so RSS on a fixture this size could not resolve
    /// the difference. That measurement belongs on a real repository, and it
    /// lives in the `bench/` harness.
    ///
    /// # The two bounds, and what each one catches
    ///
    /// **The ceiling** is `one closure + a scrap`. Before the fix the peak was
    /// the closure **twice** — `let mut seen = excluded_objects.clone();` —
    /// plus a `Vec` of the same ids and a separate `seen` set inside the
    /// exclusion walk. Seen RED: with the clone restored this asserts
    /// 2.0 × closure against a bound of 1.02 ×.
    ///
    /// **The floor** is what stops the guard being hollow, and it is not
    /// decoration: a `select_objects` that never built the exclusion closure at
    /// all would hold almost nothing, pass any ceiling, and be a far worse bug,
    /// because it would send the client the whole repository on every
    /// incremental fetch. `peak >= closure` says the closure really was built.
    /// The selection assertion below is the same claim from the other side.
    #[test]
    fn a_fetch_holds_the_clients_object_closure_once() {
        // 300 commits, each with its own tree and its own blob, so the closure
        // is ~900 ids. Far enough above the handful of scrap ids that the two
        // bounds cannot overlap.
        const HISTORY: usize = 300;

        let fixture = Fixture::build(Kind::Sha1);
        let client_tip = fixture.extend(fixture.four, HISTORY, "client");
        let server_tip = fixture.extend(client_tip, 1, "server");

        let graph = Graph::new(&fixture.git, fixture.kind);
        // `Full`, deliberately and explicitly: this test's subject is the
        // *whole* client closure being held exactly once, and under the
        // shipping [`Exclusion::Edges`] there is no whole client closure to
        // hold. `Edges` gets its own half at the bottom, and the two together
        // say more than either did alone.
        let options = SelectOptions {
            exclusion: Exclusion::Full,
            ..walked()
        };

        // How big one copy of the client's closure is, measured by the same
        // walker rather than predicted: a clone with no excludes selects
        // exactly the set an incremental fetch has to exclude.
        let closure = select_objects(&graph, &[client_tip], &[], &options)
            .expect("the client's own closure")
            .objects
            .len();
        assert!(
            closure > 3 * HISTORY,
            "the fixture is not the shape this test needs: {HISTORY} commits \
             should close over more than {} objects, not {closure}",
            3 * HISTORY
        );

        meter::reset_peak();
        assert_eq!(
            meter::live(),
            0,
            "a set from the sizing call above outlived it"
        );
        let selection = select_objects(&graph, &[server_tip], &[client_tip], &options)
            .expect("the incremental fetch");
        let peak = meter::peak();

        // One new commit, its tree and its blob. Nothing the client already
        // had. Without this the memory bounds could be satisfied by a walk that
        // excluded far too much.
        assert_eq!(
            selection.objects.len(),
            3,
            "an incremental fetch one commit behind selected {} objects",
            selection.objects.len()
        );

        assert!(
            peak >= closure,
            "the walk peaked at {peak} live object ids against a client closure \
             of {closure}: the exclusion set was never built, so this fetch is \
             about to send objects the client already has"
        );
        let ceiling = closure + 4 * selection.objects.len() + 16;
        assert!(
            peak <= ceiling,
            "the walk held {peak} object ids at once for a client closure of \
             {closure}, {:.2}× one copy, ceiling {ceiling}. The exclusion \
             closure is being materialised more than once; at ten \
             million objects each extra copy is ~450 MB per in-flight fetch",
            peak as f64 / closure as f64
        );

        // ── the closure is now HANDED OUT, and that has a lifetime cost ──────
        //
        // The thin voucher needs the client's closure after the walk, to vouch for the
        // base of a thin delta, so `Selection::client_has` carries it to the
        // pack writer instead of dropping it. The peak asserted above is
        // unchanged — it is the same set moved, not a second copy — but it now
        // lives as long as the `Selection` does, which means it coexists with
        // the writer's own tables rather than dying before them. That is a
        // real entry in `limits.rs`'s ledger and it is asserted here rather
        // than described.
        let full_selected = selection.objects.len();
        let vouched = selection.client_has.len();
        assert!(
            vouched >= closure,
            "the walk handed back {vouched} ids to vouch with against a client \
             closure of {closure}: the set being carried is not the closure, so \
             a thin delta's base would be refused for want of evidence that was \
             computed and thrown away"
        );
        assert_eq!(
            meter::live(),
            vouched,
            "the ids alive after the walk are not exactly the set it handed \
             back, so something else is being retained too"
        );
        drop(selection);
        assert_eq!(
            meter::live(),
            0,
            "the client's closure outlived the Selection that owns it"
        );

        // And now the shipping mode. `Edges` opens up the client's tips and
        // the edge and nothing else, so the set it holds is bounded by the
        // frontier rather than by the client's history — which is the entire
        // read-side claim of the serial walk, asserted here as *live ids*
        // rather than as a timing.
        meter::reset_peak();
        let edge_selection = select_objects(&graph, &[server_tip], &[client_tip], &walked())
            .expect("the incremental fetch, edge exclusion");
        let edge_peak = meter::peak();

        // The pack must not have grown. `Edges` is allowed to send a *superset*
        // and this fixture is the shape where the superset is empty; if that
        // ever stops being true here, the trade has moved and this is where it
        // shows up rather than in a byte count nobody reads.
        assert_eq!(
            edge_selection.objects.len(),
            full_selected,
            "edge exclusion selected {} objects where the full closure selected \
             {full_selected}: the frontier walk is missing objects the client \
             demonstrably has",
            edge_selection.objects.len()
        );
        // And the set carried to the writer shrinks with it. This is what
        // bounds the thin voucher's new lifetime cost in the SHIPPING
        // configuration: the voucher is frontier-sized, not history-sized, so
        // a fetch does not hold the client's whole history alive while it
        // writes the pack.
        assert!(
            edge_selection.client_has.len() < vouched,
            "the voucher carried under edge exclusion is {} ids against the full \
             closure's {vouched}, so the frontier walk is handing the pack writer \
             the client's whole history after all",
            edge_selection.client_has.len()
        );
        // Seen RED by making `Exclusion::from_env`'s default `Full` and
        // `SelectOptions::default()` follow it: `edge_peak` then equals `peak`
        // and this fires with "held 915 ids ... against the full closure's 915".
        assert!(
            edge_peak < peak,
            "edge exclusion held {edge_peak} ids at once against the full \
             closure's {peak} over a client history of {HISTORY} commits. It is \
             walking the client's whole history after all, and the read-side \
             win is not there"
        );
    }

    /// The traversal budget still bounds the **exclusion** walk.
    ///
    /// That walk stopped collecting its ids into a `Vec` when the extra copy
    /// was removed and tallies them instead, with the budget charged against
    /// whichever of the two the walk is using. A budget that silently stopped
    /// applying to the one walk sized by *the client's* history would be an
    /// unbounded client-driven allocation, which is the whole class this fix
    /// is in.
    ///
    /// # The first version of this test could not fail, and that is the point
    ///
    /// It used a budget of 10 against 200 commits, so the **commit** walk in
    /// step 3 tripped first and the assertion was green whatever the tree walk
    /// did. Proved by breaking `Emit::Tally`'s `len()` to return `0` and
    /// watching it stay green. The budget here is therefore chosen to sit in
    /// the gap between the two: comfortably above the commit count, and below
    /// the object count those commits close over. Seen RED with the same
    /// `len()` broken: "the exclusion walk ran to completion under a budget of
    /// 400 and selected 3 objects".
    #[test]
    fn the_budget_bounds_the_exclusion_walk_not_only_the_commit_walk() {
        // 300 + the fixture's own 5 commits, each with one tree and one blob:
        // ~305 commits to walk, ~610 objects under them.
        const HISTORY: usize = 300;
        const BUDGET: usize = 400;

        let fixture = Fixture::build(Kind::Sha1);
        let client_tip = fixture.extend(fixture.four, HISTORY, "client");
        let server_tip = fixture.extend(client_tip, 1, "server");

        // The commit walk alone must NOT reach the budget, or this test is the
        // hollow one it replaced. Measured against the same graph.
        let graph = Graph::new(&fixture.git, fixture.kind);
        let commits = graph
            .commit_closure(
                &[client_tip],
                &HashSet::new(),
                &HashSet::new(),
                None,
                Missing::Skip,
            )
            .expect("the client's commits");
        assert!(
            commits.ids.len() < BUDGET,
            "the {} commits alone reach a budget of {BUDGET}, so this test              cannot tell the two walks apart",
            commits.ids.len()
        );

        let graph = Graph::new(&fixture.git, fixture.kind).with_budget(Some(BUDGET));
        let outcome = select_objects(
            &graph,
            &[server_tip],
            &[client_tip],
            &SelectOptions {
                exclusion: Exclusion::Full,
                ..walked()
            },
        );

        match outcome {
            Err(Error::WalkBudget { limit }) => assert_eq!(limit, BUDGET),
            Err(other) => panic!("the budget should have stopped this walk, not {other}"),
            Ok(selection) => panic!(
                "the exclusion walk ran to completion under a budget of \
                 {BUDGET} and selected {} objects",
                selection.objects.len()
            ),
        }

        // The other half of the same fact, and the reason the budget stopped
        // being reached in practice: under `Exclusion::Edges` this identical
        // fetch, on the identical graph, under the identical budget, does not
        // come near it. The budget still guards the walk — it is the walk that
        // got smaller.
        let edges = select_objects(&graph, &[server_tip], &[client_tip], &walked()).expect(
            "edge exclusion tripped a budget of 400 on a 305-commit history; \
             it is supposed to open up the frontier, not the history",
        );
        assert_eq!(
            edges.objects.len(),
            3,
            "edge exclusion selected {} objects where the full closure selects 3",
            edges.objects.len()
        );
    }

    /// A fetch does not read the history the client just told it about.
    ///
    /// # What is measured
    ///
    /// Object **reads**, counted by the store itself. Not wall clock, which on
    /// a `MemObjectStore` is a `BTreeMap` lookup either way and would report a
    /// win of nothing; not the pack, which must not move at all. The claim of
    /// [`Limiting::Merged`] is that a fetch stops asking for commits the client
    /// already named, and a count of what the store was asked for is that claim
    /// as applied output.
    ///
    /// # The three bounds
    ///
    /// **The floor on the old walk.** `TwoPass` must read more commits than the
    /// client's history is long, or the fixture is not the shape that can tell
    /// the two apart and every ratio below is against nothing.
    ///
    /// **The ceiling on the new one.** `Merged` must read fewer than a tenth as
    /// many. Seen RED by handing both sides `Limiting::TwoPass`: "the merged
    /// walk read 916 objects against the two-pass walk's 916".
    ///
    /// **The pack must not move.** Same object *set*, both ways. The dangerous
    /// direction is a merged walk that stops too early and sends fewer objects
    /// than the client needs, which is a broken fetch rather than a slow one;
    /// a set comparison catches it in either direction. Seen RED by raising
    /// [`SLOP`]'s countdown into an unconditional `break` — "the merged walk
    /// selected 3 objects where the two-pass walk selected 18".
    #[test]
    fn a_fetch_does_not_read_the_history_the_client_already_told_it_about() {
        use crate::common::Counting;

        const HISTORY: usize = 300;
        const DAY: usize = 5;

        let fixture = Fixture::build(Kind::Sha1);
        // Committer dates ascending across BOTH calls, which `extend`'s default
        // does not give: it restarts its clock per call, so the commits a fetch
        // is about to send would be dated older than the entire history behind
        // them. That is a real shape — a rebased branch looks like it — but it
        // is the shape a date-ordered walk cannot shortcut, and pricing the
        // walk on it would be pricing its worst case as its behaviour.
        let client_tip = fixture.extend_dated(fixture.four, HISTORY, "client", |i| {
            1_700_200_000 + i as i64
        });
        let server_tip = fixture.extend_dated(client_tip, DAY, "server", |i| {
            1_700_200_000 + (HISTORY + i) as i64
        });

        let measure = |limiting: Limiting| -> (usize, Vec<ObjectId>) {
            let store = Counting::new(fixture.git.inner.clone());
            let graph = Graph::new(&store, Kind::Sha1);
            let options = SelectOptions {
                limiting,
                ..walked()
            };
            let selection = select_objects(&graph, &[server_tip], &[client_tip], &options)
                .expect("the incremental fetch");
            let mut ids = selection.objects;
            ids.sort();
            (store.touches(), ids)
        };

        let (two_pass_reads, two_pass_objects) = measure(Limiting::TwoPass);
        let (merged_reads, merged_objects) = measure(Limiting::Merged);

        assert!(
            two_pass_reads > HISTORY,
            "the two-pass walk touched the store {two_pass_reads} times over a \
             client history of {HISTORY} commits, so it is not walking that \
             history and this test cannot tell the two walks apart"
        );
        assert_eq!(
            merged_objects,
            two_pass_objects,
            "the merged walk selected {} objects where the two-pass walk \
             selected {}. Fewer is a broken fetch, not a win",
            merged_objects.len(),
            two_pass_objects.len()
        );
        assert!(
            merged_objects.len() >= 3 * DAY,
            "an incremental fetch of {DAY} commits selected only {} objects, so \
             the comparison above is between two empty sets",
            merged_objects.len()
        );
        assert!(
            merged_reads * 10 < two_pass_reads,
            "the merged walk touched the store {merged_reads} times against the \
             two-pass walk's {two_pass_reads}, {:.2}x. It is still enumerating \
             the client's history, which is the whole cost this removes",
            two_pass_reads as f64 / merged_reads as f64
        );
    }

    /// The merged walk and the two-pass walk pick the same objects, on the
    /// shapes where they could disagree.
    ///
    /// The measurement above runs one linear fetch. These are the shapes a
    /// date-ordered walk can get wrong and a breadth-first one cannot: a
    /// **merge** whose two parents are on different sides of the frontier, a
    /// `have` the client names that is also a `want`, a `have` that is an
    /// ancestor of another `have`, and a fetch whose wants are dated *older*
    /// than the history behind them, which is what `SLOP` exists for.
    ///
    /// Equality is asserted on the sorted id list, because the merged walk
    /// emits in committer-date order and the two-pass walk breadth-first, and
    /// the *set* is what a pack has to contain.
    ///
    /// Seen RED by dropping [`SLOP`] to `0`, which ends the walk on the first
    /// uninteresting pop: "skewed wants: merged selected 3, two-pass 15".
    #[test]
    fn the_two_walks_agree_on_the_shapes_that_could_split_them() {
        let fixture = Fixture::build(Kind::Sha1);
        let old_line = fixture.extend_dated(fixture.three, 8, "old", |i| 1_700_300_000 + i as i64);
        let new_line = fixture.extend_dated(fixture.two, 8, "new", |i| 1_700_400_000 + i as i64);
        // A merge of a recent line and a stale one: its second parent is older
        // than commits the client already has, which is exactly where a walk
        // ordered by date can pop things in a surprising order.
        let merged_tip = crate::common::commit(
            &fixture.git,
            fixture.kind,
            "merged.txt",
            "merged",
            &[new_line, old_line],
            1_700_400_100,
        );
        // Wants dated older than everything behind them: a rebased branch, and
        // the case `SLOP` absorbs.
        let skewed = fixture.extend_dated(new_line, 4, "skew", |_| 1_700_000_000);

        // The `bool` is whether the fetch is expected to select anything at
        // all. Without it the "nothing to send" case would satisfy an equality
        // between two empty lists and quietly become the hollow half of this
        // guard — while still being worth asserting, because a merged walk that
        // sent the client its own tip back would show up here and nowhere else.
        let cases: [(&str, &[ObjectId], &[ObjectId], bool); 5] = [
            (
                "merge across the frontier",
                &[merged_tip],
                &[old_line],
                true,
            ),
            (
                "a want that is also a have",
                &[merged_tip],
                &[merged_tip],
                false,
            ),
            (
                "one have behind another",
                &[merged_tip],
                &[new_line, fixture.two],
                true,
            ),
            ("skewed wants", &[skewed], &[new_line], true),
            ("no haves at all", &[merged_tip], &[], true),
        ];

        for (name, tips, haves, expect_objects) in cases {
            let pick = |limiting: Limiting| -> Vec<ObjectId> {
                let graph = Graph::new(&fixture.git, fixture.kind);
                let mut ids = select_objects(
                    &graph,
                    tips,
                    haves,
                    &SelectOptions {
                        limiting,
                        ..walked()
                    },
                )
                .unwrap_or_else(|err| panic!("{name}: {err}"))
                .objects;
                ids.sort();
                ids
            };
            let merged = pick(Limiting::Merged);
            let two_pass = pick(Limiting::TwoPass);
            assert_eq!(
                !two_pass.is_empty(),
                expect_objects,
                "{name}: the two-pass walk selected {} objects, which is not the \
                 shape this case is here to compare",
                two_pass.len()
            );
            assert_eq!(
                merged,
                two_pass,
                "{name}: merged selected {}, two-pass {}",
                merged.len(),
                two_pass.len()
            );
        }
    }

    /// A `have` older than its own parent still stops the walk behind it.
    ///
    /// # The one thing the two comparisons above cannot see
    ///
    /// They assert that the merged walk and the two-pass walk pick the same
    /// objects. git's `mark_parents_uninteresting` is invisible to that,
    /// because the only shape where it changes anything is a shape where the
    /// two walks are *allowed* to differ — the merged one sends a documented
    /// superset there, so an equality assertion is either already failing or
    /// already blind. Proved: with the whole function replaced by an early
    /// `return`, all five tests in this module stay green.
    ///
    /// So this asserts the applied output directly, on the shape that needs it:
    ///
    /// ```text
    ///   A (t=1000) ── B (t=2000) ── W (t=3000)   the want
    ///                     └──────── U (t=1500)   the have, OLDER than B
    /// ```
    ///
    /// `W` is popped first and takes `B` for interesting, so `B`'s parent `A`
    /// is admitted interesting too. `U` arrives afterwards and marks `B`
    /// uninteresting — and `A` is only reached by pushing that mark on down
    /// through a commit already read, which is what the function does and the
    /// only thing that does it.
    ///
    /// Seen RED with that early `return` in place: "the merged walk sent 3
    /// commits, including A, which sits behind a commit the client holds".
    ///
    /// `B` itself **is** sent, and that is the trade rather than a bug: it was
    /// emitted before `U` was ever popped. The two-pass walk sends only `W`.
    /// The assertion records both numbers so the size of the superset is a
    /// stated fact instead of a surprise.
    #[test]
    fn a_have_older_than_its_own_parent_still_stops_the_walk_behind_it() {
        use crate::common::commit;
        use gunnar_store::mem::MemGitServe;

        let store = MemGitServe::new(Kind::Sha1);
        let a = commit(&store, Kind::Sha1, "a.txt", "a", &[], 1_000);
        let b = commit(&store, Kind::Sha1, "b.txt", "b", &[a], 2_000);
        let u = commit(&store, Kind::Sha1, "u.txt", "u", &[b], 1_500);
        let w = commit(&store, Kind::Sha1, "w.txt", "w", &[b], 3_000);

        let graph = Graph::new(&store, Kind::Sha1);
        let pick = |limiting: Limiting| -> Vec<ObjectId> {
            select_objects(
                &graph,
                &[w],
                &[u],
                &SelectOptions {
                    limiting,
                    ..walked()
                },
            )
            .expect("the skewed fetch")
            .commits
        };

        let merged = pick(Limiting::Merged);
        assert!(
            !merged.contains(&a),
            "the merged walk sent {} commits, including A, which sits behind a \
             commit the client holds: the client's side of the graph stopped \
             propagating at the first commit already taken for interesting",
            merged.len()
        );
        assert_eq!(
            merged,
            vec![w, b],
            "the merged walk sent {merged:?}; W and B is the documented \
             superset on this shape, and anything else has moved it"
        );
        assert_eq!(
            pick(Limiting::TwoPass),
            vec![w],
            "the two-pass walk is the exact answer here and must stay it, or \
             the superset above is being measured against a moving baseline"
        );
    }

    /// The edge is the parent the two histories actually share, not the tip the
    /// client named.
    ///
    /// # Why this was missing, and what it was missing from
    ///
    /// [`Exclusion::Edges`] opens up two things in step 5: the commits the
    /// client named as `have`s, and the **edge**. On a linear fetch those are
    /// the same commit, so every test in this crate that fetches a line forward
    /// is green with the edge set emptied — proved, by emptying it. The edge
    /// only does work when the two histories join *behind* what the client
    /// named:
    ///
    /// ```text
    ///   A ── C1 ── C2      the client's line; C2 is the `have`
    ///    └── S1            the server's;      S1 is the `want`
    /// ```
    ///
    /// `S1` carries the same file and the same content as `A`, so it shares
    /// `A`'s tree and blob object for object. `A` is two hops behind the `have`
    /// and is named by nobody. A step 5 that opened up only what the client
    /// named would put that tree and that blob in the pack again.
    ///
    /// Asserted for **both** walks, because both compute an edge set and this
    /// is the only test that can tell either of them is doing it.
    ///
    /// Seen RED, both ways: `Limiting::Merged`'s edge pass turned into
    /// `.take(0)` and `commit_closure`'s `out.edges.push(parent)` removed —
    /// "merged: a fetch sharing a tree and a blob with the client packed 3
    /// objects, not 1" and the same line for two-pass.
    #[test]
    fn the_exclusion_opens_up_the_parent_the_two_histories_share() {
        use crate::common::commit;
        use gunnar_store::mem::MemGitServe;

        let store = MemGitServe::new(Kind::Sha1);
        let a = commit(&store, Kind::Sha1, "shared.txt", "shared", &[], 1_000);
        let c1 = commit(&store, Kind::Sha1, "c1.txt", "c1", &[a], 2_000);
        let c2 = commit(&store, Kind::Sha1, "c2.txt", "c2", &[c1], 3_000);
        // Same path, same content, so the same tree and the same blob as `A`.
        let s1 = commit(&store, Kind::Sha1, "shared.txt", "shared", &[a], 4_000);
        assert_ne!(
            s1, a,
            "the two commits must differ, only their content may not"
        );

        let graph = Graph::new(&store, Kind::Sha1);
        for limiting in [Limiting::Merged, Limiting::TwoPass] {
            let selection = select_objects(
                &graph,
                &[s1],
                &[c2],
                &SelectOptions {
                    limiting,
                    ..walked()
                },
            )
            .expect("the branch fetch");
            assert_eq!(
                selection.objects,
                vec![s1],
                "{limiting:?}: a fetch sharing a tree and a blob with the client \
                 packed {} objects, not 1. The edge behind the client's named \
                 tip was never opened up, so content the client demonstrably \
                 holds is going back over the wire",
                selection.objects.len()
            );
        }
    }

    /// `CommitClosure::trees[i]` really is `ids[i]`'s root tree.
    ///
    /// # Why this needs its own guard, when 500 other tests already pack packs
    ///
    /// Nothing downstream can tell a **mispaired** tree from a correct one. Every
    /// tree in `trees` is a real tree of a real commit in the same closure, so a
    /// walk that paired them by an off-by-one would still visit exactly the same
    /// set of trees, still emit a complete and self-consistent pack, and still
    /// satisfy every "the client got what it asked for" assertion in the suite —
    /// on a linear history, where the closure is one commit per tree, it would
    /// produce a **byte-identical** pack. The bug only becomes visible on a
    /// shape where the sets differ, and it would then look like a corrupt
    /// repository rather than like a walker bug. So the pairing is asserted
    /// here, element by element, against a **second and independent** read of
    /// each commit through `Graph::commit`, which is the full
    /// `CommitRef::from_bytes` decoder rather than the `CommitRefIter` one the
    /// walk now uses. That the two decoders agree is the other half of what is
    /// being checked.
    ///
    /// Seen RED, both ways: `out.trees.push(tree)` moved above `out.ids.push(id)`
    /// with a one-element skew fails on the first pair; returning
    /// `graph.commit(&id)?.tree` for the *previous* commit fails likewise. The
    /// budget test above and the one-set test both stay green under either edit.
    ///
    /// # The floor
    ///
    /// A `zip` over two empty vectors asserts nothing and passes. `HISTORY` is
    /// checked to have actually arrived, because that is the shape this guard
    /// would otherwise silently take.
    #[test]
    fn the_commit_walk_hands_back_each_commits_own_root_tree() {
        const HISTORY: usize = 40;

        let fixture = Fixture::build(Kind::Sha1);
        let tip = fixture.extend(fixture.four, HISTORY, "p010");
        let graph = Graph::new(&fixture.git, fixture.kind);

        let closure = graph
            .commit_closure(
                &[tip],
                &HashSet::new(),
                &HashSet::new(),
                None,
                Missing::Fail,
            )
            .expect("the commit closure");

        assert_eq!(
            closure.ids.len(),
            closure.trees.len(),
            "the two halves of a CommitClosure must be the same length or the \
             pairing is not a pairing"
        );
        assert!(
            closure.ids.len() > HISTORY,
            "the fixture is not the shape this test needs: {HISTORY} commits on \
             top of the base should close over more than {HISTORY} commits, not \
             {}. A zip over a shorter list would assert nothing.",
            closure.ids.len()
        );

        for (i, (id, tree)) in closure.ids.iter().zip(&closure.trees).enumerate() {
            let independent = graph
                .commit(id)
                .expect("re-reading a commit the walk just read")
                .tree;
            assert_eq!(
                *tree, independent,
                "commit {i} ({id}) was paired with tree {tree}, but its own root \
                 tree is {independent}"
            );
        }
    }

    // ── the bytes ceiling ──────────────────────────────────────────────────

    /// The number of ids a `walk_bytes` ceiling of `objects` objects is.
    fn ceiling_for(objects: usize) -> usize {
        objects * crate::WALK_BYTES_PER_OBJECT
    }

    /// **A fetch is charged for what it HOLDS, not for what it sends**, and a
    /// `blob:none` filter is where the two come apart.
    ///
    /// # Why this shape and not a big clone
    ///
    /// The old ceiling compared `Emit::len()` — the objects the walk is about
    /// to send — against the budget. On a full clone that is nearly the whole
    /// set held and the difference does not show. Under `filter=blob:none` the
    /// walk still puts **every blob** into its `seen` set, because it has to
    /// know it has been there, and then declines to emit any of them. So the
    /// thing being counted stops growing while the thing being allocated keeps
    /// going, and a ceiling written against the first cannot see the second.
    ///
    /// # Both edges, and the floor
    ///
    /// | ceiling | must |
    /// |---|---|
    /// | the ids the walk EMITS | refuse — it is holding more than that |
    /// | the ids the walk HOLDS, and a scrap | serve |
    ///
    /// The second is not decoration. A ceiling that refused at both would be
    /// satisfied by a charge that is simply too high — or by a walk that cannot
    /// serve anything — and this test would then be green about a server that
    /// clones nothing.
    ///
    /// Seen RED with `charge_walk` reading zero instead of [`meter::live`]:
    /// *"a ceiling the size of what this fetch sends must refuse it"*.
    ///
    /// # What it does NOT guard, said plainly
    ///
    /// [`WALK_BYTES_PER_OBJECT`](crate::WALK_BYTES_PER_OBJECT). Doubling that
    /// constant leaves this test green, because `ceiling_for` sizes the
    /// ceiling from the same constant — deliberately, so the test is about the
    /// mechanism and survives a re-measurement, but it does mean nothing here
    /// can tell a well-chosen rate from a wrong one. The rate is a measurement
    /// and only a measurement can check it: `bench/examples/walkbudget-probe.rs`,
    /// and the table on the constant is what it produced.
    #[test]
    fn the_ceiling_charges_what_the_walk_holds_not_what_it_sends() {
        const HISTORY: usize = 300;
        let fixture = Fixture::build(Kind::Sha1);
        let tip = fixture.extend(fixture.four, HISTORY, "held");

        let blobless = Filter::parse("blob:none").expect("blob:none parses");
        // The ceiling under test lives on the `Graph`, not here: it is a
        // property of the traversal scope, which is what has to release the
        // pool reservation when the request ends.
        let options = |filter| SelectOptions { filter, ..walked() };

        // What the blob:none walk HOLDS, and what it SENDS. Both measured by
        // the same walker rather than predicted — `selection.objects.len()` is
        // not the first of them, which is the whole point of this test and was
        // worth finding out the hard way: on this fixture the walk sends 915
        // ids and holds 980, because `seen` carries the peeled tips and tags
        // the emitted vector never does.
        let (sent, held) = {
            let graph = Graph::new(&fixture.git, fixture.kind);
            let sent = select_objects(&graph, &[tip], &[], &options(Some(&blobless)))
                .expect("the blob:none clone")
                .objects
                .len();
            (sent, meter::peak())
        };
        assert!(
            held > sent + 100,
            "the fixture is not the shape this test needs: a blob:none clone \
             sent {sent} objects and held {held}, and the two have to be far \
             enough apart that a ceiling can sit between them"
        );

        // ── the ceiling at what it SENDS must refuse ────────────────────────
        let graph =
            Graph::new(&fixture.git, fixture.kind).with_walk_limit(Some(ceiling_for(sent)), None);
        let refused = select_objects(&graph, &[tip], &[], &options(Some(&blobless)))
            .expect_err("a ceiling the size of what this fetch sends must refuse it");
        match refused {
            Error::WalkTooLarge { held: bytes, limit } => {
                assert_eq!(
                    limit,
                    ceiling_for(sent),
                    "the refusal named another ceiling"
                );
                assert!(
                    bytes > limit,
                    "the refusal said it was holding {bytes} bytes against a \
                     ceiling of {limit}, which is not over it"
                );
            }
            other => panic!(
                "a blob:none fetch over the byte ceiling failed with {other}, not WalkTooLarge — \
                 so the ceiling is not what stopped it and this test is measuring another fault"
            ),
        }

        // ── and the ceiling at what it HOLDS must serve ─────────────────────
        let graph =
            Graph::new(&fixture.git, fixture.kind).with_walk_limit(Some(ceiling_for(held)), None);
        let served = select_objects(&graph, &[tip], &[], &options(Some(&blobless)))
            .expect("a ceiling above what this fetch holds must serve it");
        assert_eq!(
            served.objects.len(),
            sent,
            "the served arm sent {} objects where the unbounded one sent {sent}: a ceiling that \
             changes the ANSWER rather than refusing is worse than the fault being measured",
            served.objects.len()
        );
    }

    /// The ceiling is **off** when `walk_bytes` is `None`, and that is the arm
    /// that keeps the one above honest: without it, a `Graph` that refused
    /// every walk would satisfy the refusal half and the serve half would only
    /// be saying the number was large.
    #[test]
    fn no_ceiling_means_no_ceiling() {
        let fixture = Fixture::build(Kind::Sha1);
        let tip = fixture.extend(fixture.four, 50, "off");
        let graph = Graph::new(&fixture.git, fixture.kind).with_walk_limit(None, None);
        let all = select_objects(&graph, &[tip], &[], &walked())
            .expect("an unbounded walk serves")
            .objects
            .len();
        assert!(
            all > 50,
            "the fixture holds {all} objects, too few to mean anything"
        );
    }

    /// **The pool is what bounds the box**: two walks, one purse, and the
    /// second is refused because the first is holding it.
    ///
    /// # Why this is a different assertion from the one above
    ///
    /// A per-request ceiling that admits one large clone admits 256 of them,
    /// because `gunnar serve` counts transfers and not bytes. So the fact under
    /// test is not "a walk can be refused" — it is "a walk that would have been
    /// admitted on its own is refused because of what somebody else is holding",
    /// and that cannot be shown with one walk.
    ///
    /// Seen RED twice, and the two failures name different halves. With
    /// `Graph::charge_walk`'s pool arm disabled: *"the first walk reserved 0 of
    /// the 16777216 it holds, so there is nothing for the second to be refused
    /// by"*. With `Drop for Graph` releasing nothing: *"the pool did not get
    /// its bytes back, so this server refuses everything after enough
    /// clones"* — which is the failure mode where a pool closes for good.
    #[test]
    fn the_pool_refuses_a_walk_because_of_what_another_is_holding() {
        // Big enough to cross `WalkPool::FLOOR_BYTES`, and not one commit more.
        // The floor is 8 MiB, which is 21 845 ids at the charge rate, and
        // `extend` lays down a commit, a tree and a blob apiece. The shipped
        // constants are used rather than lowered for the test: a floor that is
        // only ever crossed by a number the test chose is not the floor the
        // server has.
        let history = crate::WalkPool::FLOOR_BYTES / crate::WALK_BYTES_PER_OBJECT / 3 + 500;
        let fixture = Fixture::build(Kind::Sha1);
        let tip = fixture.extend(fixture.four, history, "pool");

        let held = {
            let graph = Graph::new(&fixture.git, fixture.kind);
            select_objects(&graph, &[tip], &[], &walked()).expect("the whole clone");
            meter::peak()
        };
        assert!(
            ceiling_for(held) >= crate::WalkPool::FLOOR_BYTES,
            "this fixture holds {held} ids, {} bytes, which is under the pool's floor of {} \
             — the walk would never touch the pool and every arm below would pass against \
             nothing",
            ceiling_for(held),
            crate::WalkPool::FLOOR_BYTES
        );

        // A purse sized to exactly what one of these walks reserves: room for
        // the first and none for the second. Computed from the measurement
        // rather than typed, so the test cannot drift out of the shape it needs
        // when the fixture or the charge rate moves.
        let reservation = ceiling_for(held).next_multiple_of(crate::WalkPool::CHUNK_BYTES);
        let pool = crate::WalkPool::new(reservation);

        let outer =
            Graph::new(&fixture.git, fixture.kind).with_walk_limit(None, Some(pool.clone()));
        let holding = select_objects(&outer, &[tip], &[], &walked())
            .expect("the first walk fits the empty pool");
        assert!(!holding.objects.is_empty());
        assert_eq!(
            pool.in_flight_bytes(),
            reservation,
            "the first walk reserved {} of the {reservation} it holds, so there is nothing for \
             the second to be refused by",
            pool.in_flight_bytes()
        );

        // ── the second walk, on its own perfectly reasonable ────────────────
        let inner =
            Graph::new(&fixture.git, fixture.kind).with_walk_limit(None, Some(pool.clone()));
        let refused =
            select_objects(&inner, &[tip], &[], &walked()).expect_err("a full pool must refuse");
        match refused {
            Error::WalkPoolFull { limit, .. } => assert_eq!(limit, reservation),
            other => panic!(
                "a walk against a full pool failed with {other}, not WalkPoolFull — a client \
                 cannot tell a busy server from a request it must never send again"
            ),
        }
        assert_eq!(pool.refusals(), 1, "the refusal was not counted");
        drop(inner);

        // ── and the purse comes back ────────────────────────────────────────
        drop(outer);
        drop(holding);
        assert_eq!(
            pool.in_flight_bytes(),
            0,
            "the pool did not get its bytes back, so this server refuses everything after enough \
             clones"
        );
        let again =
            Graph::new(&fixture.git, fixture.kind).with_walk_limit(None, Some(pool.clone()));
        select_objects(&again, &[tip], &[], &walked())
            .expect("the same walk must serve once the first has finished");
    }

    /// A release the pool never handed out cannot wrap it shut.
    ///
    /// `saturating_sub` on a `usize` that went negative would be `usize::MAX`,
    /// and a pool whose `in_flight` is `usize::MAX` refuses every transfer
    /// forever. Seen RED with a plain `fetch_sub`.
    #[test]
    fn walk_pool_never_goes_negative() {
        let pool = crate::WalkPool::new(1024);
        pool.release(4096);
        assert_eq!(pool.in_flight_bytes(), 0);
        pool.try_reserve(1024)
            .expect("a pool at zero has all of itself");
        assert_eq!(pool.in_flight_bytes(), 1024);
    }

    /// A pool sized to this machine is not sized to a constant.
    #[test]
    fn a_pool_for_this_machine_reads_the_machine() {
        let pool = crate::WalkPool::for_this_machine();
        assert!(
            pool.limit_bytes() > 0,
            "a pool of zero bytes refuses every transfer"
        );
        if let Ok(text) = std::fs::read_to_string("/proc/meminfo") {
            let kb: usize = text
                .lines()
                .find_map(|l| l.strip_prefix("MemTotal:")?.trim().strip_suffix(" kB"))
                .and_then(|v| v.trim().parse().ok())
                .expect("MemTotal is in /proc/meminfo");
            assert_eq!(
                pool.limit_bytes(),
                kb * 1024 / 2,
                "the pool is not half of this machine's memory, so it is a constant wearing a \
                 measurement's name"
            );
        }
    }
}