Skip to main content

znippy_plugin_git/
resolve.rs

1//! Pack entries → **oids**, which is the one thing [`crate::pack_walk`] cannot
2//! tell you and the `objects` table is keyed by.
3//!
4//! This is the **indexer's** work, not the ack path's (§13.9: "the index is built
5//! after the ack, over a channel"). It inflates every entry, applies every delta
6//! chain and hashes the result, which is what `git index-pack` does and what
7//! makes it the expensive half of a push. Nothing here is called before a client
8//! is told its push landed.
9//!
10//! # What it keeps and what it throws away
11//!
12//! Peak memory is deliberately not "the whole history inflated". **An object is
13//! held only while an unresolved entry still has to read it**, and for nothing
14//! else:
15//!
16//! | object | held | until |
17//! |---|---|---|
18//! | a delta base | yes | its **last** dependent has consumed it |
19//! | everything else — blob, commit, tree, tag | **no** | it leaves through the [`PayloadSink`] and is dropped |
20//!
21//! The base set and, more to the point, **how many entries read each base** are
22//! known before a byte is inflated: the walk already collected every delta base,
23//! so the resolver starts with an exact reference count per base and drops a
24//! payload the moment that count reaches zero. The last dependent gets the base
25//! **moved** into it rather than cloned, so a linear delta chain — one base, one
26//! dependent, the ordinary shape of real history — copies no payload at all.
27//!
28//! ## What this table said before 2026-08-11, and what it cost
29//!
30//! It said commits, trees and tags were kept "because the `graph` and `reach`
31//! tables are built from them". They are not, and had not been since §14's
32//! exploded table became the fold's input — `absorb_one` reads
33//! `Resolved::index_entry()` and nothing else. Two copies of the object set were
34//! being held for a reader that no longer existed:
35//!
36//! * `content` kept every base **and** every typed object for the whole call.
37//!   Nothing ever removed a row, so "held only while something still needs them"
38//!   described an intention, not the code.
39//! * `Resolved::payload` kept a second copy of the same bytes, cloned in.
40//!
41//! MEASURED on oden 2026-08-11, `tests/resolve_peak_memory.rs`, peak **live**
42//! heap over `resolve_walked` against the pack's own inflated payload:
43//!
44//! | pack | before | after |
45//! |---|---:|---:|
46//! | forge-year-8a-240b-12c, 28 893 objects, 111.0 MB | 2.03x | **0.03x** |
47//! | stage-red-nobitmap, 12 328 objects, 343.5 MB | 1.34x (461.2 MB) | **0.02x** (5.9 MB) |
48//! | facett, 7 232 objects, 72.8 MB | 1.20x | **0.42x** |
49//! | znippy's own, 5 388 objects, 84.4 MB | 0.75x | **0.14x** |
50//!
51//! The worst pack in the corpus went 2.03x → 0.42x. It is `facett`, whose
52//! delta chains fan out widely enough that a real base working set is genuinely
53//! live at once — which is the bound behaving as intended rather than a residue.
54//!
55//! End to end, one `gunnar-server`, one `push --mirror`, anonymous RSS held
56//! afterwards (not peak — this memory was never given back):
57//!
58//! | fixture | before | after |
59//! |---|---:|---:|
60//! | 36 172 objects, 290 MB inflated, 5.9 MB pack | 575.7 MB | **185.1 MB** |
61//! | 352 396 objects, 2.89 GB inflated, 61.8 MB pack | 4480.1 MB | **439.9 MB** |
62//!
63//! The second row is the one that mattered: 10x the repository cost 10x the
64//! memory before and 2.4x after, so the resolver's footprint stopped tracking
65//! the history. What is left is bounded and named — redb's page cache
66//! (`ZNIPPY_GIT_REDB_CACHE_BYTES`, 64 MiB by default, measured at 129.3 MB held
67//! with an 8 MiB ceiling and 303.9 MB with 256 MiB), the exploded table's
68//! [`FLUSH_BYTES`](crate::exploded::FLUSH_BYTES) write buffer, and
69//! `Derived`'s in-memory graph, which is genuinely O(repository).
70//!
71//! # ⚠ The clone that got slower, and why it is not this code
72//!
73//! **Read this before "fixing" the reference counting.** `vs_forge_clone` on the
74//! 352 396-object fixture reported this change as a **+78 % CPU regression on
75//! every znippy arm**, and the natural reading — a base dropped at its "last
76//! use" has to be re-resolved when something later wants it again — is wrong. It
77//! is not even mechanically possible: an undercount cannot cause a re-resolve
78//! here, because there is no re-resolve to fall into. `resolve_walked` fails the
79//! whole pack by name.
80//!
81//! What it actually is, MEASURED on oden 2026-08-11, one gunnar-server, seed the
82//! fixture then clone it, steady state (clones 2-3), quiet box, only the
83//! allocator and this file varying:
84//!
85//! | resolver | allocator | clone CPU | anon after seed |
86//! |---|---|---:|---:|
87//! | before | glibc | 1.47 s | 4481 MB |
88//! | **after** | glibc | **2.62 s** | **438 MB** |
89//! | before | mimalloc | 1.18 s | 1767 MB |
90//! | **after** | **mimalloc** | **1.17 s** | **701 MB** |
91//!
92//! With a competent allocator the two resolvers are **identical** (1.18 against
93//! 1.17). The regression is entirely glibc `malloc`: the old resolver's retained
94//! 4.5 GB was accidentally acting as a **pre-warmed pool**, so the clone's own
95//! ~2 GB of allocations came off a free list instead of out of the kernel. Take
96//! the leak away and glibc charges for the memory it should have been charging
97//! for all along.
98//!
99//! Three independent measurements say the same thing, and none of them involve
100//! this file being slower:
101//!
102//! 1. **Bisect.** A build with *only* `Resolved::payload` removed — the
103//!    reference counting NOT applied, `content` still never pruned — reproduces
104//!    the whole regression (2.56–2.62 s). Adding the reference counting on top
105//!    then costs **nothing** (2.55–2.65 s) and takes anon-after-seed from
106//!    2350 MB to 438 MB. The counting is free; the *not leaking* is what glibc
107//!    punishes.
108//! 2. **Restart.** Restart the server after seeding, so the derived tables are
109//!    folded onto a fresh heap: before 2.20–2.22 s, after 2.20–2.24 s —
110//!    identical to within 1 %. The old resolver's 1.47 s is the outlier, not the
111//!    new one's 2.62 s.
112//! 3. **The resolver itself.** Over the same 6 real packs both revisions accept,
113//!    1.058 s before against 0.975 s after: this code is ~8 % *faster*.
114//!
115//! So there is no cheaper trade to find inside this module, and an LRU over
116//! bases would buy nothing. The lever, if the CPU is wanted back, is the
117//! allocator — and `LD_PRELOAD=libmimalloc.so.3` is enough to get it with no
118//! code change at all.
119//!
120//! # Where the payloads it throws away now go
121//!
122//! §14's exploded table is **eager** (decided 2026-08-08), so every object this
123//! module resolves has to reach [`crate::exploded::ExplodedTable`] — including
124//! the blobs the table above says are dropped. They still are: the payload
125//! leaves through a [`PayloadSink`] as it is produced, rather than being
126//! accumulated into the returned `Vec`. That is the difference between a peak
127//! footprint of "the bases plus the typed objects" and one of "the whole pack
128//! inflated at once", and eager resolution does not get to change it.
129//!
130//! A **thin** pack's `REF_DELTA` base is an object the client knows the server
131//! already has, and it is usually a blob. Resolving it needs that blob's
132//! *content*, which is exactly what the exploded table now holds — so the
133//! [`BaseSource`] can answer from it in one point lookup instead of re-resolving
134//! a whole pack. A thin pack whose external base is **still** not re-derivable
135//! is refused by name ([`ResolveError::MissingBase`]) rather than resolved by
136//! guesswork, unchanged.
137//!
138//! # The bounded base set (2026-08-30)
139//!
140//! The table above says a base is held until its last dependent. On
141//! torvalds/linux that set is **unbounded in principle** — 11.2 M deltas over
142//! 13.9 M objects, and nothing in the pack format promises a base's last
143//! dependent is near it. On the 16 GB appliance the indexer grew past the box
144//! and the kernel panicked (see [`crate::memguard`]). So the set now has a
145//! **byte ceiling** ([`ResolveLimits::base_budget`], default 25 % of
146//! `MemTotal`, `ZNIPPY_GIT_BASE_BUDGET_BYTES`):
147//!
148//! | event | what happens |
149//! |---|---|
150//! | a base would push the held bytes over the ceiling | the held base whose **next dependent is farthest** in the pack is dropped (Belady — the walk already lists every dependent's position, so "next use" is known, not guessed) |
151//! | the new base itself is the farthest | it is not held at all |
152//! | a dependent asks for a dropped base | it is **re-inflated from the mmapped pack** — its own chain re-applied — and counted in [`ResolveReport::reinflations`] |
153//! | the watchdog says the box is short | **every** held base is dropped, the sink and the row batch are flushed, and the worker waits |
154//!
155//! Same output, hard ceiling. The ceiling is `max(budget, largest single
156//! base)`: one object bigger than the budget is held while it is needed,
157//! because the alternative is re-inflating it once per dependent.
158//!
159//! # What is preallocated
160//!
161//! Two scratch buffers live across the whole pass: the delta instruction
162//! stream and the payload of an object nothing deltas against. An object that
163//! is a base gets its own `Vec` because it outlives the iteration; every other
164//! object — the overwhelming majority — is inflated into the reused buffer,
165//! hashed in place ([`GitHashKind::oid_of_object`], no canonical copy) and
166//! handed to the sink from there. Per object the hot path allocates the
167//! 20-byte oid of its row and nothing else.
168
169use std::collections::HashMap;
170use std::sync::Arc;
171use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
172
173use anyhow::{Result, anyhow, bail};
174
175use crate::exploded::{NoSink, PayloadSink};
176use crate::forest::{DeltaForest, EntrySet};
177use crate::index_layout::{IndexEntry, ObjType};
178use crate::memguard::MemWatchdog;
179use crate::object::{GitHashKind, GitObjectKind};
180use crate::pack_walk::{DeltaBase, PackEntry, PackWalk, walk};
181use crate::progress::{IndexProgress, Phase};
182
183/// Where an entry that deltas against something outside the pack gets its base.
184///
185/// Implemented by the store, which may or may not be able to answer — see the
186/// module docs on §14.
187pub trait BaseSource: Sync {
188    /// The object's type and inflated payload.
189    ///
190    /// `Ok(None)` means **this source does not hold that object** — the only
191    /// answer [`ResolveError::missing_base`] may be built from. `Err` means the
192    /// source *does* hold it and could not read it, and that is a different
193    /// fact with a different owner.
194    ///
195    /// # 🔴 It returned a bare `Option`, and the difference was thrown away
196    ///
197    /// `impl BaseSource for Absorber` ended in `.ok().flatten()`: a store that
198    /// failed to reconstruct a base it holds answered exactly like a store that
199    /// never had it, and the resolver then refused the *pushed pack* with *"this
200    /// repository genuinely does not have that object"*. MEASURED on oden
201    /// 2026-08-30 (lane AI): 21 of 140 ordinary pushes refused with that
202    /// sentence about an object sitting in an absorbed pack one lookup away —
203    /// see [`MAX_DELTA_LINKS`] for what was actually wrong and
204    /// `crate::thin_storm` for the red. A client cannot act on a diagnosis that
205    /// names the wrong party; neither can an operator.
206    fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>>;
207}
208
209/// A source that has nothing. The right one for a self-contained pack, and it
210/// makes "this pack was thin" an error rather than a silent success.
211pub struct NoBases;
212
213impl BaseSource for NoBases {
214    fn content(&self, _oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
215        Ok(None)
216    }
217}
218
219/// One resolved object: **exactly what the `objects` table stores**, and no
220/// payload.
221///
222/// It carried an `Option<Vec<u8>>` payload until 2026-08-11, for a reader that
223/// stopped existing when §14's exploded table became the fold's input. Every
224/// production caller uses [`index_entry`](Self::index_entry) and nothing else —
225/// `absorb_one` builds index rows, and the one other caller discards the vector
226/// entirely and reads its sink. Content has one home now
227/// ([`PayloadSink`], LAW 5) and this struct is the row, so a 343 MB pack's rows
228/// cost the length of the pack's entry list rather than the length of its
229/// history.
230#[derive(Debug, Clone)]
231pub struct Resolved {
232    pub oid: Vec<u8>,
233    /// Byte extent of the **stored, verbatim** entry.
234    pub offset: u64,
235    pub len: u64,
236    /// The type the entry carries in the pack — which for a delta is
237    /// `OfsDelta` / `RefDelta`, exactly as `.idx` cannot tell you and this index
238    /// exists to record.
239    pub stored_type: ObjType,
240    /// The type the chain resolves to. Always one of the four real git types.
241    pub kind: GitObjectKind,
242    /// Inflated, **post-delta-resolution** size — the fact `.idx` and `.rev`
243    /// together still cannot answer.
244    pub uncompressed_size: u64,
245    /// The `objects.delta_base` column: an archive **offset**, `0` for none.
246    pub delta_base: u64,
247}
248
249impl Resolved {
250    /// The row [`crate::read_stack::ObjectReadStack::append`] takes.
251    ///
252    /// `delta_base` now has a column of its own in [`IndexEntry`] and rides
253    /// across in it — it was carried on this struct alone until that column
254    /// existed, and was deliberately never smuggled into `uncompressed_size` or
255    /// any other field on the way. **This is the only place it is written into
256    /// an index row** (LAW 5): [`resolve_walked`] computes it once, in the
257    /// archive's coordinate space, and nothing downstream recomputes it from the
258    /// walk.
259    pub fn index_entry(&self) -> IndexEntry {
260        IndexEntry {
261            oid: self.oid.clone(),
262            offset: self.offset,
263            len: self.len,
264            obj_type: self.stored_type,
265            uncompressed_size: self.uncompressed_size,
266            delta_base: self.delta_base,
267        }
268    }
269}
270
271/// The one failure that is a *decision* rather than a corrupt pack.
272#[derive(Debug)]
273pub struct ResolveError;
274
275impl ResolveError {
276    /// A thin pack's external base, named.
277    pub fn missing_base(oid: &[u8]) -> anyhow::Error {
278        anyhow!(
279            "this pack is thin: it deltas against {} which is not in it, and no base source could \
280             produce that object's content — §14's exploded table does not hold it and it could \
281             not be re-derived from the verbatim packs either, so this repository genuinely does \
282             not have that object. The pack is refused by name rather than half-resolved",
283            hex::encode(oid)
284        )
285    }
286
287    /// The worker was told to stop (the repository is being deleted).
288    pub fn cancelled() -> anyhow::Error {
289        anyhow!("index work cancelled: the repository is being deleted")
290    }
291
292    /// **This box cannot index this pack**, said out loud instead of waited out.
293    ///
294    /// The memory watchdog paused, dropped everything droppable, and still could
295    /// not get back over its floor — either for
296    /// [`MemWatchdog::pause_deadline`](crate::memguard::MemWatchdog::pause_deadline)
297    /// or because the flush freed nothing three times running. Lane P2's rig
298    /// took the other branch, which was to wait forever: 1 pause, 0 resumes,
299    /// 1h45m41s, and a client that never got an ack. A refusal by name is a
300    /// worse outcome for this one push and a far better one for the operator,
301    /// who can now read the row and give the box more memory.
302    pub fn out_of_memory(reason: &str, available: u64, budget: u64, done: u64, total: u64) -> anyhow::Error {
303        anyhow!(
304            "this box cannot index this pack: the memory watchdog gave up ({reason}) with {} \
305             available against a {} floor, after {done} of {total} objects. Everything droppable \
306             was dropped and the floor was still breached — what holds the memory is the \
307             resolver's own per-object state, which is proportional to the pack. Give the box \
308             more RAM, lower ZNIPPY_GIT_MEM_BUDGET_PCT, or push a smaller pack. The pack's bytes \
309             are durable and untouched; nothing partial was indexed.",
310            crate::memguard::fmt_bytes(available),
311            crate::memguard::fmt_bytes(budget)
312        )
313    }
314}
315
316/// Default share of `MemTotal` the held base set may take.
317pub const DEFAULT_BASE_BUDGET_PCT: u64 = 25;
318/// Default rows per batch handed to the row sink.
319pub const DEFAULT_ROW_BATCH: usize = 262_144;
320/// **Ceiling on how far the workers may run ahead of the ordered stream**, in
321/// bytes of rows and of the payloads the sink asked to be given. Past it a
322/// worker waits for the emitter instead of buffering the pack — see
323/// [`ResolveLimits::lookahead_bytes`].
324pub const DEFAULT_LOOKAHEAD_BYTES: u64 = 32 << 20;
325/// A ceiling on the worker count no environment variable may cross. A pack is
326/// resolved by the box, not by a number somebody typed.
327pub const MAX_WORKERS: usize = 256;
328
329/// The knobs and hooks the indexer hands the resolver. [`ResolveLimits::none`]
330/// is the old behaviour: unbounded bases, no watchdog, no progress.
331#[derive(Clone)]
332pub struct ResolveLimits<'a> {
333    /// Ceiling on held delta-base bytes; `0` = unbounded.
334    pub base_budget: u64,
335    /// Rows per batch to the row sink.
336    pub row_batch: usize,
337    /// **How many subtrees are walked at once.** `1` is the serial resolver and
338    /// the reference every other value is compared against; the output is the
339    /// same stream at any value (see `the_same_pack_resolves_to_the_same_bytes_at_1_2_and_4_workers`).
340    pub workers: usize,
341    /// Bytes of resolved-but-not-yet-emitted objects the workers may hold. `0`
342    /// takes [`DEFAULT_LOOKAHEAD_BYTES`].
343    pub lookahead_bytes: u64,
344    /// Objects between two checkpoints (cancel flag, watchdog).
345    pub check_every: u64,
346    pub watchdog: Option<Arc<MemWatchdog>>,
347    pub cancel: Option<&'a AtomicBool>,
348    pub progress: Option<&'a IndexProgress>,
349    /// Names the repository in log lines.
350    pub label: &'a str,
351    /// Called once per watchdog pause and again every re-flush: the owner drops
352    /// whatever it holds that can be rebuilt (the store's caches) and returns
353    /// **how much it dropped**. Zero means the owner had nothing to give, which
354    /// is the fact `index.spilled` reported 627 times and nobody acted on —
355    /// see [`crate::memguard::PauseEnd::NothingToGive`].
356    pub on_pause: Option<&'a (dyn Fn() -> u64 + Sync)>,
357}
358
359impl<'a> ResolveLimits<'a> {
360    /// Unbounded, unwatched, unreported — what a test or a one-off resolve uses.
361    pub fn none() -> Self {
362        Self {
363            base_budget: 0,
364            row_batch: DEFAULT_ROW_BATCH,
365            workers: 1,
366            lookahead_bytes: DEFAULT_LOOKAHEAD_BYTES,
367            check_every: crate::memguard::CHECK_EVERY_OBJECTS,
368            watchdog: None,
369            cancel: None,
370            progress: None,
371            label: "",
372            on_pause: None,
373        }
374    }
375
376    /// The indexer's: budget and batch from the environment (once per
377    /// process), the process-wide watchdog, this store's cancel flag and
378    /// progress counter.
379    pub fn for_indexer(
380        label: &'a str,
381        cancel: &'a AtomicBool,
382        progress: &'a IndexProgress,
383        watchdog: Arc<MemWatchdog>,
384        on_pause: &'a (dyn Fn() -> u64 + Sync),
385    ) -> Self {
386        Self {
387            base_budget: base_budget_from_env(),
388            row_batch: row_batch_from_env(),
389            workers: workers_from_env(),
390            lookahead_bytes: lookahead_from_env(),
391            check_every: crate::memguard::CHECK_EVERY_OBJECTS,
392            watchdog: Some(watchdog),
393            cancel: Some(cancel),
394            progress: Some(progress),
395            label,
396            on_pause: Some(on_pause),
397        }
398    }
399}
400
401/// `ZNIPPY_GIT_BASE_BUDGET_BYTES`, else 25 % of `MemTotal`, else 1 GiB when
402/// the box cannot be measured. Read once per process.
403pub fn base_budget_from_env() -> u64 {
404    use std::sync::OnceLock;
405    static B: OnceLock<u64> = OnceLock::new();
406    *B.get_or_init(|| {
407        if let Some(b) = crate::arms::read_env(crate::arms::ENV_BASE_BUDGET_BYTES)
408            .and_then(|v| v.parse::<u64>().ok())
409        {
410            return b;
411        }
412        let total = crate::memguard::mem_total();
413        if total == 0 { 1 << 30 } else { total / 100 * DEFAULT_BASE_BUDGET_PCT }
414    })
415}
416
417/// `ZNIPPY_GIT_INDEX_BATCH_ROWS`, else [`DEFAULT_ROW_BATCH`]. Read once per process.
418pub fn row_batch_from_env() -> usize {
419    use std::sync::OnceLock;
420    static B: OnceLock<usize> = OnceLock::new();
421    *B.get_or_init(|| {
422        crate::arms::read_env(crate::arms::ENV_INDEX_BATCH_ROWS)
423            .and_then(|v| v.parse::<usize>().ok())
424            .filter(|n| *n > 0)
425            .unwrap_or(DEFAULT_ROW_BATCH)
426    })
427}
428
429/// `ZNIPPY_GIT_RESOLVE_WORKERS`, else one per core. Read once per process.
430///
431/// **The forest is the parallelism** ([`crate::forest`]): a base and its
432/// dependents are a tree, and two trees share nothing, so N workers is N
433/// independent depth-first walks over one mmapped pack. It is not N copies of
434/// the pack's O(pack) structures — the walk, the forest and the oid table are
435/// read-only and shared — and it is not N base budgets either: the held-base
436/// ceiling is one atomic across all of them (LAW 5).
437pub fn workers_from_env() -> usize {
438    use std::sync::OnceLock;
439    // The ENVIRONMENT is read once (the read-once law); the embedder's default is
440    // live, because a box learns it is a twin AFTER this process started.
441    static W: OnceLock<Option<usize>> = OnceLock::new();
442    let from_env = *W.get_or_init(|| {
443        crate::arms::read_env(crate::arms::ENV_RESOLVE_WORKERS)
444            .and_then(|v| v.parse::<usize>().ok())
445            .filter(|n| *n > 0)
446    });
447    from_env
448        .unwrap_or_else(|| match RESOLVE_WORKERS_DEFAULT.load(Ordering::Relaxed) {
449            0 => std::thread::available_parallelism().map(|x| x.get()).unwrap_or(1),
450            n => n,
451        })
452        .clamp(1, MAX_WORKERS)
453}
454
455/// What [`workers_from_env`] answers when [`crate::arms::ENV_RESOLVE_WORKERS`] is unset.
456/// `0` is "one per core" — the shipped default — and not a worker count, because
457/// [`workers_from_env`] clamps to at least one and a zero-wide resolver is no resolver.
458static RESOLVE_WORKERS_DEFAULT: AtomicUsize = AtomicUsize::new(0);
459
460/// **Set the process-wide default resolver width** for every resolve started AFTER this call
461/// (an explicit [`crate::arms::ENV_RESOLVE_WORKERS`] still wins). `0` restores one per core.
462/// The twin of [`crate::arms::set_graph_cache_default`], for an embedder that learns its role
463/// at runtime and cannot set the environment — edition 2024 makes `set_var` unsafe, and gunnar
464/// forbids `unsafe`. MEASURED (gunnar `twin_snapshot_memory.rs`, 2026-09-15): one worker per
465/// core, each inflating a whole blob, is +133 MiB of RssAnon for a 512 MiB snapshot install on
466/// a many-core box; two workers with a 16 MiB lookahead is +36 MiB, under gunnar's 64 MiB
467/// twin ceiling.
468pub fn set_resolve_workers_default(workers: usize) {
469    RESOLVE_WORKERS_DEFAULT.store(workers, Ordering::Relaxed);
470}
471
472/// `ZNIPPY_GIT_RESOLVE_LOOKAHEAD_BYTES`, else [`DEFAULT_LOOKAHEAD_BYTES`].
473/// Read once per process.
474pub fn lookahead_from_env() -> u64 {
475    use std::sync::OnceLock;
476    // Read-once environment, live embedder default — see [`workers_from_env`].
477    static L: OnceLock<Option<u64>> = OnceLock::new();
478    let from_env = *L.get_or_init(|| {
479        crate::arms::read_env(crate::arms::ENV_RESOLVE_LOOKAHEAD_BYTES)
480            .and_then(|v| v.parse::<u64>().ok())
481            .filter(|n| *n > 0)
482    });
483    from_env.unwrap_or_else(|| match RESOLVE_LOOKAHEAD_DEFAULT.load(Ordering::Relaxed) {
484        0 => DEFAULT_LOOKAHEAD_BYTES,
485        n => n,
486    })
487}
488
489/// What [`lookahead_from_env`] answers when [`crate::arms::ENV_RESOLVE_LOOKAHEAD_BYTES`] is
490/// unset. `0` is [`DEFAULT_LOOKAHEAD_BYTES`] and not "no lookahead", which is the same
491/// reading the environment door gives a `0` there.
492static RESOLVE_LOOKAHEAD_DEFAULT: AtomicU64 = AtomicU64::new(0);
493
494/// **Set the process-wide default resolver lookahead**, in bytes, for every resolve started
495/// AFTER this call (an explicit [`crate::arms::ENV_RESOLVE_LOOKAHEAD_BYTES`] still wins).
496/// `0` restores [`DEFAULT_LOOKAHEAD_BYTES`]. The twin of
497/// [`crate::arms::set_graph_cache_default`] — see [`set_resolve_workers_default`] for the
498/// measurement the pair is for.
499pub fn set_resolve_lookahead_default(bytes: u64) {
500    RESOLVE_LOOKAHEAD_DEFAULT.store(bytes, Ordering::Relaxed);
501}
502
503/// What one resolve did — the audit line's numbers, all applied output.
504#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
505pub struct ResolveReport {
506    pub rows: u64,
507    /// Bases dropped for the ceiling and inflated again for a later dependent.
508    pub reinflations: u64,
509    /// Bases dropped for the ceiling (each may cost 0..n re-inflations).
510    pub evictions: u64,
511    /// Bases dropped because the watchdog said the box was short.
512    pub watchdog_drops: u64,
513    /// Most bytes of bases held at once.
514    pub peak_base_bytes: u64,
515    /// Watchdog pauses inside this pass.
516    pub pauses: u64,
517    pub paused_ms: u64,
518    /// Row batches handed to the sink.
519    pub batches: u64,
520    /// Times a worker waited for the ordered stream instead of running further
521    /// ahead of it. A positive number is the lookahead ceiling doing its job;
522    /// a large one beside a poor speed-up is the ceiling being the bottleneck.
523    pub stalls: u64,
524    /// How many subtrees were walked at once. `1` is the serial resolver.
525    pub workers: u32,
526    /// Times a worker gave its core back because the box was short of memory
527    /// rather than pausing the whole pack. See `Round::checkpoint`.
528    pub shed: u64,
529    /// The watchdog gave up rather than resumed — the pass is being failed by
530    /// name. See [`ResolveError::out_of_memory`].
531    pub gave_up: bool,
532}
533
534impl ResolveReport {
535    /// Fold one round's numbers into the pack's. Sums, and the two extremes
536    /// taken as extremes — a merge that does not depend on which worker
537    /// produced what, so the audit line is the same at any worker count.
538    pub fn absorb(&mut self, r: ResolveReport) {
539        self.rows += r.rows;
540        self.reinflations += r.reinflations;
541        self.evictions += r.evictions;
542        self.watchdog_drops += r.watchdog_drops;
543        self.peak_base_bytes = self.peak_base_bytes.max(r.peak_base_bytes);
544        self.pauses += r.pauses;
545        self.paused_ms += r.paused_ms;
546        self.batches += r.batches;
547        self.stalls += r.stalls;
548        self.shed += r.shed;
549        self.workers = self.workers.max(r.workers);
550        self.gave_up |= r.gave_up;
551    }
552}
553
554/// Resolve every entry of `pack` to an oid, **keeping no payloads**.
555///
556/// `archive_offset` is where the pack's first byte lives in the archive, so the
557/// offsets and `delta_base`s that come back are already in the archive's
558/// coordinate space and no caller has to remember to shift them.
559///
560/// The indexer does **not** come through here — it calls
561/// [`resolve_streaming`] with the exploded table as its sink, which is what
562/// makes §14's side table fall out of the same pass rather than out of a second
563/// walk.
564pub fn resolve(
565    pack: &[u8],
566    hash: GitHashKind,
567    archive_offset: u64,
568    bases: &dyn BaseSource,
569) -> Result<Vec<Resolved>> {
570    let w = walk(pack, hash.oid_len())?;
571    resolve_walked(pack, &w, hash, archive_offset, bases, &NoSink)
572}
573
574/// Same, for a caller that already walked the pack (`put` does, for the closure
575/// check) and must not walk it twice — and for the one that wants the payloads.
576///
577/// `sink` is handed **every** object exactly once, in resolution order, with the
578/// payload its chain resolves to. Pass [`NoSink`] to keep nothing. This is the
579/// one pass §14's exploded table is built from: no second walk, no second
580/// inflate, no second resolver (LAW 5).
581///
582/// Rows come back in **entry order**. Unbounded bases (the old behaviour): a
583/// caller that wants the ceiling, the watchdog and streaming rows is the
584/// indexer, and it uses [`resolve_streaming`].
585pub fn resolve_walked(
586    pack: &[u8],
587    w: &PackWalk,
588    hash: GitHashKind,
589    archive_offset: u64,
590    bases: &dyn BaseSource,
591    sink: &dyn PayloadSink,
592) -> Result<Vec<Resolved>> {
593    let mut out: Vec<Resolved> = Vec::with_capacity(w.entries.len());
594    let limits = ResolveLimits {
595        row_batch: w.entries.len().max(1),
596        ..ResolveLimits::none()
597    };
598    resolve_streaming(pack, w, hash, archive_offset, bases, sink, limits, &mut |rows| {
599        out.extend(rows.drain(..));
600        Ok(())
601    })?;
602    // Resolution order is entry order except for the ref-deltas a later round
603    // picked up; offsets are unique and ascending in entry order, so this is
604    // exactly the order the old `out[i]` array had.
605    out.sort_unstable_by_key(|r| r.offset);
606    Ok(out)
607}
608
609/// **The resolver**: bounded bases, watchdog checkpoints, rows in batches.
610///
611/// `rows` is called with each full batch (and the final partial one) and may
612/// drain the vector; the same allocation is reused for the next batch. Nothing
613/// about the rows or the sink calls depends on the batch size — it is purely
614/// how much the caller holds at once.
615#[allow(clippy::too_many_arguments)]
616pub fn resolve_streaming(
617    pack: &[u8],
618    w: &PackWalk,
619    hash: GitHashKind,
620    archive_offset: u64,
621    bases: &dyn BaseSource,
622    sink: &dyn PayloadSink,
623    limits: ResolveLimits<'_>,
624    rows: &mut (dyn FnMut(&mut Vec<Resolved>) -> Result<()> + Send),
625) -> Result<ResolveReport> {
626    let mut r = Resolver::new(pack, w, hash, archive_offset, bases, sink, limits)?;
627    r.run(rows)?;
628    Ok(r.report)
629}
630
631
632// ════════════════════════════════════════════════════════════════════════════
633// THE FOREST DRIVER — one subtree per unit, N workers, ONE ordered stream out
634// ════════════════════════════════════════════════════════════════════════════
635
636/// How many objects one worker resolves before it hands them to the ordered
637/// emitter: one lock per few thousand objects instead of one per object.
638const HANDOVER_ROWS: usize = 4096;
639/// …or this many bytes of them, whichever comes first.
640const HANDOVER_BYTES: u64 = 1 << 20;
641/// **How large a scratch buffer may stay between objects.**
642///
643/// `delta` and `leaf` are reused for the whole pass, which is what makes the
644/// hot path allocate nothing — and a `Vec` never gives capacity back, so one
645/// 90 MB blob leaves a 90 MB buffer standing for the remaining thirteen
646/// million objects. Single-threaded that was one such buffer; with a worker per
647/// core it is `2 x workers` of them, and on a 16 GB appliance that is the
648/// difference between finishing a kernel mirror and pausing at 98 %.
649const SCRATCH_KEEP_BYTES: usize = 1 << 20;
650/// What one small heap block costs beyond the bytes asked for — header plus
651/// size-class rounding. Deliberately generous: the lookahead ceiling is a
652/// memory bound, and a memory bound that under-counts is not one.
653const ALLOC_OVERHEAD: usize = 32;
654/// How many objects one pass of the drain takes out of the emit buffer before
655/// it lets the writers back in — the lock-hold bound.
656const DRAIN_RUN: usize = 8192;
657
658/// One frame of a worker's depth-first path: a resolved base, held **only**
659/// while its own subtree is still being walked.
660struct Frame {
661    entry: u32,
662    kind: GitObjectKind,
663    /// Empty when `evicted`.
664    payload: Vec<u8>,
665    evicted: bool,
666    next_child: u32,
667}
668
669/// One resolved object on its way out: the row always, the payload only when
670/// the sink said it would read one ([`PayloadSink::wants`]).
671struct Emit {
672    kind: GitObjectKind,
673    row: Resolved,
674    payload: Option<Vec<u8>>,
675}
676
677impl Emit {
678    /// What holding this costs, for the lookahead ceiling. A count of objects
679    /// cannot tell one commit from one 40 MB blob, so the ceiling is in bytes —
680    /// and the bytes are the ones the ALLOCATOR charges, not the ones the
681    /// struct declares. `ALLOC_OVERHEAD` is the rest of what a 20-byte oid
682    /// costs: the header and the size-class rounding of its own heap block,
683    /// plus this row's share of the `Vec` it sits in. Undercounting it is how a
684    /// 128 MiB ceiling holds half a gigabyte.
685    fn weight(&self) -> u64 {
686        (std::mem::size_of::<Emit>()
687            + ALLOC_OVERHEAD
688            + self.row.oid.len()
689            + self.payload.as_ref().map_or(0, |p| p.len() + ALLOC_OVERHEAD)) as u64
690    }
691}
692
693/// A worker's own scratch: never shared, never locked, reused across the
694/// subtrees one thread walks.
695#[derive(Default)]
696struct Scratch {
697    /// Which unit (root) of this round the worker is inside — the key its
698    /// output is handed over under.
699    unit: usize,
700    delta: Vec<u8>,
701    leaf: Vec<u8>,
702    stack: Vec<Frame>,
703    out: Vec<Emit>,
704    out_bytes: u64,
705    found_refs: Vec<(Vec<u8>, u64)>,
706}
707
708std::thread_local! {
709    /// Per **thread**, not per subtree: a kernel-sized pack has 2.7 M roots and
710    /// allocating two grown buffers for each of them is 2.7 M mallocs the walk
711    /// does not need.
712    static SCRATCH: std::cell::RefCell<Scratch> = std::cell::RefCell::new(Scratch::default());
713}
714
715/// What one root has produced and the drain has not yet handed on, in
716/// depth-first order, and whether the walk of that root is finished.
717///
718/// **One `Vec`, not a queue of them.** The median subtree in a real pack is
719/// ONE object (`examples/forest_stats` on torvalds/linux: p50 1, p90 5), so a
720/// `VecDeque<Vec<Emit>>` per root was two heap allocations and a deque header
721/// to carry a single 20-byte oid — the per-root bookkeeping cost more than the
722/// object it carried, 2.7 M times.
723#[derive(Default)]
724struct RootOut {
725    rows: Vec<Emit>,
726    done: bool,
727}
728
729/// The emit buffer, keyed by **root index**.
730///
731/// # 🔴 It was keyed by ENTRY index, and on a real pack that was a serialiser
732///
733/// Emitting in entry order is the obvious way to make N workers produce the
734/// stream one worker would have: the objects come out sorted by where they sit
735/// in the pack, whoever resolved them. It is also unusable, and the pack says
736/// why. MEASURED on `torvalds/linux` (`examples/forest_stats`), 13 948 244
737/// entries, 2 732 372 roots:
738///
739/// | | subtree **size** | subtree **index span** |
740/// |---|---:|---:|
741/// | p50 | 1 | 0 |
742/// | p90 | 5 | **5 251 404** |
743/// | p99 | 70 | **11 511 634** |
744/// | max | 11 789 | **12 434 450** |
745///
746/// The subtrees are tiny — perfect parallel work — but a five-object subtree
747/// routinely **spans five million entries**, because `pack-objects` puts a base
748/// near its first dependent and the rest of the dependents wherever they fall.
749/// So an entry-ordered cursor is pinned by any subtree in flight whose span
750/// reaches the top of the pack: every other worker's output buffers behind it,
751/// the lookahead ceiling closes, and the workers wait. MEASURED, the kernel
752/// pack at the appliance's 4-core shape: **1 worker 1454.7 s, 2 workers
753/// 1572.9 s** — *slower*, at 1.24 cores of CPU, with 286 686 stalls.
754///
755/// Keyed by root, the cursor moves at the rate roots finish, and roots are
756/// claimed in ascending order, so it follows the workers instead of pinning
757/// them. The order is still fixed and still worker-count independent: roots
758/// ascending, and inside a root the depth-first sequence the children lists
759/// (ascending) determine.
760#[derive(Default)]
761struct Pending {
762    map: std::collections::BTreeMap<usize, RootOut>,
763    bytes: u64,
764    /// **The units a worker is inside right now.** The lookahead ceiling needs
765    /// it: gatling claims a unit *before* the worker runs, so "wait until the
766    /// stream catches up" would otherwise be said by the very worker the stream
767    /// is waiting for. See [`Round::backpressure`].
768    in_flight: std::collections::BTreeSet<usize>,
769    /// How many of those units belong to a worker that has **given its core
770    /// back** for the memory watchdog and is waiting rather than resolving.
771    /// `in_flight.len() - parked` is therefore how many cores this pack is
772    /// actually using, which is the number the last worker standing needs
773    /// before it decides that pausing is the only option left.
774    parked: usize,
775}
776
777/// The ordered end of the pipe. One worker at a time holds it, so the sink and
778/// the row callback see exactly the sequence one worker would have made.
779/// `cursor` is an index into this round's **roots**.
780struct Drain<'r> {
781    cursor: usize,
782    batch: Vec<Resolved>,
783    rows: &'r mut (dyn FnMut(&mut Vec<Resolved>) -> Result<()> + Send),
784}
785
786/// A base this round was promised from outside the pack, probed **once** while
787/// the round was planned so that which entries the round covers is decided
788/// before a byte is inflated.
789struct RoundBases<'r, 'w, 'a> {
790    external: &'r HashMap<&'w [u8], (GitObjectKind, Vec<u8>)>,
791    inner: &'a dyn BaseSource,
792}
793
794impl BaseSource for RoundBases<'_, '_, '_> {
795    fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
796        if let Some((k, p)) = self.external.get(oid) {
797            return Ok(Some((*k, p.clone())));
798        }
799        self.inner.content(oid)
800    }
801}
802
803/// **One round of the fixpoint, across N workers.**
804///
805/// Everything in here is either immutable for the length of the round or an
806/// atomic/lock, because a `&Round` is what every worker holds.
807struct Round<'p, 'w, 'a, 'r> {
808    pack: &'p [u8],
809    w: &'w PackWalk,
810    forest: &'r DeltaForest,
811    hash: GitHashKind,
812    oid_len: usize,
813    archive_offset: u64,
814    bases: &'a dyn BaseSource,
815    sink: &'a dyn PayloadSink,
816    limits: &'r ResolveLimits<'a>,
817    /// The oids some `REF_DELTA` in this pack names. Read-only here.
818    ref_wanted: &'r std::collections::HashSet<&'w [u8]>,
819    /// Where a named oid turned out to live, from the rounds **before** this one.
820    oid_to_offset: &'r HashMap<Vec<u8>, u64>,
821    external: &'r HashMap<&'w [u8], (GitObjectKind, Vec<u8>)>,
822    roots: &'r [u32],
823    in_round: &'r EntrySet,
824    workers: usize,
825    lookahead: u64,
826
827    pend: std::sync::Mutex<Pending>,
828    cv: std::sync::Condvar,
829    drain: std::sync::Mutex<Drain<'r>>,
830    pause_gate: std::sync::Mutex<()>,
831    stop: AtomicBool,
832    err: std::sync::Mutex<Option<(usize, anyhow::Error)>>,
833    since_check: std::sync::atomic::AtomicU64,
834
835    rows: std::sync::atomic::AtomicU64,
836    reinflations: std::sync::atomic::AtomicU64,
837    evictions: std::sync::atomic::AtomicU64,
838    watchdog_drops: std::sync::atomic::AtomicU64,
839    /// **The whole job's** held base bytes, across every worker — lane Q's one
840    /// budget, not one per thread (LAW 5).
841    live_path: std::sync::atomic::AtomicU64,
842    peak_path: std::sync::atomic::AtomicU64,
843    pauses: std::sync::atomic::AtomicU64,
844    paused_ms: std::sync::atomic::AtomicU64,
845    batches: std::sync::atomic::AtomicU64,
846    stalls: std::sync::atomic::AtomicU64,
847    shed: std::sync::atomic::AtomicU64,
848    gave_up: AtomicBool,
849    found_refs: std::sync::Mutex<Vec<(Vec<u8>, u64)>>,
850}
851
852impl<'p, 'w, 'a, 'r> Round<'p, 'w, 'a, 'r> {
853    // ── one subtree ─────────────────────────────────────────────────────────
854
855    /// The unit of work: one root, its whole subtree, in a fixed depth-first
856    /// order.
857    fn run_root(&self, ri: usize, s: &mut Scratch) -> Result<()> {
858        // A previous unit that failed may have left a path standing; nothing
859        // else can, because a finished walk pops every frame.
860        if !s.stack.is_empty() {
861            self.drop_path(s);
862            s.stack.clear();
863        }
864        s.unit = ri;
865        let root = self.roots[ri];
866        let r = (|| {
867            let (kind, payload) = self.root_content(root)?;
868            self.emit(root, kind, &payload, s)?;
869            if self.forest.children(root).is_empty() {
870                drop(payload);
871            } else {
872                self.push_frame(root, kind, payload, s);
873                self.walk(s)?;
874            }
875            Ok(())
876        })();
877        // **The unit is closed even when it failed**, so the drain is never
878        // left waiting on a root that will never say it is finished.
879        self.flush_scratch(s, true)?;
880        r
881    }
882
883    /// A root's own content. `DeltaBase::None` inflates; a `REF_DELTA` root
884    /// applies its delta to a base this round already promised — either an
885    /// entry a previous round placed, or one the [`BaseSource`] answered for.
886    fn root_content(&self, root: u32) -> Result<(GitObjectKind, Vec<u8>)> {
887        let e = &self.w.entries[root as usize];
888        match &e.delta_base {
889            DeltaBase::None => {
890                let kind = whole_kind(e.obj_type)?;
891                let mut out = Vec::with_capacity(e.uncompressed_size as usize);
892                inflate_into(self.pack, e, self.oid_len, &mut out)?;
893                Ok((kind, out))
894            }
895            DeltaBase::Ref(oid) => {
896                let (kind, base) = match self.oid_to_offset.get(oid.as_slice()).copied() {
897                    Some(o) => {
898                        let be = self
899                            .w
900                            .entries
901                            .binary_search_by_key(&o, |x| x.offset)
902                            .map_err(|_| {
903                                anyhow!("no entry starts at offset {o} — the pack is corrupt")
904                            })?;
905                        self.materialize(be as u32)?
906                    }
907                    None => match self.external.get(oid.as_slice()) {
908                        Some((k, p)) => (*k, p.clone()),
909                        None => return Err(ResolveError::missing_base(oid)),
910                    },
911                };
912                let mut delta = Vec::with_capacity(e.uncompressed_size as usize);
913                inflate_into(self.pack, e, self.oid_len, &mut delta)?;
914                let mut out = Vec::new();
915                apply_delta_into(&base, &delta, &mut out)?;
916                Ok((kind, out))
917            }
918            DeltaBase::Offset(_) => bail!(
919                "the entry at offset {} deltas against an offset, so it is not a root of a \
920                 resolve round — the forest is inconsistent with the walk",
921                e.offset
922            ),
923        }
924    }
925
926    /// **The walk.** An explicit stack, not recursion: a verbatim archive grows
927    /// one delta link per push ([`MAX_DELTA_LINKS`]), so a legitimate chain has
928    /// no depth this code may assume.
929    fn walk(&self, s: &mut Scratch) -> Result<()> {
930        while !s.stack.is_empty() {
931            let fi = s.stack.len() - 1;
932            let entry = s.stack[fi].entry;
933            let nkids = self.forest.children(entry).len();
934            let nc = s.stack[fi].next_child as usize;
935            if nc == nkids {
936                self.pop_frame(s);
937                continue;
938            }
939            let c = self.forest.children(entry)[nc];
940            s.stack[fi].next_child += 1;
941            let last = nc + 1 == nkids;
942            let kind = s.stack[fi].kind;
943
944            // The base for this child: moved out on the last one, cloned
945            // otherwise, re-inflated if the ceiling took it.
946            let base: Vec<u8> = if s.stack[fi].evicted {
947                let (_, p) = self.materialize(entry)?;
948                self.reinflations.fetch_add(1, Ordering::Relaxed);
949                if let Some(pr) = self.limits.progress {
950                    pr.add_reinflations(1);
951                }
952                if !last {
953                    let n = p.len() as u64;
954                    let budget = self.limits.base_budget;
955                    if budget == 0 || self.live_path.load(Ordering::Relaxed) + n <= budget {
956                        // Room for it now: keep the re-inflated copy for the
957                        // next child rather than paying again. No eviction to
958                        // make room — that would thrash.
959                        s.stack[fi].payload = p.clone();
960                        s.stack[fi].evicted = false;
961                        let live = self.live_path.fetch_add(n, Ordering::Relaxed) + n;
962                        self.peak_path.fetch_max(live, Ordering::Relaxed);
963                    }
964                }
965                p
966            } else if last {
967                let p = std::mem::take(&mut s.stack[fi].payload);
968                self.live_path
969                    .fetch_sub(p.len() as u64, Ordering::Relaxed);
970                s.stack[fi].evicted = true;
971                p
972            } else {
973                s.stack[fi].payload.clone()
974            };
975
976            let ce = &self.w.entries[c as usize];
977            // The scratch buffers are taken out of `s` for the length of the
978            // call and put back on **every** path, error included: a bail that
979            // leaves them behind would make the next child allocate again.
980            let mut delta = std::mem::take(&mut s.delta);
981            let mut leaf = std::mem::take(&mut s.leaf);
982            let leafy = self.forest.children(c).is_empty();
983            let mut carried: Option<Vec<u8>> = None;
984            let r = (|| -> Result<()> {
985                inflate_into(self.pack, ce, self.oid_len, &mut delta)?;
986                if leafy {
987                    // A leaf's payload outlives nothing: it is applied into the
988                    // reused buffer, hashed in place and handed over from there.
989                    apply_delta_into(&base, &delta, &mut leaf)?;
990                    drop(base);
991                    self.emit(c, kind, &leaf, s)
992                } else {
993                    let mut out = Vec::with_capacity(ce.uncompressed_size as usize);
994                    apply_delta_into(&base, &delta, &mut out)?;
995                    drop(base);
996                    self.emit(c, kind, &out, s)?;
997                    carried = Some(out);
998                    Ok(())
999                }
1000            })();
1001            // Give back what one big object borrowed: the buffers stay warm at
1002            // `SCRATCH_KEEP_BYTES` and never carry the largest object in the
1003            // pack for the rest of it.
1004            delta.shrink_to(SCRATCH_KEEP_BYTES);
1005            leaf.shrink_to(SCRATCH_KEEP_BYTES);
1006            s.delta = delta;
1007            s.leaf = leaf;
1008            r?;
1009            if let Some(out) = carried {
1010                self.push_frame(c, kind, out, s);
1011            }
1012        }
1013        Ok(())
1014    }
1015
1016    fn push_frame(&self, entry: u32, kind: GitObjectKind, payload: Vec<u8>, s: &mut Scratch) {
1017        let len = payload.len() as u64;
1018        s.stack.push(Frame {
1019            entry,
1020            kind,
1021            payload,
1022            evicted: false,
1023            next_child: 0,
1024        });
1025        self.live_path.fetch_add(len, Ordering::Relaxed);
1026        // **The ceiling first, the high-water mark after.** Recording the peak
1027        // before the eviction counted the base that was about to be dropped,
1028        // so a one-byte budget reported two live bases and the report said the
1029        // ceiling had leaked when it had not.
1030        self.enforce_budget(s);
1031        self.peak_path
1032            .fetch_max(self.live_path.load(Ordering::Relaxed), Ordering::Relaxed);
1033    }
1034
1035    fn pop_frame(&self, s: &mut Scratch) {
1036        if let Some(f) = s.stack.pop()
1037            && !f.evicted
1038        {
1039            self.live_path
1040                .fetch_sub(f.payload.len() as u64, Ordering::Relaxed);
1041        }
1042    }
1043
1044    /// **Belady on a depth-first path.** The walk returns to the shallowest
1045    /// held frame only after every deeper subtree is finished, so the
1046    /// shallowest is by construction the one whose next use is farthest away —
1047    /// which is exactly the base the ceiling should drop. No `next_use` table
1048    /// is needed to know it; the shape says so.
1049    fn enforce_budget(&self, s: &mut Scratch) {
1050        let budget = self.limits.base_budget;
1051        if budget == 0 {
1052            return;
1053        }
1054        let mut i = 0usize;
1055        // The deepest frame is the one the next child reads from, so it is
1056        // never the eviction candidate — dropping it would re-inflate it on the
1057        // very next line.
1058        while i + 1 < s.stack.len() && self.live_path.load(Ordering::Relaxed) > budget {
1059            if !s.stack[i].evicted {
1060                let n = s.stack[i].payload.len() as u64;
1061                s.stack[i].payload = Vec::new();
1062                s.stack[i].evicted = true;
1063                self.live_path.fetch_sub(n, Ordering::Relaxed);
1064                self.evictions.fetch_add(1, Ordering::Relaxed);
1065            }
1066            i += 1;
1067        }
1068    }
1069
1070    /// Every payload on this worker's path, dropped, and **how many bytes that
1071    /// freed** — the number the watchdog compares with zero.
1072    fn drop_path(&self, s: &mut Scratch) -> u64 {
1073        let mut freed = 0u64;
1074        for f in s.stack.iter_mut() {
1075            if !f.evicted {
1076                freed += f.payload.len() as u64;
1077                f.payload = Vec::new();
1078                f.evicted = true;
1079            }
1080        }
1081        self.live_path.fetch_sub(freed, Ordering::Relaxed);
1082        freed
1083    }
1084
1085    /// Re-inflate an entry from the mmapped pack: its own chain, re-applied.
1086    /// **The one chain walker** (LAW 5) — [`materialize_entry`], with this
1087    /// round's promised external bases in front of the store's.
1088    fn materialize(&self, entry: u32) -> Result<(GitObjectKind, Vec<u8>)> {
1089        let rb = RoundBases {
1090            external: self.external,
1091            inner: self.bases,
1092        };
1093        let of = |oid: &[u8]| self.oid_to_offset.get(oid).copied();
1094        materialize_entry(self.pack, self.w, entry as usize, self.oid_len, &of, &rb)
1095    }
1096
1097    // ── the ordered end ─────────────────────────────────────────────────────
1098
1099    fn emit(&self, index: u32, kind: GitObjectKind, payload: &[u8], s: &mut Scratch) -> Result<()> {
1100        let e = &self.w.entries[index as usize];
1101        let oid = self.hash.oid_of_object(kind, payload);
1102        // Only now is "does a REF_DELTA name this object" answerable, because it
1103        // is keyed by the oid this line just computed.
1104        if self.ref_wanted.contains(&oid.as_slice()) {
1105            s.found_refs.push((oid.clone(), e.offset));
1106        }
1107        let want = self.sink.wants(kind);
1108        let em = Emit {
1109            kind,
1110            row: Resolved {
1111                oid,
1112                offset: e.offset + self.archive_offset,
1113                len: e.len,
1114                stored_type: e.obj_type,
1115                kind,
1116                uncompressed_size: payload.len() as u64,
1117                delta_base: match &e.delta_base {
1118                    DeltaBase::Offset(o) => o + self.archive_offset,
1119                    _ => 0,
1120                },
1121            },
1122            payload: if want { Some(payload.to_vec()) } else { None },
1123        };
1124        s.out_bytes += em.weight();
1125        s.out.push(em);
1126        self.rows.fetch_add(1, Ordering::Relaxed);
1127        if let Some(p) = self.limits.progress {
1128            p.object_done(e.len);
1129        }
1130        if s.out.len() >= HANDOVER_ROWS || s.out_bytes >= HANDOVER_BYTES {
1131            self.flush_scratch(s, false)?;
1132        }
1133        let n = self.since_check.fetch_add(1, Ordering::Relaxed) + 1;
1134        if n >= self.limits.check_every {
1135            self.since_check.store(0, Ordering::Relaxed);
1136            self.checkpoint(s)?;
1137        }
1138        Ok(())
1139    }
1140
1141    /// Hand this worker's finished objects to the ordered emitter, then try to
1142    /// push the stream forward. Never blocks on the drain: whoever holds it is
1143    /// already doing the work. `done` closes the unit — the drain may then step
1144    /// past it to the next root.
1145    fn flush_scratch(&self, s: &mut Scratch, done: bool) -> Result<()> {
1146        if !s.out.is_empty() || done {
1147            let mut bytes = 0u64;
1148            for em in s.out.iter() {
1149                bytes += em.weight();
1150            }
1151            let unit = s.unit;
1152            {
1153                let mut p = self.lock_pend()?;
1154                let e = p.map.entry(unit).or_default();
1155                if e.rows.is_empty() {
1156                    e.rows = std::mem::take(&mut s.out);
1157                } else {
1158                    e.rows.append(&mut s.out);
1159                }
1160                e.done |= done;
1161                p.bytes += bytes;
1162            }
1163            s.out.clear();
1164            s.out.shrink_to(HANDOVER_ROWS);
1165            s.out_bytes = 0;
1166        }
1167        self.try_drain(false)
1168    }
1169
1170    fn lock_pend(&self) -> Result<std::sync::MutexGuard<'_, Pending>> {
1171        self.pend
1172            .lock()
1173            .map_err(|_| anyhow!("the resolver's emit buffer is poisoned"))
1174    }
1175
1176    /// **The stream out.** This round's roots in ascending order, and inside
1177    /// each root the depth-first sequence its children lists fix — to the sink
1178    /// first, then to the row callback.
1179    fn try_drain(&self, force: bool) -> Result<()> {
1180        let mut guard = match self.drain.try_lock() {
1181            Ok(g) => g,
1182            Err(std::sync::TryLockError::WouldBlock) => {
1183                if !force {
1184                    return Ok(());
1185                }
1186                self.drain
1187                    .lock()
1188                    .map_err(|_| anyhow!("the resolver's row drain is poisoned"))?
1189            }
1190            Err(std::sync::TryLockError::Poisoned(_)) => {
1191                bail!("the resolver's row drain is poisoned")
1192            }
1193        };
1194        let d = &mut *guard;
1195        loop {
1196            let mut run: Vec<Vec<Emit>> = Vec::new();
1197            let mut held = 0usize;
1198            let mut moved = false;
1199            {
1200                let mut p = self.lock_pend()?;
1201                while d.cursor < self.roots.len() {
1202                    let (rows, done) = match p.map.get_mut(&d.cursor) {
1203                        Some(ro) => (std::mem::take(&mut ro.rows), ro.done),
1204                        None => break,
1205                    };
1206                    if !rows.is_empty() {
1207                        let n: u64 = rows.iter().map(Emit::weight).sum();
1208                        p.bytes = p.bytes.saturating_sub(n);
1209                        held += rows.len();
1210                        run.push(rows);
1211                        moved = true;
1212                    }
1213                    if !done {
1214                        break;
1215                    }
1216                    p.map.remove(&d.cursor);
1217                    d.cursor += 1;
1218                    moved = true;
1219                    if held >= DRAIN_RUN {
1220                        break;
1221                    }
1222                }
1223                if moved {
1224                    self.cv.notify_all();
1225                }
1226            }
1227            if !moved {
1228                return Ok(());
1229            }
1230            for chunk in run {
1231                for em in chunk {
1232                    if let Some(pay) = &em.payload {
1233                        self.sink.explode(&em.row.oid, em.kind, pay)?;
1234                    }
1235                    d.batch.push(em.row);
1236                    if d.batch.len() >= self.limits.row_batch {
1237                        (d.rows)(&mut d.batch)?;
1238                        self.batches.fetch_add(1, Ordering::Relaxed);
1239                        d.batch.clear();
1240                    }
1241                }
1242            }
1243        }
1244    }
1245
1246    /// Everything left, in order, at the end of a round.
1247    fn finish(&self) -> Result<()> {
1248        self.try_drain(true)?;
1249        let mut d = self
1250            .drain
1251            .lock()
1252            .map_err(|_| anyhow!("the resolver's row drain is poisoned"))?;
1253        if !d.batch.is_empty() {
1254            let d = &mut *d;
1255            (d.rows)(&mut d.batch)?;
1256            self.batches.fetch_add(1, Ordering::Relaxed);
1257            d.batch.clear();
1258        }
1259        let p = self.lock_pend()?;
1260        if !p.map.is_empty() || d.cursor != self.roots.len() {
1261            bail!(
1262                "the resolver finished a round at root {} of {} with {} roots still unemitted — \
1263                 the round's roots and the walk disagree",
1264                d.cursor,
1265                self.roots.len(),
1266                p.map.len()
1267            );
1268        }
1269        Ok(())
1270    }
1271
1272    /// **The lookahead ceiling, and the one worker it may never stop.**
1273    ///
1274    /// A worker that has run far ahead of the ordered stream waits here rather
1275    /// than buffering the pack. The rule for *who* waits is the whole
1276    /// correctness argument: the worker holding the **lowest unit in flight**
1277    /// never waits. Gatling hands its units out in ascending order, so every
1278    /// lower unit is already finished, and the entry the ordered cursor is
1279    /// stuck on therefore belongs either to a finished unit — already in the
1280    /// buffer, and the drain will take it — or to the lowest unit in flight,
1281    /// which is running. Progress is guaranteed either way.
1282    ///
1283    /// 🔴 It said "one worker is never allowed to wait, whichever it is", and
1284    /// that HUNG: gatling claims a unit *before* it calls the closure, so the
1285    /// worker allowed to run on was frequently not the one the cursor was
1286    /// waiting for. It then consumed every remaining unit while the worker
1287    /// holding the cursor's own unit sat in `wait` forever, and the scope never
1288    /// joined. Caught by
1289    /// `the_same_pack_resolves_to_the_same_bytes_at_1_2_and_4_workers`, which
1290    /// ran for over 60 s and had to be killed.
1291    fn backpressure(&self, ri: usize) -> Result<()> {
1292        self.try_drain(false)?;
1293        let mut p = self.lock_pend()?;
1294        while p.bytes > self.lookahead
1295            && !self.stop.load(Ordering::Relaxed)
1296            && p.in_flight.first().is_some_and(|lowest| *lowest < ri)
1297        {
1298            self.stalls.fetch_add(1, Ordering::Relaxed);
1299            p = self
1300                .cv
1301                .wait(p)
1302                .map_err(|_| anyhow!("the resolver's emit buffer is poisoned"))?;
1303        }
1304        Ok(())
1305    }
1306
1307    /// Enter and leave a unit. The set is what [`backpressure`](Self::backpressure)
1308    /// reads, and leaving one is what lets a waiting worker re-decide.
1309    fn enter(&self, ri: usize) -> Result<()> {
1310        self.lock_pend()?.in_flight.insert(ri);
1311        Ok(())
1312    }
1313
1314    fn leave(&self, ri: usize) {
1315        if let Ok(mut p) = self.pend.lock() {
1316            p.in_flight.remove(&ri);
1317        }
1318        self.cv.notify_all();
1319    }
1320
1321    // ── the checkpoints ─────────────────────────────────────────────────────
1322
1323    /// Every [`crate::memguard::CHECK_EVERY_OBJECTS`] objects **of the whole
1324    /// job**: another worker's failure, the cancel flag, then the watchdog.
1325    fn checkpoint(&self, s: &mut Scratch) -> Result<()> {
1326        if self.stop.load(Ordering::Relaxed) {
1327            bail!("this pack is being failed by another worker");
1328        }
1329        if self.limits.cancel.is_some_and(|c| c.load(Ordering::Relaxed)) {
1330            return Err(ResolveError::cancelled());
1331        }
1332        let Some(dog) = self.limits.watchdog.clone() else {
1333            return Ok(());
1334        };
1335        if dog.breached().is_none() {
1336            return Ok(());
1337        }
1338        // My own share of the problem, dropped before anybody waits.
1339        let mut freed = self.drop_path(s);
1340
1341        // ── GIVE THE CORE BACK BEFORE GIVING THE PACK UP ────────────────────
1342        //
1343        // **A short box wants fewer workers, not a refused pack.** MEASURED on
1344        // the rig, the 16 GB appliance and the 8.12 GiB kernel mirror: one
1345        // worker finishes it with 5.55 GiB to spare, two stop at 96 % and four
1346        // at 86 %. The cost is not the resolver's own structures — peak RSS at
1347        // four workers (13.7 GiB) is BELOW the single-threaded fixpoint's
1348        // (14.04 GiB) — it is that four depth-first cursors over a pack whose
1349        // subtrees span millions of entries keep far more of the 8.12 GiB
1350        // mapping resident at once, and `MemAvailable` counts half of page
1351        // cache.
1352        //
1353        // So a worker that finds the box short **stops being a worker**: it
1354        // waits until it is the lowest unit in flight, which is exactly when
1355        // every worker above it has drained, and the round finishes on one
1356        // core instead of being refused on four. It never deadlocks for the
1357        // same reason [`backpressure`](Self::backpressure) does not — the
1358        // lowest unit in flight never waits, and gatling hands units out in
1359        // ascending order, so the lowest always advances.
1360        //
1361        // The ordered emitter means this changes only WHEN work happens, never
1362        // in what order it comes out: the stream is byte-identical either way.
1363        {
1364            let mut p = self.lock_pend()?;
1365            let mut waited = false;
1366            loop {
1367                if self.stop.load(Ordering::Relaxed) {
1368                    break;
1369                }
1370                let lowest = p.in_flight.first().copied();
1371                if lowest.is_some_and(|l| l < s.unit) {
1372                    // Somebody below me is still the one the ordered stream is
1373                    // waiting for: I am the expendable one. Park.
1374                    if !waited {
1375                        waited = true;
1376                        p.parked += 1;
1377                        self.shed.fetch_add(1, Ordering::Relaxed);
1378                        if let Some(pr) = self.limits.progress {
1379                            pr.set_phase(Phase::Paused);
1380                        }
1381                        self.cv.notify_all();
1382                    }
1383                    p = self
1384                        .cv
1385                        .wait(p)
1386                        .map_err(|_| anyhow!("the resolver's emit buffer is poisoned"))?;
1387                    continue;
1388                }
1389                // I am the lowest unit in flight — the one that must carry on.
1390                // **Do not pause while other workers are still resolving**:
1391                // they are what the box cannot afford, and each of them will
1392                // park at its own next checkpoint. Pausing here instead would
1393                // stop the only worker that is allowed to make progress and
1394                // then time out on a box that was about to have room.
1395                if p.in_flight.len() > p.parked + 1 {
1396                    if waited {
1397                        waited = false;
1398                        p.parked -= 1;
1399                    }
1400                    drop(p);
1401                    if let Some(pr) = self.limits.progress {
1402                        pr.set_phase(Phase::Resolve);
1403                    }
1404                    return Ok(());
1405                }
1406                break;
1407            }
1408            if waited {
1409                p.parked -= 1;
1410            }
1411            drop(p);
1412            if waited {
1413                if let Some(pr) = self.limits.progress {
1414                    pr.set_phase(Phase::Resolve);
1415                }
1416                if self.stop.load(Ordering::Relaxed) {
1417                    bail!("this pack is being failed by another worker");
1418                }
1419                // The box may well be fine now that everybody above me has
1420                // stopped. Re-ask before waking the watchdog.
1421                if dog.breached().is_none() {
1422                    return Ok(());
1423                }
1424            }
1425        }
1426
1427        if let Some(p) = self.limits.progress {
1428            p.set_phase(Phase::Paused);
1429            p.add_pause();
1430        }
1431        // **Only one worker drives the pause**; the others block on this gate,
1432        // which is the point — an indexer that keeps three threads allocating
1433        // through a pause is not paused. They have already dropped their own
1434        // bases on the way in.
1435        let gate = self
1436            .pause_gate
1437            .lock()
1438            .map_err(|_| anyhow!("the resolver's pause gate is poisoned"))?;
1439        // 🔴 **A give-up is the whole pack's, not one worker's.** Without this
1440        // line the workers queued on the gate each opened a pause of their own
1441        // after the first had already given up. MEASURED on the rig, two
1442        // workers: `gave up after 30681 ms (nothing-to-give)` at 00:59:33, then
1443        // a second pause opened in the same millisecond and ran the FULL
1444        // `PAUSE_DEADLINE` — `gave up after 300164 ms (deadline)` at 01:04:33.
1445        // Five minutes of a client waiting for a verdict the box had already
1446        // reached. One give-up stops the round.
1447        if self.stop.load(Ordering::Relaxed) {
1448            drop(gate);
1449            bail!("this pack is being failed by another worker");
1450        }
1451        if dog.breached().is_none() {
1452            drop(gate);
1453            if let Some(p) = self.limits.progress {
1454                p.set_phase(Phase::Resolve);
1455            }
1456            return Ok(());
1457        }
1458        let (cancel, on_pause, label) = (self.limits.cancel, self.limits.on_pause, self.limits.label);
1459        let mut flush = || -> Result<u64> {
1460            let mut f = std::mem::take(&mut freed);
1461            self.watchdog_drops.fetch_add(1, Ordering::Relaxed);
1462            self.sink.flush()?;
1463            // `false`: the unit is mid-walk, so the drain must not step past
1464            // this root — it is not finished, it is paused.
1465            self.flush_scratch(s, false)?;
1466            self.try_drain(true)?;
1467            if let Some(g) = on_pause {
1468                f = f.saturating_add(g());
1469            }
1470            Ok(f)
1471        };
1472        let paused = dog.pause_until_ok(label, cancel, &mut flush);
1473        let ms = match &paused {
1474            Ok(p) => p.ms,
1475            Err(_) => 0,
1476        };
1477        self.pauses.fetch_add(1, Ordering::Relaxed);
1478        self.paused_ms.fetch_add(ms, Ordering::Relaxed);
1479        if let Some(p) = self.limits.progress {
1480            p.set_phase(Phase::Resolve);
1481        }
1482        drop(gate);
1483        let paused = paused?;
1484        if self.limits.cancel.is_some_and(|c| c.load(Ordering::Relaxed)) {
1485            return Err(ResolveError::cancelled());
1486        }
1487        // **A give-up is not a resume.** The pack is refused by name; its bytes
1488        // stay durable and nothing partial is left indexed.
1489        if paused.end.gave_up() {
1490            self.gave_up.store(true, Ordering::Relaxed);
1491            // Before the error is even built: every worker queued behind the
1492            // gate must find the round already lost.
1493            self.stop.store(true, Ordering::Relaxed);
1494            if self.pend.lock().is_ok() {
1495                self.cv.notify_all();
1496            }
1497            let (done, total) = match self.limits.progress {
1498                Some(p) => {
1499                    let sn = p.snapshot();
1500                    (sn.objects_done, sn.objects_total)
1501                }
1502                None => (self.rows.load(Ordering::Relaxed), self.w.entries.len() as u64),
1503            };
1504            return Err(ResolveError::out_of_memory(
1505                paused.end.as_str(),
1506                paused.available,
1507                paused.budget,
1508                done,
1509                total,
1510            ));
1511        }
1512        Ok(())
1513    }
1514
1515    /// The first failure by **root index**, so a pack that two workers could
1516    /// fail is refused with the same sentence every time.
1517    fn fail(&self, ri: usize, e: anyhow::Error) {
1518        self.stop.store(true, Ordering::Relaxed);
1519        if let Ok(mut g) = self.err.lock() {
1520            let take = match &*g {
1521                Some((i, _)) => ri < *i,
1522                None => true,
1523            };
1524            if take {
1525                *g = Some((ri, e));
1526            }
1527        }
1528        if self.pend.lock().is_ok() {
1529            self.cv.notify_all();
1530        }
1531    }
1532
1533    fn take_err(&self) -> Option<anyhow::Error> {
1534        self.err.lock().ok().and_then(|mut g| g.take()).map(|(_, e)| e)
1535    }
1536
1537    fn report(&self) -> ResolveReport {
1538        ResolveReport {
1539            rows: self.rows.load(Ordering::Relaxed),
1540            reinflations: self.reinflations.load(Ordering::Relaxed),
1541            evictions: self.evictions.load(Ordering::Relaxed),
1542            watchdog_drops: self.watchdog_drops.load(Ordering::Relaxed),
1543            peak_base_bytes: self.peak_path.load(Ordering::Relaxed),
1544            pauses: self.pauses.load(Ordering::Relaxed),
1545            paused_ms: self.paused_ms.load(Ordering::Relaxed),
1546            batches: self.batches.load(Ordering::Relaxed),
1547            stalls: self.stalls.load(Ordering::Relaxed),
1548            shed: self.shed.load(Ordering::Relaxed),
1549            workers: self.workers as u32,
1550            gave_up: self.gave_up.load(Ordering::Relaxed),
1551        }
1552    }
1553}
1554
1555/// **The resolver.** One walk, one forest, one ordered stream out, and as many
1556/// workers as the box was told to use.
1557struct Resolver<'p, 'w, 'a> {
1558    pack: &'p [u8],
1559    w: &'w PackWalk,
1560    hash: GitHashKind,
1561    oid_len: usize,
1562    archive_offset: u64,
1563    bases: &'a dyn BaseSource,
1564    sink: &'a dyn PayloadSink,
1565    limits: ResolveLimits<'a>,
1566    forest: DeltaForest,
1567    /// The oids `REF_DELTA` entries in this pack name — the half of the
1568    /// dependent set that is not knowable from an offset.
1569    ref_wanted: std::collections::HashSet<&'w [u8]>,
1570    /// …and where they turned out to live, as the rounds place them.
1571    oid_to_offset: HashMap<Vec<u8>, u64>,
1572    report: ResolveReport,
1573}
1574
1575impl<'p, 'w, 'a> Resolver<'p, 'w, 'a> {
1576    fn new(
1577        pack: &'p [u8],
1578        w: &'w PackWalk,
1579        hash: GitHashKind,
1580        archive_offset: u64,
1581        bases: &'a dyn BaseSource,
1582        sink: &'a dyn PayloadSink,
1583        limits: ResolveLimits<'a>,
1584    ) -> Result<Self> {
1585        // **The dependents, as the forest they always were.** Built from
1586        // `PackWalk::entries` and nothing else (LAW 5): `parent`, the CSR child
1587        // lists, and the roots.
1588        let forest = DeltaForest::build(w)?;
1589        let mut ref_wanted: std::collections::HashSet<&'w [u8]> = std::collections::HashSet::new();
1590        for e in w.entries.iter() {
1591            if let DeltaBase::Ref(oid) = &e.delta_base {
1592                ref_wanted.insert(oid.as_slice());
1593            }
1594        }
1595        Ok(Self {
1596            pack,
1597            w,
1598            hash,
1599            oid_len: hash.oid_len(),
1600            archive_offset,
1601            bases,
1602            sink,
1603            forest,
1604            ref_wanted,
1605            oid_to_offset: HashMap::new(),
1606            report: ResolveReport::default(),
1607            limits,
1608        })
1609    }
1610
1611    /// The fixpoint, one round at a time — and each round is a gatling fan-out
1612    /// over its roots.
1613    ///
1614    /// A round is planned before a byte of it is inflated: its roots are the
1615    /// entries whose base is available *now*, and its entry set is the closure
1616    /// of those roots over the forest. That is what lets the ordered emitter
1617    /// know, for every index, whether to wait for it or step past it — and
1618    /// therefore what makes the stream out of N workers the stream out of one.
1619    fn run(&mut self, rows: &mut (dyn FnMut(&mut Vec<Resolved>) -> Result<()> + Send)) -> Result<()> {
1620        let n = self.w.entries.len();
1621        if let Some(p) = self.limits.progress {
1622            p.set_total(n as u64);
1623            p.set_phase(Phase::Resolve);
1624        }
1625        if n == 0 {
1626            return Ok(());
1627        }
1628        let w = self.w;
1629        let mut done = EntrySet::new(n);
1630        let mut round = 0usize;
1631        while done.len() < n {
1632            let mut roots: Vec<u32> = if round == 0 {
1633                self.forest.whole_roots().to_vec()
1634            } else {
1635                Vec::new()
1636            };
1637            // A `REF_DELTA` names a base by oid. It is resolvable this round if
1638            // an earlier round placed that oid inside the pack, or if the store
1639            // can produce it — exactly the two answers the serial fixpoint
1640            // asked for, asked once each, up front, so the round's membership
1641            // is a fact rather than a race.
1642            let mut external: HashMap<&'w [u8], (GitObjectKind, Vec<u8>)> = HashMap::new();
1643            for &r in self.forest.ref_roots() {
1644                if done.contains(r as usize) {
1645                    continue;
1646                }
1647                let DeltaBase::Ref(oid) = &w.entries[r as usize].delta_base else {
1648                    continue;
1649                };
1650                if self.oid_to_offset.contains_key(oid.as_slice())
1651                    || external.contains_key(oid.as_slice())
1652                {
1653                    roots.push(r);
1654                    continue;
1655                }
1656                if let Some(c) = self.bases.content(oid)? {
1657                    external.insert(oid.as_slice(), c);
1658                    roots.push(r);
1659                }
1660            }
1661            if roots.is_empty() {
1662                // Nothing can move: every remaining entry waits on a base
1663                // nobody can supply. Name the first one — a thin pack is the
1664                // ordinary reason.
1665                for &r in self.forest.ref_roots() {
1666                    if !done.contains(r as usize)
1667                        && let DeltaBase::Ref(oid) = &w.entries[r as usize].delta_base
1668                    {
1669                        return Err(ResolveError::missing_base(oid));
1670                    }
1671                }
1672                bail!(
1673                    "{} of {n} entries could not be resolved and none of them deltas against an \
1674                     oid — the pack is corrupt",
1675                    n - done.len()
1676                );
1677            }
1678            roots.sort_unstable();
1679            let in_round = self.forest.closure_of(&roots);
1680            let (rep, found) = self.run_round(&roots, &in_round, &external, rows)?;
1681            self.report.absorb(rep);
1682            for (oid, off) in found {
1683                self.oid_to_offset.insert(oid, off);
1684            }
1685            done.union_with(&in_round);
1686            round += 1;
1687        }
1688        Ok(())
1689    }
1690
1691    /// One round: **the gatling fan-out**. NEVER rayon (LAW 3) — every thread
1692    /// in this constellation originates in `gatling`, and this is its fork-join
1693    /// shape: a fixed `0..roots` of independent units, self-dispatched from one
1694    /// atomic, no barrier between them.
1695    fn run_round(
1696        &self,
1697        roots: &[u32],
1698        in_round: &EntrySet,
1699        external: &HashMap<&'w [u8], (GitObjectKind, Vec<u8>)>,
1700        rows: &mut (dyn FnMut(&mut Vec<Resolved>) -> Result<()> + Send),
1701    ) -> Result<(ResolveReport, Vec<(Vec<u8>, u64)>)> {
1702        let workers = self
1703            .limits
1704            .workers
1705            .max(1)
1706            .min(roots.len().max(1))
1707            .min(MAX_WORKERS);
1708        let round = Round {
1709            pack: self.pack,
1710            w: self.w,
1711            forest: &self.forest,
1712            hash: self.hash,
1713            oid_len: self.oid_len,
1714            archive_offset: self.archive_offset,
1715            bases: self.bases,
1716            sink: self.sink,
1717            limits: &self.limits,
1718            ref_wanted: &self.ref_wanted,
1719            oid_to_offset: &self.oid_to_offset,
1720            external,
1721            roots,
1722            in_round,
1723            workers,
1724            lookahead: self.limits.lookahead_bytes.max(1 << 20),
1725            pend: std::sync::Mutex::new(Pending::default()),
1726            cv: std::sync::Condvar::new(),
1727            drain: std::sync::Mutex::new(Drain {
1728                cursor: 0,
1729                batch: Vec::with_capacity(self.limits.row_batch.min(1 << 20)),
1730                rows,
1731            }),
1732            pause_gate: std::sync::Mutex::new(()),
1733            stop: AtomicBool::new(false),
1734            err: std::sync::Mutex::new(None),
1735            since_check: std::sync::atomic::AtomicU64::new(0),
1736            rows: std::sync::atomic::AtomicU64::new(0),
1737            reinflations: std::sync::atomic::AtomicU64::new(0),
1738            evictions: std::sync::atomic::AtomicU64::new(0),
1739            watchdog_drops: std::sync::atomic::AtomicU64::new(0),
1740            live_path: std::sync::atomic::AtomicU64::new(0),
1741            peak_path: std::sync::atomic::AtomicU64::new(0),
1742            pauses: std::sync::atomic::AtomicU64::new(0),
1743            paused_ms: std::sync::atomic::AtomicU64::new(0),
1744            batches: std::sync::atomic::AtomicU64::new(0),
1745            stalls: std::sync::atomic::AtomicU64::new(0),
1746            shed: std::sync::atomic::AtomicU64::new(0),
1747            gave_up: AtomicBool::new(false),
1748            found_refs: std::sync::Mutex::new(Vec::new()),
1749        };
1750        gatling_for_each_unit(roots.len(), workers, |ri| {
1751            if round.stop.load(Ordering::Relaxed) {
1752                return;
1753            }
1754            let r = SCRATCH.with(|cell| -> Result<()> {
1755                let mut s = cell.borrow_mut();
1756                round.enter(ri)?;
1757                let r = (|| {
1758                    round.backpressure(ri)?;
1759                    if round.stop.load(Ordering::Relaxed) {
1760                        return Ok(());
1761                    }
1762                    round.run_root(ri, &mut s)
1763                })();
1764                if !s.found_refs.is_empty()
1765                    && let Ok(mut g) = round.found_refs.lock()
1766                {
1767                    g.append(&mut s.found_refs);
1768                }
1769                round.leave(ri);
1770                r
1771            });
1772            if let Err(e) = r {
1773                round.fail(ri, e);
1774            }
1775        });
1776        if let Some(e) = round.take_err() {
1777            // Everything this round did stays where it is; the caller refuses
1778            // the pack by name and nothing partial is marked indexed.
1779            return Err(e);
1780        }
1781        round.finish()?;
1782        let mut found = round
1783            .found_refs
1784            .lock()
1785            .map_err(|_| anyhow!("the resolver's ref table is poisoned"))?
1786            .split_off(0);
1787        // The order N workers found them in is not an order; sorting makes the
1788        // map the same map at any worker count.
1789        found.sort_unstable();
1790        found.dedup();
1791        Ok((round.report(), found))
1792    }
1793}
1794
1795/// The gatling fork-join, by its own name, with the one-worker case kept off
1796/// the thread pool entirely: `workers == 1` runs the units on the calling
1797/// thread, which is what makes a 1-worker run the reference a 2- and a
1798/// 4-worker run are compared against.
1799fn gatling_for_each_unit<F: Fn(usize) + Sync>(n: usize, workers: usize, f: F) {
1800    znippy_zoomies::gatling_forkjoin::gatling_run(n, workers, f);
1801}
1802
1803
1804/// **One object's content from a pack and an entry, with only its chain
1805/// inflated** — the lazy read path. `offset_of_oid` answers a `REF_DELTA`
1806/// base that is inside the pack (its entry offset); `bases` answers one that is
1807/// not. Nothing is cached: a chain of depth d costs d inflates and d−1 delta
1808/// applications, and nothing else.
1809pub fn materialize_entry(
1810    pack: &[u8],
1811    w: &PackWalk,
1812    entry: usize,
1813    oid_len: usize,
1814    offset_of_oid: &dyn Fn(&[u8]) -> Option<u64>,
1815    bases: &dyn BaseSource,
1816) -> Result<(GitObjectKind, Vec<u8>)> {
1817    let e = w
1818        .entries
1819        .get(entry)
1820        .ok_or_else(|| anyhow!("entry {entry} is past the walk ({} entries)", w.entries.len()))?;
1821    match &e.delta_base {
1822        DeltaBase::None => {
1823            let mut out = Vec::with_capacity(e.uncompressed_size as usize);
1824            inflate_into(pack, e, oid_len, &mut out)?;
1825            Ok((whole_kind(e.obj_type)?, out))
1826        }
1827        DeltaBase::Offset(o) => {
1828            let base_entry = w
1829                .entries
1830                .binary_search_by_key(o, |x| x.offset)
1831                .map_err(|_| anyhow!("no entry starts at offset {o} — the pack is corrupt"))?;
1832            let (kind, base) = materialize_entry(pack, w, base_entry, oid_len, offset_of_oid, bases)?;
1833            let mut delta = Vec::with_capacity(e.uncompressed_size as usize);
1834            inflate_into(pack, e, oid_len, &mut delta)?;
1835            let mut out = Vec::new();
1836            apply_delta_into(&base, &delta, &mut out)?;
1837            Ok((kind, out))
1838        }
1839        DeltaBase::Ref(oid) => {
1840            let (kind, base) = match offset_of_oid(oid) {
1841                Some(o) => {
1842                    let base_entry = w
1843                        .entries
1844                        .binary_search_by_key(&o, |x| x.offset)
1845                        .map_err(|_| anyhow!("no entry starts at offset {o} — the pack is corrupt"))?;
1846                    materialize_entry(pack, w, base_entry, oid_len, offset_of_oid, bases)?
1847                }
1848                None => bases.content(oid)?.ok_or_else(|| ResolveError::missing_base(oid))?,
1849            };
1850            let mut delta = Vec::with_capacity(e.uncompressed_size as usize);
1851            inflate_into(pack, e, oid_len, &mut delta)?;
1852            let mut out = Vec::new();
1853            apply_delta_into(&base, &delta, &mut out)?;
1854            Ok((kind, out))
1855        }
1856    }
1857}
1858
1859
1860// ════════════════════════════════════════════════════════════════════════════
1861// The LAZY read path: one object, one chain, straight out of the archive
1862// ════════════════════════════════════════════════════════════════════════════
1863
1864/// Bytes of the verbatim archive, addressed **absolutely**.
1865///
1866/// The mmap ([`crate::archive_map`]) answers without a syscall and without a
1867/// copy; a `pread` answers the tail a snapshot does not cover. Both are the
1868/// same bytes, which is why the chain walker below can take either and never
1869/// know which it got.
1870///
1871/// It exists so [`materialize_at`] can live here — beside the delta code it
1872/// reuses — without this module learning what a store is.
1873pub trait ArchiveBytes {
1874    /// Up to `len` bytes at `at`. **Short is not an error**: a probe near the
1875    /// end of the archive legitimately gets fewer, and the parsers below read a
1876    /// grammar rather than a fixed width. Empty means "there is nothing there",
1877    /// which every caller treats as a corrupt back-reference.
1878    fn at(&self, at: u64, len: u64) -> Result<std::borrow::Cow<'_, [u8]>>;
1879}
1880
1881/// A whole `&[u8]` as an archive — what a test and a single-pack caller use.
1882pub struct SliceArchive<'a>(pub &'a [u8]);
1883
1884impl ArchiveBytes for SliceArchive<'_> {
1885    fn at(&self, at: u64, len: u64) -> Result<std::borrow::Cow<'_, [u8]>> {
1886        let start = usize::try_from(at).unwrap_or(usize::MAX).min(self.0.len());
1887        let end = start.saturating_add(usize::try_from(len).unwrap_or(usize::MAX)).min(self.0.len());
1888        Ok(std::borrow::Cow::Borrowed(&self.0[start..end]))
1889    }
1890}
1891
1892/// **What one read actually cost**, measured by the read rather than predicted.
1893///
1894/// The whole point of carrying it: the cache admission decision needs to know
1895/// whether an object was expensive to produce, and the reader that just
1896/// produced it is the only thing in the system that knows for certain. The
1897/// index could carry a depth column instead — git's `.idx` does not, but the
1898/// walk computes one — and it would still be a prediction of what a read will
1899/// cost, while this is the bill.
1900#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1901pub struct ReadCost {
1902    /// Delta links followed. `0` is a whole object: one inflate of one zlib
1903    /// stream at one offset, which is what makes a direct read cheap enough
1904    /// that caching it would be a waste of somebody's quota.
1905    pub depth: u32,
1906    /// Bytes the chain produced.
1907    pub inflated: u64,
1908    /// The chain bottomed out on an object this archive does not hold, so a
1909    /// [`BaseSource`] paid for it. Rare, and never cheap.
1910    pub external_base: bool,
1911}
1912
1913/// git's own `pack.depth` ceiling doubled. **The depth at which a chain walk
1914/// stops assuming it will terminate and starts proving it** — see
1915/// [`CycleGuard`]. It is not a refusal any more, and the 🔴 below says why.
1916///
1917/// # 🔴 It WAS a refusal, and that refusal failed ordinary pushes
1918///
1919/// MEASURED, oden, 2026-08-30 (lane AI, reproducing lane AD's rig red): 140
1920/// sequential single-commit pushes of one file into one repository. Push #120
1921/// was refused with *"this pack is thin: it deltas against 509be6a7… which is
1922/// not in it, and no base source could produce that object's content … so this
1923/// repository genuinely does not have that object"* — and every one of the 20
1924/// pushes after it was refused too, forever. The object was **in the store, one
1925/// index lookup away, in a pack that had already been absorbed**. The sentence
1926/// was a lie, and the same red reproduces in this crate alone at push 100 of
1927/// 150 (`crate::thin_storm`).
1928///
1929/// The cause is what a verbatim archive does that stock git does not.
1930/// `git index-pack --fix-thin` *completes* a pushed thin pack by appending the
1931/// resolved base into it, so a stored pack's delta chains never leave it and
1932/// stay under the client's own `pack.depth`. §14 stores the client's bytes
1933/// **verbatim** — that is what makes a clone a byte-range copy — so a
1934/// `REF_DELTA` keeps naming a base in an *earlier pack*, and a repository
1935/// pushed one commit at a time grows a chain **one link per push, without
1936/// bound**. At the 101st link this ceiling turned a legitimate chain into
1937/// "corrupt pack", and [`crate::git_ops`]'s base lookup then reported the
1938/// object as ABSENT rather than as unreadable, which is how a store defect
1939/// wore a client's clothes.
1940///
1941/// A count cannot tell a deep chain from a loop. Revisiting an offset can, and
1942/// that is what [`CycleGuard`] does — so a cycle is caught *sooner* (the first
1943/// repeat, wherever it sits) and a legitimate chain is never refused. The
1944/// constant keeps its name and its value; only its job changed.
1945pub const MAX_DELTA_LINKS: usize = 100;
1946
1947/// **A termination proof for a chain walk, paid for only by the chains that
1948/// need one.**
1949///
1950/// Under [`MAX_DELTA_LINKS`] links it is two integer operations and no
1951/// allocation — the shape every ordinary read has always had. Past that depth
1952/// it starts recording the archive offsets it has stood on, and a repeat is a
1953/// cycle *by definition* rather than by a guess about how deep is too deep.
1954/// The set is only ever entered on chains a corrupt pack or a long push history
1955/// produced, so the hot path is unchanged; a legitimate chain of `d` links
1956/// costs `d - MAX_DELTA_LINKS` `u64` slots and nothing else.
1957///
1958/// It is one guard for both walkers (LAW 5): [`materialize_at`] here and
1959/// `GitStore::resolved_type_at` in [`crate::serve`], which followed the same
1960/// links under a hand-written copy of the same ceiling.
1961pub struct CycleGuard {
1962    links: usize,
1963    seen: std::collections::HashSet<u64>,
1964}
1965
1966impl Default for CycleGuard {
1967    fn default() -> Self {
1968        Self::new()
1969    }
1970}
1971
1972impl CycleGuard {
1973    pub fn new() -> Self {
1974        Self {
1975            links: 0,
1976            // Empty, so a chain shallower than the threshold allocates nothing.
1977            seen: std::collections::HashSet::new(),
1978        }
1979    }
1980
1981    /// One link, standing at archive offset `at`. `Err` **only** when this walk
1982    /// has already stood there, which no acyclic chain can do.
1983    pub fn step(&mut self, at: u64) -> Result<()> {
1984        self.links += 1;
1985        if self.links <= MAX_DELTA_LINKS {
1986            return Ok(());
1987        }
1988        if !self.seen.insert(at) {
1989            bail!(
1990                "the delta chain returns to archive offset {at} after {} links — the pack's \
1991                 back-references form a cycle, so the entry cannot be reconstructed. This is a \
1992                 corrupt pack and not a deep one: depth alone is never refused here, because a \
1993                 verbatim archive grows one link per push (see MAX_DELTA_LINKS)",
1994                self.links
1995            );
1996        }
1997        Ok(())
1998    }
1999
2000    /// How many links this walk has taken.
2001    pub fn links(&self) -> usize {
2002        self.links
2003    }
2004}
2005
2006/// Enough of an entry to hold its type/size varint plus either an `OFS_DELTA`
2007/// distance or a `REF_DELTA` oid. 64 covers a sha256 oid and the widest varint.
2008const HEADER_PROBE: u64 = 64;
2009
2010/// How much compressed stream is fetched per inflate step. The stream's length
2011/// is not written down anywhere the index keeps, so it is discovered by
2012/// inflating; a window means a 40-byte commit costs one probe rather than a
2013/// megabyte of page faults.
2014const INFLATE_WINDOW: u64 = 256 << 10;
2015
2016/// One link of a delta chain, as the entry's own grammar states it.
2017struct Link {
2018    obj_type: ObjType,
2019    /// The size the header declares: of the delta instruction stream for a
2020    /// delta, of the object itself for a whole entry.
2021    size: u64,
2022    /// Where the zlib stream starts — header skipped.
2023    data_at: u64,
2024}
2025
2026/// **One object's content, with only its own chain inflated, addressed by
2027/// archive offset.**
2028///
2029/// This is the lazy-explode read path (`ideas/lazy-explode-index-only-at-push.md`).
2030/// The eager tier's alternative was to read the *whole containing pack* back and
2031/// re-resolve it — for `torvalds/linux` an 8.12 GiB read and a 13.9 M-object
2032/// resolve **per object served**, which is why the exploded tier had to be built
2033/// eagerly at push time and why that tier reached 141.7 GiB. Reading a chain
2034/// instead costs `d` inflates and `d − 1` delta applications and touches nothing
2035/// else in the pack.
2036///
2037/// * `at` — the entry's absolute archive offset, from the `objects` table's
2038///   `offset` column (or from a pending pack's scan).
2039/// * `offset_of_oid` — resolves a `REF_DELTA`'s base oid to an archive offset;
2040///   the `objects` index answers it.
2041/// * `bases` — answers a `REF_DELTA` base the index does not hold (a thin
2042///   pack's external base). `None` from it is
2043///   [`ResolveError::missing_base`], refused by name, never guessed.
2044///
2045/// Nothing is cached and nothing is written: the caller decides whether the
2046/// bytes are worth keeping. That separation is what lets the write-back happen
2047/// off the reader's path.
2048pub fn materialize_at(
2049    archive: &dyn ArchiveBytes,
2050    at: u64,
2051    oid_len: usize,
2052    offset_of_oid: &dyn Fn(&[u8]) -> Option<u64>,
2053    bases: &dyn BaseSource,
2054) -> Result<(GitObjectKind, Vec<u8>, ReadCost)> {
2055    // ── walk the chain outward, deepest link last ──────────────────────────
2056    let mut chain: Vec<Link> = Vec::new();
2057    // Set when the chain bottoms out at an object this archive does not hold.
2058    let mut external: Option<Vec<u8>> = None;
2059    let mut cursor = at;
2060    let mut guard = CycleGuard::new();
2061    loop {
2062        guard.step(cursor)?;
2063        let head = archive.at(cursor, HEADER_PROBE)?;
2064        if head.is_empty() {
2065            bail!("the archive holds no entry at offset {cursor}");
2066        }
2067        let (t, size, n) = crate::pack_walk::type_and_size_of(&head)?;
2068        match t {
2069            ObjType::Commit | ObjType::Tree | ObjType::Blob | ObjType::Tag => {
2070                chain.push(Link { obj_type: t, size, data_at: cursor + n as u64 });
2071                break;
2072            }
2073            ObjType::OfsDelta => {
2074                let (distance, d) = crate::pack_walk::ofs_distance_of(head.get(n..).unwrap_or(&[]))?;
2075                let base = cursor.checked_sub(distance).ok_or_else(|| {
2076                    anyhow!(
2077                        "the ofs-delta at archive offset {cursor} names a base {distance} bytes \
2078                         back, which is before the start of the archive"
2079                    )
2080                })?;
2081                chain.push(Link { obj_type: t, size, data_at: cursor + (n + d) as u64 });
2082                cursor = base;
2083            }
2084            ObjType::RefDelta => {
2085                let base_oid = head
2086                    .get(n..n + oid_len)
2087                    .ok_or_else(|| {
2088                        anyhow!(
2089                            "the ref-delta entry at archive offset {cursor} has no base oid after \
2090                             its header"
2091                        )
2092                    })?
2093                    .to_vec();
2094                chain.push(Link { obj_type: t, size, data_at: cursor + (n + oid_len) as u64 });
2095                match offset_of_oid(&base_oid) {
2096                    Some(o) => cursor = o,
2097                    None => {
2098                        external = Some(base_oid);
2099                        break;
2100                    }
2101                }
2102            }
2103        }
2104    }
2105    // ── the bottom of the chain: a whole object, or somebody else's ────────
2106    let external_base = external.is_some();
2107    let (kind, mut base) = match external {
2108        Some(oid) => bases
2109            .content(&oid)?
2110            .ok_or_else(|| ResolveError::missing_base(&oid))?,
2111        None => {
2112            let last = chain.pop().expect("a chain that ended has a last link");
2113            let mut out = Vec::new();
2114            inflate_from(archive, last.data_at, last.size, &mut out)?;
2115            (whole_kind(last.obj_type)?, out)
2116        }
2117    };
2118
2119    // ── apply the deltas back down to the object that was asked for ────────
2120    let depth = chain.len() as u32;
2121    let mut delta = Vec::new();
2122    let mut out = Vec::new();
2123    while let Some(link) = chain.pop() {
2124        inflate_from(archive, link.data_at, link.size, &mut delta)?;
2125        out.clear();
2126        apply_delta_into(&base, &delta, &mut out)?;
2127        std::mem::swap(&mut base, &mut out);
2128    }
2129    let cost = ReadCost {
2130        depth,
2131        inflated: base.len() as u64,
2132        external_base,
2133    };
2134    Ok((kind, base, cost))
2135}
2136
2137/// Inflate `expected` bytes of zlib starting at `data_at`, a window at a time.
2138///
2139/// The compressed length is **not** knowable from the index — `objects.len` is
2140/// the entry's stored extent only for an entry the walk measured, and a chain
2141/// base reached by back-reference was not measured here. So the stream is its
2142/// own terminator: inflate until it says `StreamEnd` or until exactly the
2143/// declared number of bytes have come out, and disagree loudly if the two
2144/// answers differ. `out` is reserved once at the declared size and never grows.
2145fn inflate_from(
2146    archive: &dyn ArchiveBytes,
2147    data_at: u64,
2148    expected: u64,
2149    out: &mut Vec<u8>,
2150) -> Result<()> {
2151    let want = usize::try_from(expected).map_err(|_| anyhow!("object size overflows usize"))?;
2152    out.clear();
2153    out.reserve(want);
2154    let mut d = flate2::Decompress::new(true);
2155    let mut pos = data_at;
2156    while out.len() < want {
2157        let chunk = archive.at(pos, INFLATE_WINDOW)?;
2158        if chunk.is_empty() {
2159            bail!("the zlib stream at {data_at} runs past the end of the archive");
2160        }
2161        let before = d.total_in();
2162        let status = d
2163            .decompress_vec(&chunk, out, flate2::FlushDecompress::None)
2164            .map_err(|e| anyhow!("inflating the entry at {data_at}: {e}"))?;
2165        let consumed = d.total_in() - before;
2166        pos += consumed;
2167        if status == flate2::Status::StreamEnd {
2168            break;
2169        }
2170        if consumed == 0 {
2171            bail!(
2172                "the zlib stream at {data_at} stalled after {} of {expected} bytes",
2173                out.len()
2174            );
2175        }
2176    }
2177    if out.len() as u64 != expected {
2178        bail!(
2179            "the entry at {data_at} inflated to {} bytes, its header declared {expected}",
2180            out.len()
2181        );
2182    }
2183    Ok(())
2184}
2185
2186/// The real git type of a non-delta entry.
2187fn whole_kind(t: ObjType) -> Result<GitObjectKind> {
2188    Ok(match t {
2189        ObjType::Commit => GitObjectKind::Commit,
2190        ObjType::Tree => GitObjectKind::Tree,
2191        ObjType::Blob => GitObjectKind::Blob,
2192        ObjType::Tag => GitObjectKind::Tag,
2193        ObjType::OfsDelta | ObjType::RefDelta => {
2194            bail!("a delta entry has no type of its own — its base's type is the answer")
2195        }
2196    })
2197}
2198
2199/// Inflate one entry's stream into `out` (cleared first). The walk already
2200/// proved the length, so this reserves exactly the declared size and never
2201/// grows; a reused `out` allocates nothing.
2202fn inflate_into(pack: &[u8], e: &PackEntry, oid_len: usize, out: &mut Vec<u8>) -> Result<()> {
2203    let start = e.offset as usize;
2204    let end = start + e.len as usize;
2205    if end > pack.len() {
2206        bail!("entry at {start} runs past the end of the pack");
2207    }
2208    // Skip the header the walk already parsed: the zlib stream is what is left
2209    // after the type/size varint and any base reference.
2210    let header = header_len(&pack[start..end], e, oid_len)?;
2211    out.clear();
2212    out.reserve(e.uncompressed_size as usize);
2213    let mut d = flate2::Decompress::new(true);
2214    d.decompress_vec(&pack[start + header..end], out, flate2::FlushDecompress::Finish)
2215        .map_err(|err| anyhow!("inflating the entry at {start}: {err}"))?;
2216    if out.len() as u64 != e.uncompressed_size {
2217        bail!(
2218            "the entry at {start} inflated to {} bytes, the walk measured {}",
2219            out.len(),
2220            e.uncompressed_size
2221        );
2222    }
2223    Ok(())
2224}
2225
2226/// Inflate one entry's stream into a fresh buffer. What a test or a one-off
2227/// caller wants; the resolver uses [`inflate_into`].
2228#[cfg(test)]
2229fn inflate(pack: &[u8], e: &PackEntry, oid_len: usize) -> Result<Vec<u8>> {
2230    let mut out = Vec::new();
2231    inflate_into(pack, e, oid_len, &mut out)?;
2232    Ok(out)
2233}
2234
2235/// How many bytes of an entry are header rather than zlib stream. Recomputed
2236/// from the bytes rather than remembered, so it cannot drift from the walk's own
2237/// reading of them — and the walk's `len` is what bounds it.
2238fn header_len(entry: &[u8], e: &PackEntry, oid_len: usize) -> Result<usize> {
2239    let mut i = 0usize;
2240    let mut cont = true;
2241    while cont {
2242        let b = *entry
2243            .get(i)
2244            .ok_or_else(|| anyhow!("the type/size header runs off the entry"))?;
2245        cont = b & 0x80 != 0;
2246        i += 1;
2247    }
2248    match e.obj_type {
2249        ObjType::OfsDelta => {
2250            let mut cont = true;
2251            while cont {
2252                let b = *entry
2253                    .get(i)
2254                    .ok_or_else(|| anyhow!("the ofs-delta distance runs off the entry"))?;
2255                cont = b & 0x80 != 0;
2256                i += 1;
2257            }
2258        }
2259        ObjType::RefDelta => i += oid_len,
2260        _ => {}
2261    }
2262    if i >= entry.len() {
2263        bail!("the entry's header consumes all of it, leaving no stream");
2264    }
2265    Ok(i)
2266}
2267
2268/// git's delta format: two sizes, then copy-from-base and insert-literal
2269/// instructions. Fresh-buffer form of [`apply_delta_into`].
2270#[cfg(test)]
2271fn apply_delta(base: &[u8], delta: &[u8]) -> Result<Vec<u8>> {
2272    let mut out = Vec::new();
2273    apply_delta_into(base, delta, &mut out)?;
2274    Ok(out)
2275}
2276
2277/// git's delta format applied into `out` (cleared first, reserved to the
2278/// declared target size).
2279fn apply_delta_into(base: &[u8], delta: &[u8], out: &mut Vec<u8>) -> Result<()> {
2280    let mut i = 0usize;
2281    let base_size = delta_varint(delta, &mut i)?;
2282    if base_size != base.len() as u64 {
2283        bail!(
2284            "the delta expects a base of {base_size} bytes, its base is {}",
2285            base.len()
2286        );
2287    }
2288    let target_size = delta_varint(delta, &mut i)?;
2289    out.clear();
2290    out.reserve(target_size as usize);
2291
2292    while i < delta.len() {
2293        let op = delta[i];
2294        i += 1;
2295        if op & 0x80 != 0 {
2296            // Copy from the base. Offset in up to 4 bytes, size in up to 3;
2297            // a zero size means 0x10000, which is git's own special case.
2298            let mut off = 0u64;
2299            for bit in 0..4 {
2300                if op & (1 << bit) != 0 {
2301                    off |= u64::from(*delta.get(i).ok_or_else(|| anyhow!("delta ends mid-copy"))?)
2302                        << (bit * 8);
2303                    i += 1;
2304                }
2305            }
2306            let mut size = 0u64;
2307            for bit in 0..3 {
2308                if op & (0x10 << bit) != 0 {
2309                    size |= u64::from(*delta.get(i).ok_or_else(|| anyhow!("delta ends mid-copy"))?)
2310                        << (bit * 8);
2311                    i += 1;
2312                }
2313            }
2314            if size == 0 {
2315                size = 0x1_0000;
2316            }
2317            let from = off as usize;
2318            let to = from
2319                .checked_add(size as usize)
2320                .ok_or_else(|| anyhow!("a delta copy range overflows"))?;
2321            if to > base.len() {
2322                bail!(
2323                    "a delta copies base[{from}..{to}] out of a {}-byte base",
2324                    base.len()
2325                );
2326            }
2327            out.extend_from_slice(&base[from..to]);
2328        } else {
2329            // Insert literal. A zero-length insert is not a valid instruction.
2330            let n = op as usize;
2331            if n == 0 {
2332                bail!("a delta carries a zero-length insert instruction");
2333            }
2334            let end = i
2335                .checked_add(n)
2336                .ok_or_else(|| anyhow!("a delta insert overflows"))?;
2337            if end > delta.len() {
2338                bail!("a delta insert of {n} bytes runs off the end");
2339            }
2340            out.extend_from_slice(&delta[i..end]);
2341            i = end;
2342        }
2343    }
2344
2345    if out.len() as u64 != target_size {
2346        bail!(
2347            "the delta declares a {target_size}-byte result and produced {}",
2348            out.len()
2349        );
2350    }
2351    Ok(())
2352}
2353
2354/// The delta header's little-endian 7-bit varint. Not the pack entry's varint.
2355fn delta_varint(b: &[u8], i: &mut usize) -> Result<u64> {
2356    let mut v = 0u64;
2357    let mut shift = 0u32;
2358    loop {
2359        let byte = *b
2360            .get(*i)
2361            .ok_or_else(|| anyhow!("a delta size varint runs off the end"))?;
2362        *i += 1;
2363        if shift >= 64 {
2364            bail!("a delta size varint is longer than a u64");
2365        }
2366        v |= u64::from(byte & 0x7f) << shift;
2367        shift += 7;
2368        if byte & 0x80 == 0 {
2369            return Ok(v);
2370        }
2371    }
2372}
2373
2374#[cfg(test)]
2375mod tests {
2376    use super::*;
2377    use crate::object::canonical;
2378    use std::path::{Path, PathBuf};
2379
2380    /// A label that answers `.display()`, so a source that is a generated pack
2381    /// and one that is a path on disk name themselves the same way in a failure.
2382    struct Named<'a>(&'a str);
2383    impl Named<'_> {
2384        fn display(&self) -> &str {
2385            self.0
2386        }
2387    }
2388
2389    /// Parse a real `.idx` (v2) into `(oid, pack offset)` pairs. Twenty lines of
2390    /// a frozen format, and it is git's own answer to the question this module
2391    /// computes — which is what makes it a differential guard and not a
2392    /// restatement of our own code.
2393    fn read_idx(bytes: &[u8]) -> Vec<(Vec<u8>, u64)> {
2394        assert_eq!(&bytes[0..4], b"\xfftOc", "not an idx v2");
2395        assert_eq!(u32::from_be_bytes(bytes[4..8].try_into().unwrap()), 2);
2396        let fanout_end = 8 + 256 * 4;
2397        let n = u32::from_be_bytes(bytes[fanout_end - 4..fanout_end].try_into().unwrap()) as usize;
2398        let oids = fanout_end;
2399        let crcs = oids + n * 20;
2400        let offs = crcs + n * 4;
2401        let big = offs + n * 4;
2402        let mut out = Vec::with_capacity(n);
2403        for i in 0..n {
2404            let oid = bytes[oids + i * 20..oids + i * 20 + 20].to_vec();
2405            let raw = u32::from_be_bytes(bytes[offs + i * 4..offs + i * 4 + 4].try_into().unwrap());
2406            let offset = if raw & 0x8000_0000 != 0 {
2407                let j = (raw & 0x7fff_ffff) as usize;
2408                u64::from_be_bytes(bytes[big + j * 8..big + j * 8 + 8].try_into().unwrap())
2409            } else {
2410                u64::from(raw)
2411            };
2412            out.push((oid, offset));
2413        }
2414        out
2415    }
2416
2417    /// Real packs with their real indexes, from this machine's repositories.
2418    fn real_pairs(cap: usize) -> Vec<(PathBuf, PathBuf)> {
2419        let mut out = Vec::new();
2420        let Ok(repos) = std::fs::read_dir(Path::new("/home/rickard/git")) else {
2421            return out;
2422        };
2423        for repo in repos.flatten() {
2424            let dir = repo.path().join(".git/objects/pack");
2425            let Ok(files) = std::fs::read_dir(&dir) else {
2426                continue;
2427            };
2428            for f in files.flatten() {
2429                let p = f.path();
2430                if p.extension().is_some_and(|e| e == "pack")
2431                    && f.metadata().map(|m| m.len() < 32 << 20).unwrap_or(false)
2432                {
2433                    let idx = p.with_extension("idx");
2434                    if idx.exists() {
2435                        out.push((p, idx));
2436                        if out.len() >= cap {
2437                            return out;
2438                        }
2439                    }
2440                }
2441            }
2442        }
2443        out
2444    }
2445
2446    /// **Pack/index pairs to compare against, the generated corpus first.**
2447    ///
2448    /// 🔴 This used to be [`real_pairs`] alone, and that made the guard's corpus
2449    /// the developer's disk: it took the first six `.pack` files with an `.idx`
2450    /// beside them under `/home/rickard/git`, in `read_dir` order, and asserted
2451    /// it had found at least one. On a fresh clone, on CI, or on anyone else's
2452    /// machine it fails with *"no real pack found under /home/rickard/git — this
2453    /// guard has nothing to compare"* — proven 2026-08-25 by running the suite
2454    /// with a tmpfs over that directory.
2455    ///
2456    /// The corpus from [`crate::pack_fixture`] leads the list, and stock git
2457    /// writes its `.idx` — so the arbiter is still git and not us, but the pack
2458    /// it arbitrates is one this repository states in source. Real packs still
2459    /// follow when the machine has any: they are the only place a **thin** pack
2460    /// turns up, and finding those in the wild is a fact this guard records.
2461    fn sources() -> Vec<(String, Vec<u8>, Vec<u8>)> {
2462        let mut out = Vec::new();
2463        if std::process::Command::new("git")
2464            .arg("--version")
2465            .output()
2466            .is_ok()
2467        {
2468            let pack = crate::pack_fixture::corpus().pack.clone();
2469            let scratch = crate::store::tests::tmpdir("resolve-idx-oracle");
2470            let idx = crate::git_oracle::git_idx_for(&scratch, "corpus.git", &pack)
2471                .expect("stock git indexes the generated corpus");
2472            out.push(("the generated corpus".to_string(), pack, idx));
2473        }
2474        for (pack_path, idx_path) in real_pairs(6) {
2475            let (Ok(pack), Ok(idx)) = (std::fs::read(&pack_path), std::fs::read(&idx_path)) else {
2476                continue;
2477            };
2478            out.push((pack_path.display().to_string(), pack, idx));
2479        }
2480        out
2481    }
2482
2483    /// **git's own `.idx` is the arbiter.** For every real pack on this machine,
2484    /// every oid we compute and the offset we computed it at must appear in
2485    /// git's index, and the two sets must have the same size. A delta applier
2486    /// that is subtly wrong produces a different hash and cannot pass this.
2487    ///
2488    /// Seen RED by changing `size = 0x1_0000` to `size = 0x1_000` in
2489    /// [`apply_delta`] — git's zero-size copy special case:
2490    /// ".../facett/...pack-62582dd0…: the delta declares a 149392-byte result and
2491    /// produced 87952". A single wrong constant in the delta applier and a real
2492    /// repository's pack stops resolving.
2493    ///
2494    /// MEASURED while writing it: **10 899 objects over 3 packs** match git's own
2495    /// `.idx` oid for oid and offset for offset, and **3 of the 6 real packs on
2496    /// this machine are thin** — they delta against objects that are not in them,
2497    /// which is the §14 hole showing up in the wild rather than in theory.
2498    #[test]
2499    fn every_oid_we_compute_is_the_oid_git_wrote_in_its_idx() {
2500        let pairs = sources();
2501        assert!(
2502            !pairs.is_empty(),
2503            "neither stock git nor a real pack is available, so this guard has nothing to \
2504             compare and must not pass silently"
2505        );
2506        let mut objects = 0usize;
2507        let mut full = 0usize;
2508        let mut thin = 0usize;
2509        for (pack_path, pack, idx_bytes) in pairs {
2510            let pack_path = Named(&pack_path);
2511            let idx = read_idx(&idx_bytes);
2512            // MEASURED, and a surprise worth recording: real packs on this
2513            // machine DO delta against objects outside themselves. Such a pack
2514            // cannot be compared here — resolving it is exactly what §14 leaves
2515            // open — but it must not make the guard pass vacuously either, so it
2516            // is counted and named and at least one pack has to compare in full.
2517            let ours = match resolve(&pack, GitHashKind::Sha1, 0, &NoBases) {
2518                Ok(r) => r,
2519                Err(e) => {
2520                    assert!(
2521                        e.to_string().contains("thin"),
2522                        "{}: {e}",
2523                        pack_path.display()
2524                    );
2525                    eprintln!("{}: THIN, external base — skipped", pack_path.display());
2526                    thin += 1;
2527                    continue;
2528                }
2529            };
2530            full += 1;
2531
2532            assert_eq!(
2533                ours.len(),
2534                idx.len(),
2535                "{}: we resolved {} objects, git indexed {}",
2536                pack_path.display(),
2537                ours.len(),
2538                idx.len()
2539            );
2540            let theirs: HashMap<Vec<u8>, u64> = idx.into_iter().collect();
2541            for r in &ours {
2542                match theirs.get(&r.oid) {
2543                    Some(&off) => assert_eq!(
2544                        off,
2545                        r.offset,
2546                        "{}: {} is at {} in git's idx and we put it at {}",
2547                        pack_path.display(),
2548                        hex::encode(&r.oid),
2549                        off,
2550                        r.offset
2551                    ),
2552                    None => panic!(
2553                        "{}: we computed {} at offset {}, which git's idx does not contain — the \
2554                         resolution is wrong",
2555                        pack_path.display(),
2556                        hex::encode(&r.oid),
2557                        r.offset
2558                    ),
2559                }
2560            }
2561            eprintln!("{}: {} objects agree with git", pack_path.display(), ours.len());
2562            objects += ours.len();
2563        }
2564        eprintln!("{objects} objects over {full} packs agree with git's own .idx; {thin} thin");
2565        assert!(full > 0, "every pack compared was thin — nothing was compared");
2566        assert!(objects > 1000, "only {objects} objects compared");
2567    }
2568
2569    /// The three facts the objects table needs and `.idx` cannot supply: the
2570    /// **stored** type (delta or not), the **resolved** type, and the
2571    /// post-resolution size. Asserted against the pack's own delta structure, on
2572    /// a real pack, so it cannot be satisfied by a store-and-echo.
2573    ///
2574    /// Seen RED by resolving a delta to `GitObjectKind::Blob` instead of to its
2575    /// base's kind: "a delta resolves to its base's type:
2576    /// c6d50e761d025c179ffef1c3d68c56aae0a53270 vs base
2577    /// 8e15741296ebe3a0508522b76fe2ec4982396b6d — left: Blob, right: Commit".
2578    ///
2579    /// MEASURED on that pack: 4292 deltas, 2715 of them resolving to something
2580    /// other than a blob, and **all 4292** carrying a resolved size unlike their
2581    /// delta-stream size.
2582    #[test]
2583    fn a_delta_carries_its_stored_type_and_its_resolved_type_and_size() {
2584        let pack = crate::pack_fixture::corpus().pack.clone();
2585        let pack_path = Named("the generated corpus");
2586        let ours = resolve(&pack, GitHashKind::Sha1, 0, &NoBases).unwrap();
2587        // One walk, reused. Walking per entry turns this guard into an O(n^2)
2588        // pass over a 7000-object pack and it stops finishing.
2589        let stream_size: HashMap<u64, u64> = walk(&pack, 20)
2590            .unwrap()
2591            .entries
2592            .iter()
2593            .map(|e| (e.offset, e.uncompressed_size))
2594            .collect();
2595
2596        let deltas: Vec<&Resolved> = ours
2597            .iter()
2598            .filter(|r| matches!(r.stored_type, ObjType::OfsDelta | ObjType::RefDelta))
2599            .collect();
2600        assert!(
2601            !deltas.is_empty(),
2602            "{} carries no deltas, so it proves nothing about resolution",
2603            pack_path.display()
2604        );
2605
2606        // A delta's resolved type is a real git type, never a delta code, and it
2607        // matches its base's.
2608        let by_offset: HashMap<u64, &Resolved> = ours.iter().map(|r| (r.offset, r)).collect();
2609        let mut non_blob = 0usize;
2610        for d in &deltas {
2611            assert!(
2612                matches!(
2613                    d.kind,
2614                    GitObjectKind::Blob
2615                        | GitObjectKind::Tree
2616                        | GitObjectKind::Commit
2617                        | GitObjectKind::Tag
2618                ),
2619                "a resolved type must be a real git type"
2620            );
2621            if d.kind != GitObjectKind::Blob {
2622                non_blob += 1;
2623            }
2624            if d.delta_base != 0 {
2625                let base = by_offset[&d.delta_base];
2626                assert_eq!(
2627                    d.kind,
2628                    base.kind,
2629                    "a delta resolves to its base's type: {} vs base {}",
2630                    hex::encode(&d.oid),
2631                    hex::encode(&base.oid)
2632                );
2633                // The delta_base column addresses BYTES, and those bytes are an
2634                // entry we also resolved. An ordinal could not be checked like
2635                // this at all.
2636                assert!(
2637                    base.offset < d.offset,
2638                    "an ofs-delta base is always earlier in the pack"
2639                );
2640            }
2641        }
2642        assert!(
2643            non_blob > 0,
2644            "{} deltas and not one resolved to a tree/commit/tag — the resolved type is not being \
2645             taken from the base",
2646            deltas.len()
2647        );
2648
2649        // The post-resolution size is the payload's real length, not the delta
2650        // stream's. For a delta the two are almost never equal.
2651        let differing = deltas
2652            .iter()
2653            .filter(|d| stream_size[&d.offset] != d.uncompressed_size)
2654            .count();
2655        assert!(
2656            differing > 0,
2657            "not one delta's resolved size differs from its delta-stream size, which cannot be \
2658             true of a real pack — the resolved size is being copied from the header"
2659        );
2660        eprintln!(
2661            "{}: {} deltas, {non_blob} non-blob, {differing} with a resolved size unlike the \
2662             stream size",
2663            pack_path.display(),
2664            deltas.len()
2665        );
2666    }
2667
2668    /// A thin pack is refused **by name**, and the refusal says what decision is
2669    /// missing. It is not resolved to a wrong oid and it is not silently dropped.
2670    ///
2671    /// Seen RED by making the no-progress arm `bail!("stuck")` instead of
2672    /// `ResolveError::missing_base`: "names the base: stuck".
2673    #[test]
2674    fn a_thin_pack_is_refused_by_name_and_says_what_is_missing() {
2675        // One ref-delta against an oid nothing can supply.
2676        let base_oid = vec![0x5a; 20];
2677        let mut pack = b"PACK".to_vec();
2678        pack.extend_from_slice(&2u32.to_be_bytes());
2679        pack.extend_from_slice(&1u32.to_be_bytes());
2680        // A delta whose base is 4 bytes and whose target is 4 bytes: copy all.
2681        let delta = vec![0x04, 0x04, 0x90, 0x04];
2682        pack.push(0x70 | (delta.len() as u8 & 0x0f)); // type 7, size 4
2683        pack.extend_from_slice(&base_oid);
2684        pack.extend_from_slice(&{
2685            use std::io::Write;
2686            let mut e = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::default());
2687            e.write_all(&delta).unwrap();
2688            e.finish().unwrap()
2689        });
2690        pack.extend_from_slice(&[0u8; 20]);
2691
2692        let err = resolve(&pack, GitHashKind::Sha1, 0, &NoBases)
2693            .expect_err("a thin pack with no base source cannot resolve");
2694        let msg = err.to_string();
2695        assert!(msg.contains(&hex::encode(&base_oid)), "names the base: {msg}");
2696        assert!(msg.contains("§14"), "says which decision is missing: {msg}");
2697
2698        // With a source that CAN supply it, the same pack resolves — so the
2699        // refusal is about the missing content, not about ref-deltas.
2700        struct One(Vec<u8>);
2701        impl BaseSource for One {
2702            fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
2703                Ok((oid == self.0.as_slice()).then(|| (GitObjectKind::Blob, b"abcd".to_vec())))
2704            }
2705        }
2706        let ok = resolve(&pack, GitHashKind::Sha1, 0, &One(base_oid)).expect("with a base it works");
2707        assert_eq!(ok.len(), 1);
2708        assert_eq!(ok[0].kind, GitObjectKind::Blob);
2709        assert_eq!(ok[0].uncompressed_size, 4);
2710        assert_eq!(
2711            ok[0].oid,
2712            GitHashKind::Sha1.oid_of(&canonical(GitObjectKind::Blob, b"abcd")),
2713            "the resolved object is the base copied whole, so it hashes as `abcd`"
2714        );
2715    }
2716
2717    /// **The `delta_base` column, materialised — from a real pack, through all
2718    /// three index layouts and the read stack, and back out again.**
2719    ///
2720    /// This is the guard the column was added for. It asserts *applied output*:
2721    /// resolve a real pack, build every arm over the rows, and for every
2722    /// `OFS_DELTA` read the base offset **back out of the index** and require it
2723    /// to land on an object the same index holds — earlier in the archive, and
2724    /// with a byte extent that contains it. An ordinal in that column could not
2725    /// satisfy this at all, which is §13's decision checked rather than quoted.
2726    ///
2727    /// The three arms are compared row for row, because that is what caught a
2728    /// swapped `len`/`size` write in the previous sweep while two of the three
2729    /// stayed green.
2730    ///
2731    /// Seen RED three times, each for a different way the column can be wrong
2732    /// while every other fact stays right:
2733    ///
2734    /// 1. **Never written.** `Resolved::index_entry` returning `delta_base: 0`:
2735    ///    "FourTables: b76687fbb34020672608bcb1a5e027c2d8b0346a is an ofs-delta
2736    ///    with no recorded base".
2737    /// 2. **Written from the wrong field.** `delta_base: self.offset` — a
2738    ///    plausible archive offset that is non-zero, inside the pack, and lands
2739    ///    on a real entry boundary, so only the identity check catches it:
2740    ///    "FourTables: b76687fb… records base 91272 which is its own offset,
2741    ///    not its base's".
2742    /// 3. **Written in the wrong coordinate space.** `resolve_walked` yielding
2743    ///    `*o` instead of `o + archive_offset`, i.e. pack-relative rather than
2744    ///    archive-relative — the bug that an *ordinal* column would have had no
2745    ///    way to express and no way to detect: "FourTables: b76687fb… records
2746    ///    base offset 86534, which starts no object in this index". This is why
2747    ///    the fixture resolves at `AT = 4096` rather than at 0; at 0 the two
2748    ///    spaces coincide and the guard would be blind.
2749    ///
2750    /// All three restored.
2751    #[test]
2752    fn the_delta_base_column_survives_all_three_arms_over_the_corpus() {
2753        use crate::index_layout::{
2754            FourTables, IndexEntry, ObjType, ObjectIndex, OneTableFourColumns, PackedPayload,
2755        };
2756        use crate::read_stack::{ObjectReadStack, RebuildTriggers};
2757
2758        let pack = crate::pack_fixture::corpus().pack.clone();
2759        // A non-zero archive offset on purpose: `delta_base` is in the archive's
2760        // coordinate space, so a resolver that forgot to rebase it would give
2761        // pack-relative values that no longer locate anything.
2762        const AT: u64 = 4096;
2763        let rows = resolve(&pack, GitHashKind::Sha1, AT, &NoBases).unwrap();
2764        let entries: Vec<IndexEntry> = rows.iter().map(|r| r.index_entry()).collect();
2765
2766        let a = FourTables::build(&entries).unwrap();
2767        let b = OneTableFourColumns::build(&entries).unwrap();
2768        let c = PackedPayload::build(&entries).unwrap();
2769        let stack = ObjectReadStack::<OneTableFourColumns>::in_memory(RebuildTriggers::manual())
2770            .unwrap();
2771        stack.append(&entries).unwrap();
2772        stack.rebuild().unwrap();
2773
2774        // Every extent this pack's objects occupy, read back out of the index.
2775        let extents: HashMap<u64, u64> = entries
2776            .iter()
2777            .map(|e| {
2778                let r = a.lookup(&e.oid).expect("a stored oid resolves");
2779                (r.offset, r.len)
2780            })
2781            .collect();
2782        assert_eq!(extents.len(), entries.len(), "two objects share an offset");
2783
2784        let named: [(&str, &dyn ObjectIndex); 4] =
2785            [("FourTables", &a), ("OneTableFourColumns", &b), ("PackedPayload", &c), ("ObjectReadStack", &stack)];
2786        let mut ofs_deltas = 0usize;
2787        for (name, idx) in named {
2788            let mut with_base = 0usize;
2789            for e in &entries {
2790                let row = idx.lookup(&e.oid).expect("a stored oid resolves");
2791                // All four layouts, row for row, on every fact.
2792                assert_eq!(
2793                    row,
2794                    a.lookup(&e.oid).unwrap(),
2795                    "{name} disagrees with FourTables on {}",
2796                    hex::encode(&e.oid)
2797                );
2798                if row.obj_type != ObjType::OfsDelta {
2799                    continue;
2800                }
2801                with_base += 1;
2802                assert_ne!(
2803                    row.delta_base, 0,
2804                    "{name}: {} is an ofs-delta with no recorded base",
2805                    hex::encode(&e.oid)
2806                );
2807                assert_ne!(
2808                    row.delta_base,
2809                    row.offset,
2810                    "{name}: {} records base {} which is its own offset, not its base's",
2811                    hex::encode(&e.oid),
2812                    row.delta_base
2813                );
2814                assert!(
2815                    row.delta_base >= AT,
2816                    "{name}: base {} is below the archive offset {AT} — it was never rebased",
2817                    row.delta_base
2818                );
2819                // The whole claim: it addresses an entry this index holds.
2820                let base_len = extents.get(&row.delta_base).unwrap_or_else(|| {
2821                    panic!(
2822                        "{name}: {} records base offset {}, which starts no object in this index",
2823                        hex::encode(&e.oid),
2824                        row.delta_base
2825                    )
2826                });
2827                assert!(
2828                    row.delta_base + base_len <= row.offset,
2829                    "{name}: the base at {} (+{base_len}) overlaps the delta at {}",
2830                    row.delta_base,
2831                    row.offset
2832                );
2833            }
2834            assert!(
2835                with_base > 0,
2836                "{name}: {} ofs-deltas resolved and 0 of them carry a base offset — the column \
2837                 is not being written",
2838                entries
2839                    .iter()
2840                    .filter(|e| e.obj_type == ObjType::OfsDelta)
2841                    .count()
2842            );
2843            ofs_deltas = with_base;
2844        }
2845        assert!(
2846            ofs_deltas > 100,
2847            "the corpus carries only {ofs_deltas} ofs-deltas — too few to prove anything"
2848        );
2849        eprintln!(
2850            "{ofs_deltas} ofs-delta base offsets located their base entry in all four layouts"
2851        );
2852    }
2853
2854    /// **A base that several entries delta against survives until the last one
2855    /// has had it.**
2856    ///
2857    /// This replaces `a_blob_that_is_nobodys_base_is_hashed_and_dropped`, which
2858    /// read the retention policy off `Resolved::payload`. That field is gone
2859    /// (see the struct's docs), and the property it stood for — peak memory is
2860    /// the live base set, not the object set — is measured in bytes now by
2861    /// `tests/resolve_peak_memory.rs` rather than inferred from a flag.
2862    ///
2863    /// What is left here is the *risk* the reference counting introduced, and it
2864    /// is the one that matters: an off-by-one that frees a base while a
2865    /// dependent still needs it. That cannot be a silent wrong answer —
2866    /// `resolve_walked` fails the whole pack by name — so the assertion is that
2867    /// a pack full of shared bases resolves at all.
2868    ///
2869    /// **The test refuses to run on a pack that could not expose the bug.** A
2870    /// corpus of only single-dependent chains would pass this with the counting
2871    /// deleted entirely, which is the blind-guard shape LAW 2 is about, so the
2872    /// fan-out is asserted before the resolution is.
2873    ///
2874    /// Seen RED by `*left -= 1;` → `*left = 0;` in [`consume`] (drop every base
2875    /// on first use): "entry at offset 3141 deltas against offset 2724 which no
2876    /// entry starts at — the pack is corrupt", on the first pack tried.
2877    #[test]
2878    fn a_base_many_entries_share_outlives_all_of_them() {
2879        let mut checked = 0usize;
2880        let mut widest = 0usize;
2881        let mut corpora: Vec<(String, Vec<u8>)> =
2882            vec![("the generated corpus".to_string(), crate::pack_fixture::corpus().pack.clone())];
2883        for (pack_path, _) in real_pairs(4) {
2884            if let Ok(bytes) = std::fs::read(&pack_path) {
2885                corpora.push((pack_path.display().to_string(), bytes));
2886            }
2887        }
2888        for (pack_path, pack) in corpora {
2889            let pack_path = Named(&pack_path);
2890            let w = walk(&pack, 20).unwrap();
2891
2892            // How many entries name each base. Anything above 1 is a base the
2893            // counting has to hold across more than one consumer.
2894            let mut uses: HashMap<u64, usize> = HashMap::new();
2895            for e in &w.entries {
2896                if let DeltaBase::Offset(o) = e.delta_base {
2897                    *uses.entry(o).or_default() += 1;
2898                }
2899            }
2900            let shared = uses.values().filter(|n| **n > 1).count();
2901            let fan_out = uses.values().copied().max().unwrap_or(0);
2902            if shared == 0 {
2903                continue;
2904            }
2905            widest = widest.max(fan_out);
2906
2907            let rows = resolve(&pack, GitHashKind::Sha1, 0, &NoBases)
2908                .unwrap_or_else(|e| panic!("{}: {e}", pack_path.display()));
2909            assert_eq!(
2910                rows.len(),
2911                w.entries.len(),
2912                "{}: {} of {} entries resolved",
2913                pack_path.display(),
2914                rows.len(),
2915                w.entries.len()
2916            );
2917            eprintln!(
2918                "{}: {shared} bases shared by more than one entry, widest fan-out {fan_out}, all \
2919                 {} entries resolved",
2920                pack_path.display(),
2921                rows.len()
2922            );
2923            checked += 1;
2924        }
2925        assert!(
2926            checked > 0 && widest > 2,
2927            "no pack compared had a base shared by more than two entries (checked {checked}, \
2928             widest {widest}) — this test cannot see an early free and must not report a pass"
2929        );
2930    }
2931
2932    /// The corpus plus whatever real packs the box has, as `(name, bytes)`.
2933    fn corpora(cap: usize) -> Vec<(String, Vec<u8>)> {
2934        let mut out: Vec<(String, Vec<u8>)> =
2935            vec![("the generated corpus".to_string(), crate::pack_fixture::corpus().pack.clone())];
2936        for (pack_path, _) in real_pairs(cap) {
2937            if let Ok(bytes) = std::fs::read(&pack_path) {
2938                out.push((pack_path.display().to_string(), bytes));
2939            }
2940        }
2941        out
2942    }
2943
2944    /// Every `(oid, offset, size, delta_base)` the resolver produced, sorted.
2945    fn rows_of(rows: &[Resolved]) -> Vec<(Vec<u8>, u64, u64, u64)> {
2946        let mut v: Vec<_> = rows
2947            .iter()
2948            .map(|r| (r.oid.clone(), r.offset, r.uncompressed_size, r.delta_base))
2949            .collect();
2950        v.sort();
2951        v
2952    }
2953
2954    fn stream(pack: &[u8], limits: ResolveLimits<'_>) -> (Vec<Resolved>, ResolveReport, u64) {
2955        let w = walk(pack, 20).unwrap();
2956        let mut out = Vec::new();
2957        let mut calls = 0u64;
2958        let report = resolve_streaming(
2959            pack,
2960            &w,
2961            GitHashKind::Sha1,
2962            0,
2963            &NoBases,
2964            &NoSink,
2965            limits,
2966            &mut |rows| {
2967                calls += 1;
2968                out.extend(rows.drain(..));
2969                Ok(())
2970            },
2971        )
2972        .unwrap();
2973        (out, report, calls)
2974    }
2975
2976    /// **The bounded base set: same rows, hard ceiling, re-inflations counted.**
2977    ///
2978    /// A budget of one byte is the harshest ceiling the walk can be given: only
2979    /// the base the very next delta reads is held, and every shallower one on
2980    /// the path is dropped and inflated again from the pack. The oids, sizes
2981    /// and delta bases must come out identical to the unbounded pass, the peak
2982    /// must never exceed one base, and the count of re-inflations must be
2983    /// positive — otherwise the ceiling was never hit and the test proved
2984    /// nothing.
2985    ///
2986    /// # What "the ceiling can bite here" means, and why the predicate changed
2987    ///
2988    /// It used to be "some base has more than one dependent", which was the
2989    /// shape the **fixpoint** evicted on: it held every base from its own entry
2990    /// until its last dependent's, so two live bases needed only two bases.
2991    /// The forest walk holds the *path*, so a second live base needs a path two
2992    /// deep with work still to do above it — a node with at least two children
2993    /// one of which is itself a base. A pack without that shape now genuinely
2994    /// never evicts, and asserting that it does would be asserting the old
2995    /// algorithm.
2996    #[test]
2997    fn a_bounded_base_set_yields_the_same_rows_under_a_hard_ceiling_and_counts_reinflations() {
2998        let mut checked = 0;
2999        for (name, pack) in corpora(4) {
3000            let w = walk(&pack, 20).unwrap();
3001            let mut uses: HashMap<u64, usize> = HashMap::new();
3002            for e in &w.entries {
3003                if let DeltaBase::Offset(o) = e.delta_base {
3004                    *uses.entry(o).or_default() += 1;
3005                }
3006            }
3007            let forest = crate::forest::DeltaForest::build(&w).unwrap();
3008            let two_deep = (0..w.entries.len() as u32).any(|i| {
3009                let kids = forest.children(i);
3010                kids.len() > 1 && kids[..kids.len() - 1].iter().any(|c| !forest.children(*c).is_empty())
3011            });
3012            if !uses.values().any(|n| *n > 1) || !two_deep {
3013                continue;
3014            }
3015            let (free, free_report, _) = stream(&pack, ResolveLimits::none());
3016            assert_eq!(free_report.reinflations, 0, "{name}: unbounded must never re-inflate");
3017            assert_eq!(free_report.evictions, 0);
3018            let largest_base = free
3019                .iter()
3020                .filter(|r| uses.contains_key(&r.offset))
3021                .map(|r| r.uncompressed_size)
3022                .max()
3023                .unwrap();
3024            let tight = ResolveLimits { base_budget: 1, ..ResolveLimits::none() };
3025            let (bounded, report, _) = stream(&pack, tight);
3026            assert_eq!(rows_of(&free), rows_of(&bounded), "{name}: the ceiling changed the output");
3027            assert!(report.reinflations > 0, "{name}: no re-inflation — the ceiling never bit");
3028            assert!(report.evictions > 0, "{name}: nothing evicted");
3029            assert!(
3030                report.peak_base_bytes <= largest_base,
3031                "{name}: peak held {} > the largest single base {largest_base} — the ceiling leaked",
3032                report.peak_base_bytes
3033            );
3034            assert!(
3035                free_report.peak_base_bytes >= report.peak_base_bytes,
3036                "{name}: the unbounded pass held less ({}) than the bounded one ({})",
3037                free_report.peak_base_bytes,
3038                report.peak_base_bytes
3039            );
3040            // A generous budget: same rows, fewer re-inflations than the
3041            // harsh one — the eviction is Belady, so more room is never worse.
3042            let mid = ResolveLimits { base_budget: largest_base * 3, ..ResolveLimits::none() };
3043            let (mid_rows, mid_report, _) = stream(&pack, mid);
3044            assert_eq!(rows_of(&free), rows_of(&mid_rows), "{name}: mid budget changed the output");
3045            assert!(
3046                mid_report.reinflations <= report.reinflations,
3047                "{name}: more room, more re-inflations ({} > {})",
3048                mid_report.reinflations,
3049                report.reinflations
3050            );
3051            assert!(mid_report.peak_base_bytes <= largest_base * 3);
3052            eprintln!(
3053                "{name}: {} objects, unbounded peak {} B; budget 1 B → peak {} B, {} evictions, {} re-inflations; budget {} B → {} re-inflations",
3054                free.len(),
3055                free_report.peak_base_bytes,
3056                report.peak_base_bytes,
3057                report.evictions,
3058                report.reinflations,
3059                largest_base * 3,
3060                mid_report.reinflations
3061            );
3062            checked += 1;
3063        }
3064        assert!(
3065            checked > 0,
3066            "no pack in the corpus has a path the ceiling can bite on — nothing was proven"
3067        );
3068    }
3069
3070    /// Rows arrive in batches of the size asked for, the last one partial, and
3071    /// nothing is lost or doubled across the batch boundary.
3072    #[test]
3073    fn rows_are_handed_over_in_batches_and_the_union_is_the_whole_pack() {
3074        let pack = crate::pack_fixture::corpus().pack.clone();
3075        let n = walk(&pack, 20).unwrap().entries.len();
3076        assert!(n > 7);
3077        let (all, report, calls) = stream(&pack, ResolveLimits { row_batch: 7, ..ResolveLimits::none() });
3078        assert_eq!(all.len(), n);
3079        assert_eq!(calls, n.div_ceil(7) as u64, "batches of 7 over {n} rows");
3080        assert_eq!(report.batches, calls);
3081        assert_eq!(report.rows, n as u64);
3082        let (one, _, calls1) = stream(&pack, ResolveLimits { row_batch: usize::MAX, ..ResolveLimits::none() });
3083        assert_eq!(calls1, 1);
3084        assert_eq!(rows_of(&all), rows_of(&one));
3085        let mut oids: Vec<_> = all.iter().map(|r| r.oid.clone()).collect();
3086        oids.sort();
3087        oids.dedup();
3088        assert_eq!(oids.len(), n, "a row was doubled across a batch boundary");
3089    }
3090
3091    /// **The watchdog inside a pack**: with a fake meminfo under the floor the
3092    /// resolver drops every held base, flushes, and waits until the fake says
3093    /// memory is back — then finishes with identical rows.
3094    #[test]
3095    fn the_watchdog_pauses_a_resolve_mid_pack_drops_the_bases_and_the_rows_are_unchanged() {
3096        use std::sync::Arc;
3097        use std::time::Duration;
3098        let dir = crate::store::tests::tmpdir("resolve-watchdog");
3099        let meminfo = dir.join("meminfo");
3100        let write = |avail_kib: u64| {
3101            std::fs::write(
3102                &meminfo,
3103                format!("MemTotal:       16000000 kB\nMemAvailable:   {avail_kib} kB\n"),
3104            )
3105            .unwrap()
3106        };
3107        write(1_000_000);
3108        let dog = Arc::new(MemWatchdog::with(&meminfo, 3_200_000 * 1024, Duration::from_millis(5)));
3109        let pack = crate::pack_fixture::corpus().pack.clone();
3110        let (free, _, _) = stream(&pack, ResolveLimits::none());
3111        let m2 = meminfo.clone();
3112        // **Raise only once the pause is actually open.** Sleeping a flat 150 ms
3113        // and hoping the resolver got there first is a race the loaded box
3114        // loses: under a full parallel suite the first checkpoint can land at
3115        // 140 ms, and the assertion below then reads a 10 ms pause. Wait for
3116        // `is_paused`, then hold for the interval the assertion is about.
3117        let d2 = dog.clone();
3118        let raiser = std::thread::spawn(move || {
3119            let deadline = std::time::Instant::now() + Duration::from_secs(30);
3120            while !d2.is_paused() && std::time::Instant::now() < deadline {
3121                std::thread::sleep(Duration::from_millis(2));
3122            }
3123            std::thread::sleep(Duration::from_millis(150));
3124            std::fs::write(&m2, "MemTotal:       16000000 kB\nMemAvailable:   9000000 kB\n").unwrap();
3125        });
3126        let limits = ResolveLimits {
3127            check_every: 3,
3128            watchdog: Some(dog.clone()),
3129            ..ResolveLimits::none()
3130        };
3131        let (rows, report, _) = stream(&pack, limits);
3132        raiser.join().unwrap();
3133        assert_eq!(rows_of(&free), rows_of(&rows), "the pause changed the output");
3134        assert_eq!(report.pauses, 1, "one pause: the floor was breached once and the raise ended it");
3135        assert!(report.watchdog_drops >= 1, "the pause must drop the held bases");
3136        assert!(report.paused_ms >= 100, "waited {} ms; must have waited for the raise", report.paused_ms);
3137        assert_eq!(dog.pauses(), 1);
3138    }
3139
3140    /// **A short box loses workers, not the pack.**
3141    ///
3142    /// The watchdog says the floor is breached and never un-says it. With one
3143    /// worker that is a pause and then a refusal by name. With four it must
3144    /// first be a *shed*: every worker above the lowest unit in flight gives
3145    /// its core back and waits, the round drains onto one core, and the pack
3146    /// still comes out — byte for byte the same stream a one-worker run makes.
3147    ///
3148    /// MEASURED on the rig and this is what it is for: the 16 GB appliance
3149    /// finishes the 8.12 GiB kernel mirror at one worker with 5.55 GiB to
3150    /// spare, and refuses it at 96 % with two and 86 % with four.
3151    #[test]
3152    fn a_short_box_sheds_its_workers_before_it_refuses_the_pack() {
3153        use std::sync::Arc;
3154        use std::time::Duration;
3155        let dir = crate::store::tests::tmpdir("resolve-shed");
3156        let meminfo = dir.join("meminfo");
3157        // Breached from the first checkpoint, and raised only once the LAST
3158        // worker standing has run out of others to shed and opened a pause —
3159        // which is the shape the rig measured: the box was short because of
3160        // the extra workers, and giving them back is what makes room.
3161        std::fs::write(&meminfo, "MemTotal:       16000000 kB\nMemAvailable:   1000000 kB\n").unwrap();
3162        let pack = crate::pack_fixture::corpus().pack.clone();
3163        let one = tape_of(&pack, 1, false).0;
3164        let dog = Arc::new(
3165            MemWatchdog::with(&meminfo, 3_200_000 * 1024, Duration::from_millis(2))
3166                .bounded(1 << 20, Duration::from_secs(30)),
3167        );
3168        let (m2, d2) = (meminfo.clone(), dog.clone());
3169        let raiser = std::thread::spawn(move || {
3170            let deadline = std::time::Instant::now() + Duration::from_secs(30);
3171            while !d2.is_paused() && std::time::Instant::now() < deadline {
3172                std::thread::sleep(Duration::from_millis(2));
3173            }
3174            std::fs::write(&m2, "MemTotal:       16000000 kB\nMemAvailable:   9000000 kB\n").unwrap();
3175        });
3176        let w = walk(&pack, 20).unwrap();
3177        let tape = Tape::default();
3178        let sink = TapeSink { tape: &tape, only_commits: false };
3179        let limits = ResolveLimits {
3180            workers: 4,
3181            row_batch: 64,
3182            lookahead_bytes: 1 << 16,
3183            check_every: 2,
3184            watchdog: Some(dog.clone()),
3185            ..ResolveLimits::none()
3186        };
3187        let report = resolve_streaming(
3188            &pack,
3189            &w,
3190            GitHashKind::Sha1,
3191            0,
3192            &NoBases,
3193            &sink,
3194            limits,
3195            &mut |rows| {
3196                tape.put(b'B', &[&(rows.len() as u64).to_le_bytes()]);
3197                for r in rows.iter() {
3198                    tape.put(
3199                        b'R',
3200                        &[
3201                            &r.oid,
3202                            &r.offset.to_le_bytes(),
3203                            &r.len.to_le_bytes(),
3204                            &[r.stored_type as u8, r.kind as u8],
3205                            &r.uncompressed_size.to_le_bytes(),
3206                            &r.delta_base.to_le_bytes(),
3207                        ],
3208                    );
3209                }
3210                rows.clear();
3211                Ok(())
3212            },
3213        );
3214        raiser.join().unwrap();
3215        let report = report.expect("a shed must not refuse the pack");
3216        assert!(
3217            report.shed > 0,
3218            "the box was short from the first checkpoint and NOTHING was shed — the guard \
3219             proved nothing (workers {}, pauses {})",
3220            report.workers,
3221            report.pauses
3222        );
3223        assert_eq!(report.rows, w.entries.len() as u64, "every entry is still a row");
3224        assert!(tape.take() == one, "the shed changed the stream");
3225    }
3226
3227    /// A cancel flag set mid-pack aborts at the next checkpoint with the named
3228    /// error — the pack is not finished and nothing pretends it was.
3229    #[test]
3230    fn a_cancel_aborts_the_pack_at_the_next_checkpoint() {
3231        let cancel = AtomicBool::new(true);
3232        let pack = crate::pack_fixture::corpus().pack.clone();
3233        let w = walk(&pack, 20).unwrap();
3234        let limits = ResolveLimits { check_every: 2, cancel: Some(&cancel), ..ResolveLimits::none() };
3235        let mut got = 0usize;
3236        let err = resolve_streaming(&pack, &w, GitHashKind::Sha1, 0, &NoBases, &NoSink, limits, &mut |rows| {
3237            got += rows.len();
3238            rows.clear();
3239            Ok(())
3240        })
3241        .unwrap_err();
3242        assert!(err.to_string().contains("cancelled"), "{err:#}");
3243        assert!(got < w.entries.len(), "the cancel did not stop the pack ({got} rows)");
3244    }
3245
3246    // ════════════════════════════════════════════════════════════════════════
3247    // THE PARALLEL RESOLVER — one stream out, whatever the worker count
3248    // ════════════════════════════════════════════════════════════════════════
3249
3250    /// **Everything the resolver hands out, as bytes.**
3251    ///
3252    /// Not a set and not a sorted digest: the *sequence*, with the sink's calls
3253    /// and the row callback's batches interleaved exactly as they happened. A
3254    /// comparison over sorted rows would pass on any ordering at all, and
3255    /// ordering is the whole thing a worker count could change.
3256    #[derive(Default)]
3257    struct Tape {
3258        bytes: std::sync::Mutex<Vec<u8>>,
3259    }
3260
3261    impl Tape {
3262        fn put(&self, tag: u8, parts: &[&[u8]]) {
3263            let mut b = self.bytes.lock().unwrap_or_else(|e| e.into_inner());
3264            b.push(tag);
3265            for p in parts {
3266                b.extend_from_slice(&(p.len() as u64).to_le_bytes());
3267                b.extend_from_slice(p);
3268            }
3269        }
3270        fn take(&self) -> Vec<u8> {
3271            std::mem::take(&mut self.bytes.lock().unwrap_or_else(|e| e.into_inner()))
3272        }
3273    }
3274
3275    /// A sink that records every call — and, on the second arm, one that says it
3276    /// wants only commits, which is the shipped `ExplodePolicy::Commits` shape.
3277    struct TapeSink<'t> {
3278        tape: &'t Tape,
3279        only_commits: bool,
3280    }
3281
3282    impl crate::exploded::PayloadSink for TapeSink<'_> {
3283        fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
3284            self.tape.put(b'E', &[oid, &[kind as u8], payload]);
3285            Ok(())
3286        }
3287        fn wants(&self, kind: GitObjectKind) -> bool {
3288            !self.only_commits || matches!(kind, GitObjectKind::Commit | GitObjectKind::Tag)
3289        }
3290    }
3291
3292    /// One resolve, recorded. `workers` is the only thing that varies.
3293    fn tape_of(pack: &[u8], workers: usize, only_commits: bool) -> (Vec<u8>, ResolveReport) {
3294        let w = walk(pack, 20).unwrap();
3295        let tape = Tape::default();
3296        let sink = TapeSink { tape: &tape, only_commits };
3297        let limits = ResolveLimits {
3298            workers,
3299            // Small enough that a real pack crosses it many times, so the
3300            // batch boundaries are part of what is being compared.
3301            row_batch: 64,
3302            // …and small enough that the lookahead ceiling is actually reached,
3303            // so the backpressure path is on the tape too.
3304            lookahead_bytes: 1 << 16,
3305            ..ResolveLimits::none()
3306        };
3307        let report = resolve_streaming(
3308            pack,
3309            &w,
3310            GitHashKind::Sha1,
3311            0,
3312            &NoBases,
3313            &sink,
3314            limits,
3315            &mut |rows| {
3316                tape.put(b'B', &[&(rows.len() as u64).to_le_bytes()]);
3317                for r in rows.iter() {
3318                    tape.put(
3319                        b'R',
3320                        &[
3321                            &r.oid,
3322                            &r.offset.to_le_bytes(),
3323                            &r.len.to_le_bytes(),
3324                            &[r.stored_type as u8, r.kind as u8],
3325                            &r.uncompressed_size.to_le_bytes(),
3326                            &r.delta_base.to_le_bytes(),
3327                        ],
3328                    );
3329                }
3330                rows.clear();
3331                Ok(())
3332            },
3333        )
3334        .expect("resolve");
3335        (tape.take(), report)
3336    }
3337
3338    /// **The determinism proof.** The same pack, resolved at 1, 2 and 4
3339    /// workers, produces byte-identical output — the same rows, in the same
3340    /// order, in the same batches, with the same sink calls between them.
3341    #[test]
3342    fn the_same_pack_resolves_to_the_same_bytes_at_1_2_and_4_workers() {
3343        let mut proved = 0usize;
3344        for (name, pack) in corpora(4) {
3345            for only_commits in [false, true] {
3346                let (one, r1) = tape_of(&pack, 1, only_commits);
3347                let (two, r2) = tape_of(&pack, 2, only_commits);
3348                let (four, r4) = tape_of(&pack, 4, only_commits);
3349                assert!(!one.is_empty(), "{name}: the tape is empty, so it proves nothing");
3350                assert_eq!(one.len(), two.len(), "{name}: 1 vs 2 workers, tape length");
3351                assert!(one == two, "{name}: 1 vs 2 workers, byte for byte");
3352                assert!(one == four, "{name}: 1 vs 4 workers, byte for byte");
3353                assert_eq!(r1.rows, r2.rows, "{name}");
3354                assert_eq!(r1.rows, r4.rows, "{name}");
3355                assert_eq!(r1.batches, r4.batches, "{name}");
3356                assert_eq!(r1.workers, 1, "{name}");
3357                proved += 1;
3358            }
3359        }
3360        assert!(proved >= 2, "no pack was compared — this guard proved nothing");
3361    }
3362
3363    /// 🔴 **LAW 2: the comparison above can fail.** If the tape were blind to
3364    /// order — a sorted digest, a set of oids, a row count — every one of those
3365    /// assertions would pass on a resolver that emitted its objects in any
3366    /// order at all. So: take the real 4-worker tape, move ONE row later in it,
3367    /// and prove the comparison rejects it.
3368    #[test]
3369    fn the_determinism_tape_rejects_a_stream_whose_order_moved() {
3370        let pack = corpora(2).into_iter().next().expect("a pack").1;
3371        let (good, _) = tape_of(&pack, 4, false);
3372        // Find the two `R` records the tape starts with and swap them. Nothing
3373        // about the *contents* changes; only the order does.
3374        let a = good.iter().position(|b| *b == b'R').expect("a row record");
3375        let rec = 1 + 6 * 8 + 20 + 8 + 8 + 2 + 8 + 8; // tag + 6 lengths + fields
3376        assert!(good.len() > a + 2 * rec, "the tape is too short to reorder");
3377        let mut bad = good.clone();
3378        for i in 0..rec {
3379            bad.swap(a + i, a + rec + i);
3380        }
3381        assert_ne!(good, bad, "the two rows were identical — swapping them proved nothing");
3382        assert!(good != bad, "an order-blind comparison would have called these equal");
3383    }
3384
3385    /// A sink that answers `wants = false` is **not called** for that kind, and
3386    /// the resolver still produces every row. This is what keeps the ordered
3387    /// hand-over from buffering 90 % of a pack's bytes.
3388    #[test]
3389    fn a_sink_that_wants_nothing_is_never_handed_a_payload() {
3390        for (name, pack) in corpora(3) {
3391            let w = walk(&pack, 20).unwrap();
3392            let tape = Tape::default();
3393            let sink = TapeSink { tape: &tape, only_commits: true };
3394            let mut rows = 0u64;
3395            let report = resolve_streaming(
3396                &pack,
3397                &w,
3398                GitHashKind::Sha1,
3399                0,
3400                &NoBases,
3401                &sink,
3402                ResolveLimits { workers: 4, ..ResolveLimits::none() },
3403                &mut |r| {
3404                    rows += r.len() as u64;
3405                    r.clear();
3406                    Ok(())
3407                },
3408            )
3409            .expect("resolve");
3410            assert_eq!(rows, w.entries.len() as u64, "{name}: every entry is a row");
3411            assert_eq!(report.rows, rows, "{name}");
3412            let calls = tape
3413                .take()
3414                .chunks(1)
3415                .filter(|c| c[0] == b'E')
3416                .count();
3417            // Every explode the sink DID get is a commit or a tag; and a pack
3418            // with any blob or tree at all must have been spared some.
3419            let non_commit = w
3420                .entries
3421                .iter()
3422                .filter(|e| matches!(e.obj_type, ObjType::Blob | ObjType::Tree))
3423                .count();
3424            if non_commit > 0 {
3425                assert!(
3426                    calls < w.entries.len(),
3427                    "{name}: the sink was handed all {} objects although it wanted only commits",
3428                    w.entries.len()
3429                );
3430            }
3431        }
3432    }
3433
3434    /// **A pack whose delta shape is CHOSEN, not found.** `roots` independent
3435    /// whole blobs, each the root of a binary tree of `depth` levels of
3436    /// `OFS_DELTA`s, every object exactly `body_bytes` long.
3437    ///
3438    /// The shape is the whole point of the guard below: `roots` roots is what
3439    /// gives the gatling fan-out `roots` independent units to hand four workers
3440    /// (`run_round` takes `min(workers, roots)`), the branching is what keeps a
3441    /// frame live while a deeper subtree runs — a base whose last child has not
3442    /// been reached yet is held — and one uniform object size is what makes
3443    /// "the largest single base" a number this source states rather than one a
3444    /// found pack happens to have.
3445    fn base_ceiling_pack(roots: usize, depth: u32, body_bytes: usize) -> Vec<u8> {
3446        use crate::pack_fixture::{build_pack, Item};
3447        // splitmix64, four lines, so the pack is byte-identical on every box:
3448        // incompressible bodies, so the pack's shape is the constants above and
3449        // not deflate's opinion of them.
3450        let mut state = 0x5eed_0000_0000_0001u64;
3451        let mut body = |n: usize| -> Vec<u8> {
3452            let mut v = Vec::with_capacity(n + 8);
3453            while v.len() < n {
3454                state = state.wrapping_add(0x9e37_79b9_7f4a_7c15);
3455                let mut z = state;
3456                z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
3457                z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
3458                v.extend_from_slice(&(z ^ (z >> 31)).to_le_bytes());
3459            }
3460            v.truncate(n);
3461            v
3462        };
3463        let mut items: Vec<Item> = Vec::new();
3464        for _ in 0..roots {
3465            let mut level = vec![items.len()];
3466            items.push(Item::Whole(GitObjectKind::Blob, body(body_bytes)));
3467            for _ in 1..depth {
3468                let mut next = Vec::with_capacity(level.len() * 2);
3469                for base in level {
3470                    for _ in 0..2 {
3471                        next.push(items.len());
3472                        items.push(Item::OfsDelta { base, body: body(body_bytes) });
3473                    }
3474                }
3475                level = next;
3476            }
3477        }
3478        build_pack(&items)
3479    }
3480
3481    /// **N workers get N walks, not N budgets.** The held-base ceiling is one
3482    /// atomic across every worker ([`workers_from_env`]'s doc says so in as many
3483    /// words: *"it is not N base budgets either"*), so four workers on the same
3484    /// pack hold what one budget allows plus what the walk structurally cannot
3485    /// drop — and nowhere near the four budgets a per-thread ceiling would give
3486    /// them.
3487    ///
3488    /// # 🔴 The two defects this replaces, both measured 2026-09-08
3489    ///
3490    /// It read `corpora(4)`, i.e. **whatever `.pack` files sat under
3491    /// `/home/rickard/git` in `read_dir` order**, and asserted
3492    /// `peak <= budget + biggest` with
3493    ///
3494    /// ```text
3495    /// let biggest = w.entries.iter().map(|e| e.uncompressed_size).max()
3496    /// ```
3497    ///
3498    /// ```text
3499    /// /home/rickard/git/Njord/.git/objects/pack/pack-7edf01cd….pack:
3500    ///   arm 1 held 267014 bytes against a 65536 budget and a 176423-byte largest object
3501    /// ```
3502    ///
3503    /// * **`biggest` was not the largest base.** `PackEntry::uncompressed_size`
3504    ///   is the size field of the *pack entry*, and for an `OFS_DELTA` that is
3505    ///   the length of the **delta stream**, not of the object it reconstructs.
3506    ///   The peak is a sum of *resolved base payloads*, so the old assertion
3507    ///   compared a peak in one unit against a maximum in another and was
3508    ///   simply not about the same bytes. The sibling guard
3509    ///   [`a_bounded_base_set_costs_re_inflations_not_rows`] already had this
3510    ///   right: it takes the maximum over the **resolved rows** whose offset is
3511    ///   used as a base, and that is what this one takes now.
3512    /// * **One base per worker, not one per pack.** `enforce_budget` never
3513    ///   drops a worker's *deepest* frame — it is the base the very next child
3514    ///   reads, and dropping it would re-inflate it on the following line — so
3515    ///   the floor under the ceiling is one held base **per walk**, and there
3516    ///   are `workers` walks. `budget + 1 × largest` was the serial arm's bound
3517    ///   applied to a four-worker arm. The bound that is actually promised is
3518    ///   `budget + workers × largest`: the shared ceiling, plus at most one
3519    ///   undroppable base per worker (which also absorbs the only race there
3520    ///   is — `live_path` is a relaxed atomic, so the peak one worker records
3521    ///   may include another worker's push that has not reached its own
3522    ///   `enforce_budget` yet).
3523    ///
3524    /// Neither correction weakens the guard, because the guard's *point* is the
3525    /// last assertion: `budget + workers × largest` is far below `workers ×
3526    /// budget`, which is what a per-thread ceiling would have allowed. The
3527    /// external corpus is gone with them — a test that needs another checkout
3528    /// on the same disk is not a test, and the pack is built here instead, sized
3529    /// so the ceiling demonstrably bites (`evictions > 0`, and the unbounded
3530    /// pass holds more than the bounded one).
3531    #[test]
3532    fn four_workers_hold_no_more_bases_than_the_one_budget_allows() {
3533        const ROOTS: usize = 4;
3534        const DEPTH: u32 = 8;
3535        const OBJECT: usize = 4 * 1024;
3536        // 4 roots × (2^8 − 1) = 1020 objects of 4 KiB. One walk holds at most
3537        // the seven bases above a leaf, so it wants 28 KiB and four walks want
3538        // 112 KiB — while the budget is 20 KiB: more than `workers × largest`
3539        // (16 KiB, so what binds is the ceiling and not the floor the walk
3540        // cannot drop) and less than even ONE walk wants (so it binds on the
3541        // serial arm too, and that arm is not a free pass).
3542        const BUDGET: u64 = 20 * 1024;
3543        let pack = base_ceiling_pack(ROOTS, DEPTH, OBJECT);
3544        let w = walk(&pack, 20).unwrap();
3545
3546        // Which offsets are used as a base, so "the largest base" is a maximum
3547        // over the objects the ceiling actually holds — resolved sizes, off the
3548        // rows, not pack-entry size fields.
3549        let mut is_base: std::collections::HashSet<u64> = std::collections::HashSet::new();
3550        for e in &w.entries {
3551            if let DeltaBase::Offset(o) = e.delta_base {
3552                is_base.insert(o);
3553            }
3554        }
3555
3556        let mut four_worker_peak = 0u64;
3557        for workers in [1usize, 4] {
3558            // The unbounded pass **at the same worker count**, because that is
3559            // the only thing the bounded one is comparable with: N walks hold N
3560            // paths whether or not there is a ceiling, so a one-worker baseline
3561            // would make the four-worker arm look like a leak.
3562            let free_limits = ResolveLimits { workers, ..ResolveLimits::none() };
3563            let (free, free_report, _) = stream(&pack, free_limits);
3564            assert_eq!(free_report.evictions, 0, "{workers} workers: the unbounded pass evicted");
3565            let largest = free
3566                .iter()
3567                .filter(|r| is_base.contains(&r.offset))
3568                .map(|r| r.uncompressed_size)
3569                .max()
3570                .expect("the pack has bases");
3571            assert_eq!(largest, OBJECT as u64, "every object in this pack is one size");
3572
3573            let limits = ResolveLimits { workers, base_budget: BUDGET, ..ResolveLimits::none() };
3574            let mut rows = Vec::new();
3575            let report = resolve_streaming(
3576                &pack,
3577                &w,
3578                GitHashKind::Sha1,
3579                0,
3580                &NoBases,
3581                &NoSink,
3582                limits,
3583                &mut |r| {
3584                    rows.extend(r.drain(..));
3585                    Ok(())
3586                },
3587            )
3588            .expect("resolve");
3589            assert_eq!(report.workers as usize, workers, "the fan-out took the worker count");
3590            assert_eq!(
3591                rows_of(&free),
3592                rows_of(&rows),
3593                "{workers} workers: the ceiling changed the output"
3594            );
3595            assert!(
3596                report.evictions > 0,
3597                "{workers} workers: nothing was evicted, so the ceiling never bit and this \
3598                 guard proves nothing"
3599            );
3600            assert!(
3601                free_report.peak_base_bytes > report.peak_base_bytes,
3602                "{workers} workers: unbounded held {} and bounded {} — the ceiling did nothing",
3603                free_report.peak_base_bytes,
3604                report.peak_base_bytes
3605            );
3606            let allowed = BUDGET + workers as u64 * largest;
3607            assert!(
3608                report.peak_base_bytes <= allowed,
3609                "{workers} workers: held {} bytes against a {BUDGET}-byte budget plus {workers} × \
3610                 the {largest}-byte largest base = {allowed}",
3611                report.peak_base_bytes
3612            );
3613            if workers == 4 {
3614                four_worker_peak = report.peak_base_bytes;
3615                // The whole claim, stated as the number a PER-THREAD ceiling
3616                // would have produced: four walks each allowed the budget plus
3617                // their own undroppable base.
3618                let per_thread = workers as u64 * (BUDGET + largest);
3619                assert!(
3620                    four_worker_peak < per_thread,
3621                    "four workers held {four_worker_peak} bytes, which is four ceilings of \
3622                     {BUDGET} + {largest} = {per_thread}: the budget is per thread, not per pack"
3623                );
3624            }
3625            eprintln!(
3626                "{workers} worker(s): peak {} B (unbounded {} B), budget {BUDGET} B, largest base \
3627                 {largest} B, {} evictions, {} re-inflations",
3628                report.peak_base_bytes,
3629                free_report.peak_base_bytes,
3630                report.evictions,
3631                report.reinflations
3632            );
3633        }
3634    }
3635}