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