znippy_plugin_git/git_ops.rs
1//! **`GitOps` — ours, typed, and base znippy never sees it.**
2//!
3//! §13.2: `ArchiveTypePlugin` is base znippy's contract and
4//! [`crate::NativeGitPlugin`] implements it — base stores, the plugin annotates,
5//! exactly as maven, python, rust-toolchain and media already work. `GitOps` is
6//! the *other* trait: the one gunnar calls, with git's own vocabulary in it.
7//!
8//! §13.3: `ArchiveTypePlugin::run_command(cmd, args)` is **CLI-only string
9//! dispatch** and delegates here. A server's hot path does not go through
10//! stringly-typed dispatch, so nothing below takes a `&str` command or an
11//! `args: &[String]`.
12//!
13//! ```text
14//! gunnar ──typed──► GitOps (this file, twelve methods)
15//! │
16//! ┌────────────────┼──────────────────┬─────────────┐
17//! ▼ ▼ ▼ ▼
18//! ArchiveWrite ObjectIndex RefLog Gc
19//! SafeWriter ObjectReadStack push log NewGeneration
20//! (blob fsync, (stree → Arrow (Arrow IPC (link, compact,
21//! journal row) → redb tail) frame = tx) verify, rename)
22//! ```
23//!
24//! Every arm under that diagram already existed and was already measured. This
25//! file adds no storage, no index and no durability mechanism of its own — if a
26//! method here looks like it is inventing one, that is a bug.
27//!
28//! # The one thing that is NOT here
29//!
30//! `run_command` still lives on the plugin in [`crate::lib`], and it now
31//! delegates. Nothing git-shaped went into base znippy for any of this.
32
33use std::collections::{BTreeMap, HashMap};
34use std::fs::File;
35use std::path::{Path, PathBuf};
36use std::sync::{Arc, Mutex, RwLock};
37
38use anyhow::{anyhow, bail, Context, Result};
39use znippy_common::ReservedSection;
40
41use crate::arms::StoreConfig;
42use crate::exploded::ExplodedStats;
43use crate::exploded_arrow::ExplodedArchive;
44use crate::gc::{Gc, GcReport};
45use crate::graph::{assign_generations, CommitNode};
46use crate::index_layout::{ObjectIndex, OneTableFourColumns};
47use crate::indexer::{IndexJob, ObjectAbsorb, PushPath};
48use crate::object::{GitHashKind, GitObjectKind};
49use crate::pack_walk::PackWalk;
50use crate::reach::ReachEntry;
51use crate::read_stack::{ObjectReadStack, RebuildTriggers};
52use crate::refs::{RefLog, RefUpdate};
53use crate::resolve::BaseSource;
54
55// ── the contract, now owned by `git-storage-trait` ────────────────────────────
56//
57// `GitOps` (the neutral ELEVEN) and its value types live in the
58// `git-storage-trait` crate at this repo's root, so a gix backend
59// (edda/storage-git-gix) can implement the same contract without linking Arrow
60// or OpenZL. The twelfth method, `seal()`, returns `znippy_common::
61// ReservedSection` (an Arrow type a gix backend has no analog for), so it is an
62// **inherent** method on [`GitStore`] / [`SelectedStore`], not part of the
63// trait. The re-exports below keep every existing path
64// (`znippy_plugin_git::git_ops::{GitOps, Oid, …}`) alive.
65pub use git_storage_trait::{Extent, GitOps, Oid, RefRow, Stored, TxId};
66
67
68// ── the batch shape, which is measured and must not be papered over ───────────
69
70/// Which path a lookup of `n` oids takes.
71///
72/// The number this dispatch was written for is `index_layout_bench`'s: on the bare
73/// Arrow arms, `lookup_batch` with a **single** element costs 818 ns against
74/// `lookup`'s 482 ns — 1.7x worse — because the batch path allocates a result
75/// vector, a key vector and a second pass over them to serve one oid.
76///
77/// **RE-MEASURED against the shipping read stack, and the magnitude does not
78/// carry over.** oden 2026-08-07, release, 1-min loadavg 4.94 (another tenant's
79/// work — see LAW: never measure on a busy box, and read these as a ratio rather
80/// than as absolutes), `ObjectReadStack<OneTableFourColumns>` over a real
81/// repository's 2687-object pack, 20 000 iterations
82/// ([`crate::store::tests::the_batch_of_one_really_is_slower_than_the_serial_path`]):
83///
84/// | call | per oid |
85/// |---|---:|
86/// | `lookup` | **628 ns** |
87/// | `lookup_batch` with 1 oid | **639 ns** — 1.02x worse |
88/// | `lookup_batch` with 100 oids | **161 ns** — 3.9x better |
89///
90/// So: the *direction* holds and the dispatch is right — a batch of one is never
91/// faster — but on this stack at this size it costs 2%, not 70%. The real finding
92/// is the third row: the batch path is worth **3.9x** at
93/// [`BATCH_SATURATES_AT`], which is a much stronger reason to keep the two paths
94/// apart than the batch-of-one penalty ever was. Both numbers are stated because
95/// quoting only the 1.7x would be quoting a figure this stack does not reproduce.
96///
97/// The dispatch is a named function with its own guard precisely so it cannot
98/// quietly become `lookup_batch` for every n.
99#[derive(Debug, Clone, Copy, PartialEq, Eq)]
100pub enum LookupPath {
101 /// One oid: `ObjectIndex::lookup`.
102 Serial,
103 /// Two or more: `ObjectIndex::lookup_batch`, one pass.
104 Batch,
105}
106
107/// Oids at which the batch path stops getting faster. Reported, not enforced:
108/// splitting a bigger batch would cost a pass and buy nothing.
109pub const BATCH_SATURATES_AT: usize = 100;
110
111/// The dispatch, in one place.
112pub fn lookup_path(n: usize) -> LookupPath {
113 if n == 1 {
114 LookupPath::Serial
115 } else {
116 LookupPath::Batch
117 }
118}
119
120// ── the store ─────────────────────────────────────────────────────────────────
121
122/// A pack whose bytes are durable but whose objects the index has not absorbed.
123///
124/// §13.12's `indexed` bit, in the form this store needs it: membership in
125/// [`PackState::pending`] *is* the bit being clear. A read that arrives while it
126/// is non-empty falls back to absorbing it — slower, never wrong.
127#[derive(Debug, Clone, Copy, PartialEq, Eq)]
128struct PendingPack {
129 pack_id: u64,
130 offset: u64,
131 len: u64,
132}
133
134impl PendingPack {
135 fn from_job(j: IndexJob) -> Self {
136 Self {
137 pack_id: j.pack_id,
138 offset: j.offset,
139 len: j.len,
140 }
141 }
142}
143
144/// **§13.12's `indexed` bit — one bit per pack, indexed by pack ordinal.**
145///
146/// A plain bitset, and the two rejected alternatives say why:
147///
148/// * **not a hash set.** Pack ordinals are dense and small: they are assigned
149/// `0..N` within **one repository** (§13.14 scopes a negotiation, and therefore
150/// this store, to one repository — account-wide they would be sparse and the
151/// array would grow with the account's total packs). Dense means the bit is one
152/// shift and one AND with no hashing, and 10 000 packs is **1250 bytes** of
153/// `words`, which lives in L1.
154/// * **not a bloom filter.** A bloom's false positive would be *actively wrong*
155/// here: it would report a pack absorbed when it is not, the read path would
156/// stop falling back, and negotiation would be told `absent` about an object
157/// whose bytes are durable — which makes a client withhold objects and lose
158/// data. The bitset has no such mode. `1` means absorbed, `0` means fall back,
159/// and both are exact.
160///
161/// **The bit is never stored.** It is derived on open by diffing two facts that
162/// are already durable — the journal's extents against the index's rows — in
163/// [`Absorber::adopt_journal`]. There is nothing for a derived bit to drift from.
164///
165/// **Why the extents sit beside it.** A `0` bit is only actionable with the
166/// extent to absorb, so `extent[i]` carries it; `None` means "this store has not
167/// been told about ordinal `i`", which is not the same as "not absorbed" and must
168/// not be counted as work.
169///
170/// **Why `absorbed` is a separate fact from `extent`.** The ack path does two
171/// things in this order: record the pack pending, then hand its extent to the
172/// account's channel. The indexer's worker is a different thread and can reach
173/// `absorb` *before* the first of those returns. With only a pending list, that
174/// race ends in a bit that is set after the rows are already in and is never
175/// cleared, so every later read pays a pointless fallback for the life of the
176/// process. The bit is what the two writers agree on: [`PackState::note`] will
177/// not queue a pack that is already done, and `absorb` will not do the work
178/// twice.
179///
180/// Touched only under short critical sections — never across an absorb, which is
181/// what the separate [`Absorber::gate`] is for. **The ack path must never block
182/// behind index work.**
183#[derive(Default)]
184struct PackState {
185 /// One bit per pack ordinal, 64 to a word. `1` = its rows are in the index.
186 words: Vec<u64>,
187 /// The extent of each known ordinal, so a `0` bit can be acted on.
188 extent: Vec<Option<Extent>>,
189 /// Known ordinals whose bit is `0` — what [`GitStore::unindexed_packs`]
190 /// reports, kept as a count so the read path's check is a field load rather
191 /// than a scan.
192 unabsorbed: usize,
193}
194
195impl PackState {
196 /// The bit.
197 fn is_absorbed(&self, pack_id: u64) -> bool {
198 let i = pack_id as usize;
199 self.words
200 .get(i / 64)
201 .is_some_and(|w| w >> (i % 64) & 1 == 1)
202 }
203
204 /// Record a durable pack whose rows are not in yet. The bit **going clear**.
205 ///
206 /// **Idempotent, and the extent slot is the single thing that makes it so.**
207 /// A second `note` for an ordinal that already has one — the ack path
208 /// arriving after the drain has already absorbed the pack, or a re-queue on
209 /// open racing the same pack's push — finds the slot filled and counts no new
210 /// work. [`PackState::mark_absorbed`] fills that same slot, so a pack that is
211 /// already done cannot be queued behind its own completion, and there is no
212 /// second condition here that could disagree with the first.
213 fn note(&mut self, pack_id: u64, extent: Extent) {
214 let i = pack_id as usize;
215 self.grow_to(i);
216 if self.extent[i].is_none() {
217 self.extent[i] = Some(extent);
218 self.unabsorbed += 1;
219 }
220 }
221
222 /// Room for ordinal `i` in both arrays. Dense ordinals are what makes this a
223 /// `resize` and not an insert.
224 fn grow_to(&mut self, i: usize) {
225 if self.words.len() <= i / 64 {
226 self.words.resize(i / 64 + 1, 0);
227 }
228 if self.extent.len() <= i {
229 self.extent.resize(i + 1, None);
230 }
231 }
232
233 /// The bit **going up**: this pack's rows are in the index.
234 fn mark_absorbed(&mut self, pack_id: u64, extent: Extent) {
235 let i = pack_id as usize;
236 self.grow_to(i);
237 if self.words[i / 64] >> (i % 64) & 1 == 0 {
238 self.words[i / 64] |= 1 << (i % 64);
239 if self.extent[i].is_some() {
240 self.unabsorbed -= 1;
241 }
242 }
243 self.extent[i] = Some(extent);
244 }
245
246 /// Every known pack whose bit is clear, in ordinal order.
247 fn pending(&self) -> Vec<PendingPack> {
248 self.extent
249 .iter()
250 .enumerate()
251 .filter(|(i, e)| e.is_some() && !self.is_absorbed(*i as u64))
252 .map(|(i, e)| {
253 let (offset, len) = e.expect("filtered on Some");
254 PendingPack {
255 pack_id: i as u64,
256 offset,
257 len,
258 }
259 })
260 .collect()
261 }
262}
263
264/// The derived tables: the commit graph, the tree payloads the reachability
265/// bitmaps are built from, and this store's own ordinal space.
266///
267/// **The ordinal space is this struct's, not the projection's.** An
268/// [`crate::index_layout::IndexRow::ordinal`] is a row address inside one
269/// projection generation and is re-derived by every rebuild, so a bitmap built
270/// over projection ordinals would silently address different objects after any
271/// rebuild. The bitmaps here are over `ordinal`, which is rebuilt in the same
272/// fold that rebuilds them — the two cannot drift because one call produces both.
273/// **Nothing in here is a source of truth.** Every field is folded from the
274/// `objects` table and §14's exploded table by [`Absorber::refold`], and both of
275/// those are on disk — which is what makes a clean reopen come back with the
276/// same graph it shut down with. Before the exploded table existed, `graph` and
277/// `trees` were *accumulated* here as packs were absorbed and stored nowhere, so
278/// a clean shutdown lost them silently: MEASURED on oden 2026-08-08, a reopened
279/// store over a fully absorbed 2687-object pack had all 2687 rows and **0 of 551
280/// commits**, with `reachable()` returning a commit alone instead of its closure
281/// and `gc` seeing an empty live set.
282#[derive(Default)]
283struct Derived {
284 /// Parents-before-children, generations assigned. Folded from the exploded
285 /// table's commit payloads.
286 graph: Vec<CommitNode>,
287 /// Tree payloads by oid hex — what [`crate::reach::ObjectFacts`] wants.
288 /// Folded from the exploded table's tree payloads.
289 trees: HashMap<String, Vec<u8>>,
290 /// oid hex → ordinal, and its inverse.
291 ordinal: HashMap<String, u32>,
292 oids: Vec<String>,
293 /// The same inverse in **raw bytes**: ordinal `o` is
294 /// `oids_raw[o * oid_len .. (o + 1) * oid_len]`, one flat allocation for the
295 /// whole store.
296 ///
297 /// # Why both, rather than deriving one from the other per request
298 ///
299 /// Because deriving it per request is what this field replaces, and it was
300 /// the most expensive thing on a fetch that no stage was named after.
301 /// `reachable_oids` answered in oid **hex** and every serving caller
302 /// immediately `hex::decode`d it straight back to the bytes it came from:
303 /// one `String` plus one `Vec<u8>` per object in the answer, and the answer
304 /// for the `have` side is the closure of what the client already holds —
305 /// very nearly the whole repository. MEASURED on oden 2026-08-15, the
306 /// `nornir` mirror (12 303 objects), a 69-object negotiated fetch: **12 112
307 /// oids** through that round trip, **twice** per request (`select`'s voucher
308 /// and `emit_set`'s thin bases), plus a third `to_vec` per object to key
309 /// them by offset.
310 ///
311 /// The hex form stays because the ordinal map, the commit graph and the
312 /// tree payloads are keyed by it and those are not this change's subject;
313 /// the raw form is what every *serving* answer is now built out of. Both are
314 /// folded from one sorted vector of raw oids, so they cannot fall out of
315 /// step: see [`Absorber::refold`].
316 oids_raw: Vec<u8>,
317 /// The bitmaps, or empty if nothing has needed them since the last fold.
318 ///
319 /// **Behind an `Arc` since 2026-08-14, and that is a measurement and not a
320 /// tidy-up.** [`Absorber::reach_bitmaps`] used to hand out `d.reach.clone()`
321 /// — a deep copy of every commit's `RoaringBitmap` plus its oid `String` —
322 /// and `crate::serve::GitServe::select` calls
323 /// [`GitStore::reachable_oids`] **twice** per request (once for `want`, once
324 /// to close over `have`). MEASURED on oden with `stage-probe`, the
325 /// znippy-served `nornir` mirror (12 455 objects), a 4-object negotiated
326 /// fetch: the `select.bitmap` stage was **75.2 % of the request's CPU** and
327 /// **645 876 allocations**; at four requests sharing one warm store it was
328 /// still 62 762 allocations *per request*. None of that copying is work the
329 /// answer depends on — the table is immutable between folds — so the `Arc`
330 /// removes it rather than dividing it across cores.
331 reach: Arc<Vec<ReachEntry>>,
332 /// Every commit's oid as **raw bytes**, the form
333 /// [`crate::serve::GitServe::select`] tests a `want` against.
334 ///
335 /// `None` means a graph row's oid did not parse as hex, which `select`
336 /// answers by declining the whole request. That check is why this is
337 /// `Option` rather than a plain set: it used to run **per request** —
338 /// `graph_snapshot()` cloned the whole `Vec<CommitNode>` and then
339 /// `hex::decode`d every commit into a fresh `Vec<u8>` — and the answer
340 /// cannot change between folds, so it is folded once here and borrowed
341 /// after. The refusal is preserved exactly; only the arithmetic moved.
342 commit_raw: Option<Arc<std::collections::HashSet<Vec<u8>>>>,
343 /// Verbatim packs, so any absorbed object's content can be re-derived from
344 /// the truth on demand. `(archive offset, len)`.
345 packs: Vec<(u64, u64)>,
346}
347
348/// **The object-level index, and the one thing that writes into it.**
349///
350/// Its own allocation rather than a field group inside [`GitStore`], and §13.9-12
351/// is the only reason: the drain runs on the account indexer's worker thread, so
352/// whatever it absorbs into cannot be reachable solely through a `&GitStore`.
353/// Splitting it out is what lets the background worker and a falling-back read
354/// call **the same** absorb instead of two copies of it (LAW 5) — the alternative
355/// was a second implementation on the worker side, which is precisely the
356/// twinning that law forbids.
357///
358/// It is a **leaf**: nothing in here points back at the store, so the worker's
359/// `Arc` closes no cycle and the store still drops (and joins its worker)
360/// normally.
361/// Entries below which [`GitStore::resolve_emit_payloads`] does **not** fan out.
362///
363/// A `std::thread::scope` spawn is ~10–20 µs a worker and the per-entry work it
364/// would divide is a bounds check plus a varint; a 100-object fetch would pay
365/// four spawns to save four microseconds of parsing. The threshold is what keeps
366/// `crate::serve`'s "one core's worth per transfer" true for every request that
367/// is not a bulk clone — and a bulk clone is the only request the copy this
368/// replaces was ever measured on.
369const FAN_OUT_AT: usize = 4096;
370
371/// Commits that get a reachability bitmap on the **live** selection path.
372///
373/// # It was `usize::MAX`, and that was the single most expensive line in a fetch
374///
375/// The comment that stood here said a sampled table *"is for the sealed archive
376/// where the cost is disk"* and that every commit must be bitmapped on the live
377/// path. That was not a preference, it was forced:
378/// [`GitStore::reachable_oids`] had no walk, so a `want` with no bitmap could
379/// only contribute itself — a silent under-send — and `crate::serve`'s `select`
380/// therefore had to refuse unless *"in the graph"* and *"has a bitmap"* were the
381/// same fact.
382///
383/// MEASURED on oden 2026-08-14, one 4-object negotiated fetch out of the
384/// znippy-served `nornir` mirror (1 581 commits, 12 455 objects): the
385/// `select.bitmap` stage was **75 % of the request's CPU**, and **530 218 of its
386/// 575 349 allocations were this build** — which `refold` throws away again on
387/// the next push. The gix arm answered the identical fetch in 6.8 ms because it
388/// *reads* a `.bitmap` instead of building one.
389///
390/// [`crate::reach::accumulate`] is the walk that makes the cap legal, and 512 is
391/// the number [`crate::reach::ReachPolicy::default`] already uses for the sealed
392/// archive — the same figure git's own bitmap selection is of the order of, and
393/// deliberately not a second, different constant to keep in step.
394const LIVE_REACH_COMMITS: usize = 512;
395
396/// [`LIVE_REACH_COMMITS`], or `ZNIPPY_GIT_REACH_COMMITS`.
397///
398/// **An operator knob, not an arm.** It selects no implementation and changes no
399/// byte of any answer: the walk covers whatever the table does not, so the cap
400/// trades build cost against walk cost and nothing else. That property is the
401/// whole subject of
402/// `the_bounded_walk_answers_exactly_what_a_full_bitmap_table_answers`, which
403/// drives this very knob to its two extremes over one store.
404///
405/// Read **once per process** through [`crate::arms::read_env`], so it is
406/// counted with every other environment read this crate makes, and so a
407/// reachability request does not cost a `getenv` — until 2026-08-21 it did,
408/// once per `reach_bitmaps`/`reachable_oids` call, uncounted.
409fn live_reach_policy() -> crate::reach::ReachPolicy {
410 static MAX: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
411 let max_commits = *MAX.get_or_init(|| {
412 crate::arms::read_env(crate::arms::ENV_REACH_COMMITS)
413 .and_then(|v| v.parse::<usize>().ok())
414 .unwrap_or(LIVE_REACH_COMMITS)
415 });
416 crate::reach::ReachPolicy { max_commits }
417}
418
419/// Ceiling on phase-1 workers **per request**.
420///
421/// The serving tier admits ~32 concurrent transfers, and the fan-out primitive
422/// has no reentrancy detection, so an uncapped `gatling_for_each` here is
423/// 32 × ncores threads on a 128-core box. Four is a bound that leaves the
424/// admission gate in charge of the machine.
425const MAX_RESOLVE_WORKERS: usize = 4;
426
427/// Workers for phase 1: [`MAX_RESOLVE_WORKERS`], or `ZNIPPY_GIT_EMIT_WORKERS`.
428///
429/// **An operator knob, not an arm.** It selects no implementation and changes no
430/// byte of the emitted pack — a pack emitted at one worker and at sixteen is
431/// byte-identical, because phase 1 only decides *where each entry's bytes are*
432/// and phase 2 writes them in one serial order either way. `0` is read as "as
433/// many as this box has cores", which is `gatling_for_each`'s own convention.
434///
435/// Read **once per process** through [`crate::arms::read_env`] — see
436/// [`live_reach_policy`] for why; this one was a `getenv` per emitted request
437/// until 2026-08-21.
438fn emit_workers() -> usize {
439 static WORKERS: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
440 *WORKERS.get_or_init(|| {
441 crate::arms::read_env(crate::arms::ENV_EMIT_WORKERS)
442 .and_then(|v| v.parse::<usize>().ok())
443 .unwrap_or(MAX_RESOLVE_WORKERS)
444 })
445}
446
447pub(crate) struct Absorber<S: ObjectIndex = OneTableFourColumns> {
448 blobs: PathBuf,
449 /// Read handle on the verbatim packs, for `pread` of an extent.
450 ///
451 /// The **fallback**, since 2026-08-14, not the ordinary path: `map` below
452 /// answers every extent the mapping covers, and this reads the ones past its
453 /// end (a push that landed after the snapshot was taken).
454 reader: File,
455 /// The same file, mapped read-only. See [`crate::archive_map`] for why an
456 /// append-only file `gc` never rewrites is safe to map, and for the ~20 % of
457 /// serve cycles the `pread`-per-object shape was spending on copies.
458 map: crate::archive_map::ArchiveMap,
459 hash: GitHashKind,
460 /// The `objects` table: stree → Arrow → redb tail.
461 ///
462 /// `S` is the payload layout the projection is built in
463 /// ([`crate::arms::IndexArm`]) and it is a **type parameter, not a field**:
464 /// this is the hot path, every lookup goes through it, and the whole reason
465 /// the arms exist is to measure the differences between them. A `dyn` here
466 /// would put a virtual call on every single oid.
467 objects: ObjectReadStack<S>,
468 /// **§14's exploded objects table** — oid → resolved type and content, built
469 /// eagerly by this absorb and by nothing else.
470 ///
471 /// Beside the `objects` table rather than inside it because the two have
472 /// different lifetimes: `objects` is the index and losing it loses the
473 /// repository's addressing, while this one is derived, verifiable against the
474 /// verbatim bytes and deletable at any time. They are separate files so that
475 /// "droppable" is a `rm`, not a schema migration.
476 ///
477 /// SWAPPED 2026-08-13 from the redb `ExplodedTable` to **one Arrow IPC
478 /// table**, payload in the row. redb held 16.8 GB of kernel payload in a
479 /// 204 GB file — 12× — and serialised the gatling fan-out behind its one
480 /// writer at 96.8% of a single core. See [`crate::exploded_arrow`].
481 exploded: ExplodedArchive,
482 derived: RwLock<Derived>,
483 packs: Mutex<PackState>,
484 /// Serialises **absorption**, which is long. Held across a whole pack.
485 ///
486 /// Two absorbs of one pack are duplicated *work* and nothing worse: every
487 /// writer under it is keyed by oid — [`ObjectReadStack::append`] swallows an
488 /// identical second row, the exploded table overwrites its own key, and the
489 /// commit graph is folded from that table rather than pushed at, so it cannot
490 /// double. (It could before §14's table existed, when the graph was a `Vec`
491 /// accumulated per absorb; that is why this paragraph used to say the gate
492 /// was load-bearing for correctness and now does not.) The gate is still what
493 /// makes the ordering below hold: **the rows go in, and only then is the bit
494 /// cleared**, so a read either finds the bit still clear (and waits here) or
495 /// finds the rows. Neither order can produce "absent".
496 gate: Mutex<()>,
497}
498
499/// One repository's git store.
500///
501/// | on disk | what it is | who owns the format |
502/// |---|---|---|
503/// | `<root>/objects.pack` | the verbatim pack bytes, appended | [`SafeWriter`] |
504/// | `<root>/objects.pack.journal` | the durable extent rows | [`SafeWriter`] |
505/// | `<root>/objects.tail` | the un-sealed index tail | [`ObjectReadStack`] (redb) |
506/// | `<root>/objects.exploded` | §14's resolved content, **droppable** | [`ExplodedArchive`] (one Arrow IPC table) |
507/// | `<root>/refs.log` | the ref transaction log | [`RefLog`] (Arrow IPC frames) |
508/// | `<root>/<name>.znippy` | the sealed archive GC compacts | base znippy |
509///
510/// Nothing in that table is new. The store is the composition, not a fifth
511/// format.
512/// # Which implementation of each trait, and how a caller picks
513///
514/// Three of the four boxes in the diagram at the top of this file have more than
515/// one implementation. [`StoreConfig`] is the choice, [`crate::arms`] documents
516/// what each arm promises, and the dispatch is split deliberately:
517///
518/// | trait | selected by | dispatch | cost |
519/// |---|---|---|---|
520/// | [`ObjectIndex`] | the type parameter `S` | **static** | none — the hot path is monomorphised |
521/// | [`ArchiveWrite`](crate::archive_write::ArchiveWrite) | [`StoreConfig::writer`] | `dyn`, inside [`PushPath`] as it already was | one virtual call **per pack append**, not per object |
522/// | [`Gc`] | [`StoreConfig::gc`] | `dyn` | one virtual call per `gc()` run |
523///
524/// `S` defaults to [`OneTableFourColumns`], so `GitStore` still names exactly
525/// what it named before and every existing caller compiles unchanged.
526pub struct GitStore<S: ObjectIndex = OneTableFourColumns> {
527 root: PathBuf,
528 /// The verbatim pack bytes. The **truth** (§14).
529 blobs: PathBuf,
530 /// The sealed znippy archive `gc` compacts. Only GC and `seal` touch it.
531 archive: PathBuf,
532 /// The selected [`ArchiveWrite`](crate::archive_write::ArchiveWrite) arm
533 /// plus the per-account index channel. The ack path.
534 push: PushPath,
535 account: String,
536 hash: GitHashKind,
537 /// The object-level index and the absorb behind it — **shared with the
538 /// account indexer's worker**, which is how a push ends in object rows
539 /// without a read having to ask for them.
540 absorber: Arc<Absorber<S>>,
541 refs: RefLog,
542 /// Serialises the ref log's read-compare-append. A CAS that read the
543 /// namespace, then appended, without holding this could be overtaken between
544 /// the two and would overwrite the update it did not see.
545 ref_gate: Mutex<()>,
546 /// The selected [`Gc`] arm. Built once, here, and run by
547 /// [`GitOps::gc`] — a `Box<dyn>` because a GC is a whole-archive
548 /// compaction that happens on a maintenance timer, so one virtual call
549 /// against it is unmeasurable and a third type parameter on every signature
550 /// in the crate is not free to read.
551 gc: Box<dyn Gc + Send + Sync>,
552 /// What this store was built with, for a bench row or a config dump. A
553 /// description of the arms already constructed — **nothing consults it on a
554 /// serving path**.
555 arms: StoreConfig,
556}
557
558impl GitStore<OneTableFourColumns> {
559 /// Open (creating) the store for one repository under `root`.
560 ///
561 /// **The default arms, and it reads no environment to get them.** An
562 /// operator who exported `ZNIPPY_GIT_WRITER=fast` does not change what this
563 /// builds — a caller that never asked for a selector must not have its
564 /// durability contract moved out from under it. Use
565 /// [`open_from_env`] when the selection is wanted.
566 ///
567 /// ⚠ One variable *is* read here, and it is not an arm:
568 /// `ZNIPPY_GIT_REDB_CACHE_BYTES` (see
569 /// [`crate::arms::redb_cache_bytes`]). It selects no implementation,
570 /// changes no durability contract and leaves the bytes on disk identical —
571 /// it caps redb's page cache, whose own default is 1 GiB *per database*, of
572 /// which this store opens two. The sentence above is about arms, and a
573 /// memory ceiling is not one.
574 pub fn open(root: &Path, account: &str) -> Result<Self> {
575 Self::open_with(root, account, GitHashKind::Sha1)
576 }
577
578 /// Same, with an explicit oid width. sha256 repositories exist and the store
579 /// does not get to assume otherwise; every oid it stores is
580 /// [`GitHashKind::oid_len`] bytes wide and the tail refuses a mixed batch.
581 ///
582 /// # What a reopen recovers, and why nothing had to be stored for it
583 ///
584 /// A store opened over an `objects.pack` some earlier process was still
585 /// indexing has to end up in the same state that process would have reached.
586 /// §13.12's `indexed` bit is what says so, and it is **derived here rather
587 /// than persisted**: a pack is unabsorbed *iff* its extent is in the journal
588 /// and its rows are not in the index, and both of those are already durable
589 /// (see [`Absorber::adopt_journal`]). The packs that come back clear are put
590 /// straight onto the account's channel, so the drain finishes the interrupted
591 /// work with no read having to ask — and until it does, every read falls back
592 /// exactly as it would have before the crash. **Slower, never wrong**, across
593 /// a restart as well as within one.
594 ///
595 /// The bits are set **before** the channel is fed, which is the same ordering
596 /// the ack path uses for the same reason: a read that arrives in between finds
597 /// the bit clear and waits, never an index that does not mention the pack.
598 ///
599 /// # What a reopen recovers of the DERIVED tables, and how that was fixed
600 ///
601 /// [`Derived::graph`] and [`Derived::trees`] come back **in full**, for every
602 /// pack, whether or not anything is re-queued. They are folded from §14's
603 /// exploded table ([`Absorber::refold`]), which is on disk, so a clean
604 /// shutdown carries them across.
605 ///
606 /// This was a live correctness hole until 2026-08-08 and it is recorded
607 /// rather than quietly repaired. The graph was *accumulated in RAM* as packs
608 /// were absorbed and stored nowhere; a store killed mid-absorb got it back
609 /// because the interrupted pack was re-queued and re-resolved, but a store
610 /// that shut down **cleanly** re-queued nothing and came back with an empty
611 /// graph. MEASURED on oden 2026-08-08: 2687 rows and **0 of 551 commits**,
612 /// `reachable` on a commit returning that commit alone instead of its
613 /// closure — a *quiet* wrong answer — and `gc` seeing an empty live set. The
614 /// fix is the table, because the payloads the graph is folded from had no
615 /// home; [`tests::a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable`]
616 /// is what proves it, over a real process that exits normally.
617 ///
618 /// The exploded table is still **droppable**: if its row count is short of
619 /// the `objects` table's, [`Absorber::adopt_journal`] declares no pack
620 /// absorbed and every one of them is re-queued and re-exploded. Absent means
621 /// rebuild, never wrong (§13.12, applied unchanged to a derived table).
622 ///
623 /// **Also the default arms, and also no environment read** — see
624 /// [`open`](GitStore::open).
625 pub fn open_with(root: &Path, account: &str, hash: GitHashKind) -> Result<Self> {
626 // The default ARMS, and the environment's CEILING: the one variable
627 // this constructor reads, because an operator capping a server's
628 // footprint must not have to pick a different constructor to do it.
629 // See [`crate::arms::redb_cache_bytes`].
630 Self::open_with_arms(
631 root,
632 account,
633 hash,
634 StoreConfig::DEFAULT.with_redb_cache_bytes(crate::arms::redb_cache_bytes()?),
635 )
636 }
637}
638
639/// Where a stored entry's delta base sits, relative to the request being served.
640///
641/// The `Outside` arm carries the base in **both** namings because the two delta
642/// forms know only one each — an `OFS_DELTA` its archive offset, a `REF_DELTA`
643/// its oid — and [`GitStore::emit_set`]'s thin arm has to ask "does the client
644/// hold this" in whichever one the entry speaks.
645enum BaseOf {
646 /// Not a delta at all. There is nothing to place it after and nothing to
647 /// re-head; the stored entry is copied as it stands.
648 Whole,
649 /// A delta whose base **is** in the request, at this archive offset.
650 ///
651 /// The offset is carried rather than recomputed because a `REF_DELTA` does
652 /// not know it — it names its base by oid, and its own `delta_base` column
653 /// is `0` by construction — and the emitter needs it twice: as the
654 /// `EmitEntry::delta_base` that [`crate::pack_walk::topological_order`]
655 /// orders on, and as the base an in-pack ref-delta is re-headed to point at.
656 Inside { at: u64 },
657 /// A delta whose base the request does not carry.
658 Outside {
659 /// The base's archive offset, `0` when the entry named an oid instead.
660 offset: u64,
661 /// The base's oid, `None` when the entry named an offset instead — or
662 /// when the entry is too short to read one out of.
663 oid: Option<Vec<u8>>,
664 },
665}
666
667/// The bases a **thin** pack may name without carrying them: what the receiver
668/// already held before this transfer, keyed both ways.
669///
670/// Built by [`GitStore::client_bases`], which is where the argument for trusting
671/// it is written down.
672struct ExternalBases {
673 /// The held object's oid, keyed by its archive offset — the coordinate an
674 /// `OFS_DELTA` names its base in, and the only reason this map exists.
675 by_offset: HashMap<u64, OidKey>,
676 /// The same objects keyed the way a `REF_DELTA` names them.
677 held: std::collections::HashSet<OidKey>,
678}
679
680/// **A raw oid as a fixed-size, `Copy` hash key — 20 bytes for SHA-1, 32 for
681/// SHA-256, and no heap allocation either way.**
682///
683/// The two maps in [`ExternalBases`] are repository-sized: the client's `have`
684/// closure on an incremental fetch is very nearly every object in the store.
685/// Keying them by `Vec<u8>` is one heap allocation per entry per request —
686/// 12 112 of them for a 69-object fetch, measured on oden 2026-08-15 — for a
687/// question (*does the client hold this base?*) that never looks at more than
688/// `oid_len` bytes.
689///
690/// Padded to the widest hash rather than parameterised by width: a store holds
691/// exactly one hash kind, so every key in one set has the same real length and
692/// two different oids can never collide through the padding. The length is
693/// carried anyway, because a key that silently compared equal across widths is
694/// the kind of wrong that only shows up in a mixed-hash mirror.
695#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
696struct OidKey {
697 bytes: [u8; 32],
698 len: u8,
699}
700
701impl OidKey {
702 /// The key for `oid`, or a refusal for one wider than any git hash.
703 fn new(oid: &[u8]) -> Result<Self> {
704 let len = oid.len();
705 if len == 0 || len > 32 {
706 bail!("{len} is not the width of any git object id (1..=32 bytes)");
707 }
708 let mut bytes = [0u8; 32];
709 bytes[..len].copy_from_slice(oid);
710 Ok(OidKey {
711 bytes,
712 len: len as u8,
713 })
714 }
715
716 /// The oid back, at its real width.
717 fn as_slice(&self) -> &[u8] {
718 &self.bytes[..self.len as usize]
719 }
720}
721
722impl<S: ObjectIndex + 'static> GitStore<S> {
723 /// **The one constructor**, with every arm named.
724 ///
725 /// `S` picks the [`ObjectIndex`] layout at compile time; `arms` picks the
726 /// [`ArchiveWrite`](crate::archive_write::ArchiveWrite) and [`Gc`]
727 /// implementations at run time. [`open`](GitStore::open) and
728 /// [`open_with`](GitStore::open_with) are this function with
729 /// [`StoreConfig::DEFAULT`], which is why nothing about them moved.
730 ///
731 /// `arms.index` is **not** consulted here — `S` is the index arm, and a
732 /// value that disagreed with the type would be a second source of truth.
733 /// [`open_from_env`] is the one place the two are tied together, and it does
734 /// it by choosing `S` from the value.
735 ///
736 /// ```no_run
737 /// use znippy_plugin_git::arms::{GcArm, StoreConfig, WriterArm};
738 /// use znippy_plugin_git::index_layout::PackedPayload;
739 /// use znippy_plugin_git::{GitHashKind, GitStore};
740 ///
741 /// // A rebuildable import mirror: the ceiling writer, the packed payload
742 /// // layout, and an in-place compaction.
743 /// let arms = StoreConfig::DEFAULT
744 /// .with_writer(WriterArm::Fast)
745 /// .with_gc(GcArm::CompactInPlace);
746 /// let store = GitStore::<PackedPayload>::open_with_arms(
747 /// std::path::Path::new("/srv/repo"),
748 /// "rickard",
749 /// GitHashKind::Sha1,
750 /// arms,
751 /// )?;
752 /// # Ok::<(), anyhow::Error>(())
753 /// ```
754 pub fn open_with_arms(
755 root: &Path,
756 account: &str,
757 hash: GitHashKind,
758 arms: StoreConfig,
759 ) -> Result<Self> {
760 std::fs::create_dir_all(root)
761 .with_context(|| format!("creating the store root {}", root.display()))?;
762 let blobs = root.join("objects.pack");
763 let writer = arms.writer.create(&blobs)?;
764 // `None` for an arm that keeps no durable ack log — see
765 // [`crate::arms::WriterArm::journal`]. Everything below that touches the
766 // journal is written against that `Option` rather than assuming one.
767 let journal = arms.writer.journal(&blobs);
768 // **One ceiling for both databases**, carried in on the config rather
769 // than read here: `StoreConfig::from_env` reads it with the arms, and
770 // `open`/`open_with` put the environment's value on `DEFAULT` before
771 // getting here, so every constructor still honours the variable and
772 // `arms.to_string()` prints what was actually applied. Not an arm — see
773 // [`crate::arms::redb_cache_bytes`] — redb's own default is 1 GiB *per
774 // database*.
775 let cache_bytes = arms.redb_cache_bytes;
776 let objects = ObjectReadStack::<S>::open(
777 &root.join("objects.tail"),
778 RebuildTriggers::default(),
779 cache_bytes,
780 )?;
781 let _ = cache_bytes; // no page cache to size: the blob is pread, the index is a slice
782 let exploded = ExplodedArchive::open(&root.join("objects.exploded"))?;
783 let absorber = Arc::new(Absorber::<S>::open(&blobs, hash, objects, exploded)?);
784
785 // **The `indexed` bit, derived.** Every extent this archive has ever
786 // acked, diffed against the rows the index holds. `SafeWriter::create`
787 // above appends to that journal and never truncates it, which is what
788 // makes the older extents still be here to diff against.
789 //
790 // An arm with no journal has nothing to derive the bit from, and that is
791 // the honest consequence of choosing it rather than a gap: a
792 // `FastWriter` store has no durable record that a pack was acked, so a
793 // reopen re-queues nothing and there is nothing for it to re-queue.
794 let acked = match journal.as_deref() {
795 Some(p) if p.exists() => crate::archive_write::read_journal(p).with_context(|| {
796 format!(
797 "reading {} — it is the durable half of the indexed bit and a store \
798 cannot be opened without knowing which packs it owes index work for",
799 p.display()
800 )
801 })?,
802 _ => Vec::new(),
803 };
804 let requeue = absorber.adopt_journal(&acked)?;
805
806 // **The wiring.** The absorber goes into the push path, which hands it to
807 // every account indexer it starts, so the drain behind `push_pack` ends
808 // in object rows instead of one pack row. Nothing about the ack changes:
809 // `append` still returns after the two fsyncs and this is all downstream
810 // of that (§13.9).
811 let push = PushPath::with_absorber(
812 writer,
813 &blobs,
814 journal,
815 absorber.clone() as Arc<dyn ObjectAbsorb>,
816 )?;
817 let refs = RefLog::new(root.join("refs.log"));
818 let archive = root.join("repository.znippy");
819
820 let store = Self {
821 root: root.to_path_buf(),
822 blobs,
823 archive,
824 push,
825 account: account.to_string(),
826 hash,
827 absorber,
828 refs,
829 ref_gate: Mutex::new(()),
830 gc: arms.gc.create(),
831 arms,
832 };
833 // The interrupted work, back on the channel it fell off. Nothing reads
834 // here and nothing blocks: this is the same 24-byte handoff a push makes,
835 // and the drain picks it up on its own thread (§13.9-11).
836 let indexer = store.push.indexer(&store.account);
837 for p in &requeue {
838 indexer.submit(IndexJob {
839 pack_id: p.pack_id,
840 offset: p.offset,
841 len: p.len,
842 })?;
843 }
844 // A store reopened over an existing tail already knows its objects; the
845 // derived tables are rebuilt from them rather than from a side-car,
846 // because a side-car is a second truth.
847 store.refold()?;
848 Ok(store)
849 }
850
851 /// The arms this store was built with. A description of what was
852 /// constructed, not a switch — nothing on a serving path reads it.
853 pub fn arms(&self) -> StoreConfig {
854 self.arms
855 }
856
857 /// The selected writer's name, as it goes on a bench row.
858 pub fn writer_name(&self) -> &'static str {
859 self.push.name()
860 }
861
862 /// What the selected writer's `append` actually promises. Printed next to a
863 /// throughput, because a throughput without it is not a comparison.
864 pub fn writer_durability(&self) -> &'static str {
865 self.push.durability()
866 }
867
868 /// The selected [`Gc`] arm.
869 pub(crate) fn gc_arm(&self) -> &dyn Gc {
870 self.gc.as_ref()
871 }
872
873 /// Where the verbatim packs are.
874 pub fn blobs_path(&self) -> &Path {
875 &self.blobs
876 }
877
878 /// The sealed archive GC operates on.
879 pub fn archive_path(&self) -> &Path {
880 &self.archive
881 }
882
883 pub fn root(&self) -> &Path {
884 &self.root
885 }
886
887 pub fn hash_kind(&self) -> GitHashKind {
888 self.hash
889 }
890
891 /// The `objects` table, for a caller that wants the index directly (the
892 /// bench does).
893 pub fn index(&self) -> &ObjectReadStack<S> {
894 &self.absorber.objects
895 }
896
897 /// Objects in the `objects` table.
898 pub fn object_count(&self) -> usize {
899 self.absorber.objects.len()
900 }
901
902 /// Commits in the graph.
903 pub fn commit_count(&self) -> usize {
904 self.absorber
905 .derived
906 .read()
907 .expect("derived lock")
908 .graph
909 .len()
910 }
911
912 /// **One object's resolved content** — §14's exploded table, in one lookup.
913 ///
914 /// Not one of the twelve and deliberately not on [`GitOps`]: the twelve are
915 /// the wire vocabulary and they hand back **stored** bytes ([`GitOps::get`]),
916 /// which for a delta entry is a delta. This is the other question — *what
917 /// does this object contain* — and it is the one §14's table exists to make
918 /// a point lookup instead of a chain walk.
919 ///
920 /// It absorbs any pending pack first, for the same reason
921 /// [`GitOps::has`] does: a store with durable bytes it has not indexed cannot
922 /// answer "absent" without lying.
923 pub fn content(&self, oid: Oid<'_>) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
924 if self.unindexed_packs() > 0 {
925 self.absorb_pending()?;
926 }
927 self.absorber.resolved(oid)
928 }
929
930 /// §14's table, counted: rows, rows written, reads it served, reads that fell
931 /// through to re-deriving from the verbatim truth.
932 ///
933 /// Applied output. Both content paths return identical bytes, so `served`
934 /// against `rederived` is the only thing that can say which one ran.
935 pub fn exploded_stats(&self) -> ExplodedStats {
936 self.absorber.exploded.engine_stats()
937 }
938
939 /// Where §14's table lives. Deleting this file while no store holds it is the
940 /// sanctioned way to drop it; the next open re-explodes every pack.
941 pub fn exploded_path(&self) -> &Path {
942 self.absorber.exploded.path()
943 }
944
945 /// Every live row of one kind, with its payload — what the graph fold reads,
946 /// exposed so a measurement can ask **the table** how many bytes it holds
947 /// rather than dividing a file size by what the pack claimed.
948 pub fn exploded_of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
949 self.absorber.exploded.of_kind(kind)
950 }
951
952 /// The commit graph as it stands, generations included. A snapshot: the
953 /// caller holds no lock and the next fold replaces it wholesale.
954 pub fn graph_snapshot(&self) -> Vec<CommitNode> {
955 self.absorber
956 .derived
957 .read()
958 .expect("derived lock")
959 .graph
960 .clone()
961 }
962
963 /// Packs whose objects are not in the index yet — §13.12's bit, counted.
964 pub fn unindexed_packs(&self) -> usize {
965 self.absorber.unindexed_packs()
966 }
967
968 /// This account's indexer: the thread that drains the channel a push queues
969 /// on. `is_indexed(pack_id)` on it is §13.12's bit at the pack level, and it
970 /// is set only once that pack's **objects** are in the index.
971 pub fn indexer(&self) -> std::sync::Arc<crate::indexer::AccountIndexer> {
972 self.push.indexer(&self.account)
973 }
974
975 /// Block until the background drain has absorbed everything this store has
976 /// pushed. What a maintenance tick, a seal or a test uses instead of racing
977 /// it.
978 pub fn wait_indexed(&self) {
979 self.push.pool().wait_caught_up();
980 }
981
982 /// **The indexer's half of the split**, and not one of the twelve.
983 ///
984 /// Runs on the account indexer's worker (§13.9), and on a read that arrives
985 /// before that worker got there — the same function either way, which is why
986 /// there is no such thing as a background index that behaves differently from
987 /// the fallback one.
988 pub fn absorb_pending(&self) -> Result<usize> {
989 self.absorber.absorb_pending()
990 }
991
992 /// **The object ids a stored pack introduced**, and not one of the twelve.
993 ///
994 /// [`GitOps::put_pack`] cannot answer this and never will: the walk it does
995 /// reads pack *entries*, and a pack entry does not carry its oid — that is
996 /// why [`crate::pack_walk::Closure`] can only name the `REF_DELTA` bases and
997 /// says so. The oids exist once the indexer has absorbed the pack, so this
998 /// absorbs first rather than racing the drain, and is therefore the one
999 /// caller that deliberately pays §13.9's deferral back.
1000 ///
1001 /// **Only call it when something needs the set.** A push whose policy has no
1002 /// gate over objects must not pay this, which is the whole reason it is a
1003 /// separate call and not a field on `TxId`.
1004 ///
1005 /// The extent *is* the identity: §14 stores a pack verbatim as one
1006 /// contiguous append, so "introduced by this pack" and "indexed at an offset
1007 /// inside this pack's extent" are the same set, with no pack column to keep
1008 /// in step. Bases the push deltas against live in *other* extents and are
1009 /// excluded by construction — which is what a thin push needs, and what
1010 /// reading a written `.idx` back could never do.
1011 ///
1012 /// Cost is one pass over the index, so it is O(objects in the repository),
1013 /// not O(objects pushed). Recorded rather than hidden: a per-pack ordinal
1014 /// range would make it O(pushed), and needs the indexer to keep one.
1015 pub fn oids_in_extent(&self, extent: Extent) -> Result<Vec<Vec<u8>>> {
1016 self.absorb_pending()
1017 .context("absorbing the pending index work before enumerating a pack's objects")?;
1018 let (start, len) = extent;
1019 let end = start.saturating_add(len);
1020 let all = self.index().oids_in_order()?;
1021 let refs: Vec<&[u8]> = all.iter().map(Vec::as_slice).collect();
1022 let extents = self.index().extents_batch(&refs);
1023 Ok(all
1024 .into_iter()
1025 .zip(extents)
1026 .filter_map(|(oid, at)| match at {
1027 Some((offset, _)) if offset >= start && offset < end => Some(oid),
1028 _ => None,
1029 })
1030 .collect())
1031 }
1032
1033 /// **Everything needed to emit a pack for EXACTLY `oids`** — nothing added,
1034 /// nothing dropped.
1035 ///
1036 /// Not one of the twelve. It sits beside [`absorb_pending`](Self::absorb_pending)
1037 /// for the same reason those do: it is about how storage is *shaped*, not
1038 /// about what a caller stores or reads.
1039 ///
1040 /// # 🔴 It used to add the delta bases, and that shipped broken clones
1041 ///
1042 /// Until 2026-08-11 this closed the request over its delta bases: an
1043 /// `OFS_DELTA` names its base by position, so a base outside the set cannot
1044 /// be encoded, and *adding the base* looked like the answer that is always
1045 /// safe. It is not, and the failure is silent on the server:
1046 ///
1047 /// ```text
1048 /// git clone --bare --single-branch --branch base <url>
1049 /// fatal: did not receive expected object 8601ec33920b7d701c9887fa04916501136d6e90
1050 /// fatal: fetch-pack: invalid index-pack output
1051 /// event="git.upload_pack.served" objects=33773 ← the server said SUCCESS
1052 /// ```
1053 ///
1054 /// Measured on the `h2h-linear-sha1-2048c-1024f-16k` fixture: `base` reaches
1055 /// **31 805** objects, the delta-base closure added **1 968** more — 33 773,
1056 /// exactly the count the server logged — and 507 of those additions were
1057 /// **trees**. A tree that is not reachable from the wants drags its own
1058 /// children in as a *requirement*, because `git index-pack
1059 /// --check-self-contained-and-connected` (which is what a clone runs, and
1060 /// which sets `strict`) walks every received object's links and then demands
1061 /// each one exist. Those 507 trees named **204** objects the pack did not
1062 /// contain. `8601ec33…` is one of them.
1063 ///
1064 /// So a delta base pulled into a narrowed request is not free: it satisfies
1065 /// the *pack format* and violates *connectivity*. Closing over the added
1066 /// bases' children in turn does terminate — the same fixture reaches a fixed
1067 /// point at 34 534 objects — but it is an over-send by construction, and on
1068 /// a `--filter=blob:none` fetch it adds back exactly the blobs the filter
1069 /// excluded, which is the defect
1070 /// `emit_pack_emits_exactly_the_set_it_is_given_and_never_its_closure`
1071 /// exists to forbid.
1072 ///
1073 /// # What it does instead: the base decides how the entry is encoded
1074 ///
1075 /// The set is never widened. For each requested object:
1076 ///
1077 /// * **base inside the request** → the stored entry is copied byte for byte,
1078 /// which is the whole point of this engine and is what a full clone does
1079 /// for every single object;
1080 /// * **base outside the request** → the entry is **rebuilt** from §14's
1081 /// resolved content and marked
1082 /// [`EmitEntry::recompressed`](crate::pack_walk::EmitEntry::recompressed) —
1083 /// as a computed delta against a base the request *does* carry where one
1084 /// can be found (see [`Self::in_request_chain_ancestor`] and
1085 /// [`crate::delta`]), and whole where it cannot.
1086 ///
1087 /// That is what stock `pack-objects` does — it reuses a stored delta only
1088 /// when the base is also being packed — and it keeps the receipt a fact:
1089 /// `copied` and `recompressed` add up to the objects written, and
1090 /// `recompressed` is **0 for a whole-repository clone** because such a
1091 /// selection contains every base. The cost lands on the boundary a narrowed
1092 /// request cuts through and nowhere else: 1 968 of 33 773 entries, 5.9 %, on
1093 /// the fixture above.
1094 ///
1095 /// `ofs_delta_ok` is the client's `ofs-delta` capability. An old client that
1096 /// cannot parse one is not refused any more: the entry is re-headed as a
1097 /// `REF_DELTA` naming its base by oid, **carrying the same compressed delta
1098 /// payload**, so that arm stays a copy too. Only a base outside the request
1099 /// costs a rebuild.
1100 ///
1101 /// # `thin_haves` — the boundary entry that does NOT have to be rebuilt
1102 ///
1103 /// `Some(tips)` says the client consented to a thin pack *and* the
1104 /// negotiation found common ground; [`crate::serve::GitStore::emit_oids`] is
1105 /// where the two halves are required together, because either alone is a
1106 /// failed fetch rather than a smaller one. Given it, a base outside the
1107 /// request is checked against [`the client's own objects`](GitStore::client_bases),
1108 /// and one the client already holds is named by oid in a `REF_DELTA` that
1109 /// carries the **stored delta stream unchanged**. That entry counts as
1110 /// `copied`, because it is: nothing was inflated and nothing was deflated.
1111 ///
1112 /// `None` — a clone, or a client that did not ask for a thin pack — behaves
1113 /// exactly as before, and a **full clone is byte-identical either way**,
1114 /// because a whole-repository request has no base outside itself for any of
1115 /// this to apply to.
1116 ///
1117 /// It does nothing for the *clone* half of the boundary cost, and cannot: a
1118 /// clone's receiver holds nothing, so there is no external base to name.
1119 /// That half is the arm below it — [`crate::delta`] computes a delta against
1120 /// something the pack **does** carry, which is the only move left when there
1121 /// is no receiver to lean on. The two do not overlap: `thin_haves` reaches a
1122 /// boundary entry first and costs nothing when it fires, so a fetch stays a
1123 /// pure copy and only a clone pays for a computed delta.
1124 ///
1125 /// # An oid this repository does not hold is REFUSED
1126 ///
1127 /// It used to be skipped, which is the same silent-under-send shape as the
1128 /// closure defect above wearing a different hat: N asked, N-1 emitted,
1129 /// `PackStats` reporting success, and the client the first to know. gunnar's
1130 /// in-memory arm has always refused it, so the skip also meant the two
1131 /// engines behind one contract disagreed about what a missing object means.
1132 /// A partial-clone filter does not depend on the skip and cannot: a filter
1133 /// *removes* oids from the selection, it never adds one the store lacks.
1134 ///
1135 /// # Cost
1136 ///
1137 /// O(objects requested). The previous shape read `oids_in_order()` and
1138 /// `lookup_batch` over the **whole repository** on every call, to build an
1139 /// offset→oid map it needed only for the closure; with no closure the
1140 /// selection's own offsets answer "is this base in the request", so that
1141 /// scan is gone and no reverse index has to exist.
1142 pub fn emit_set(
1143 &self,
1144 oids: &[Oid<'_>],
1145 ofs_delta_ok: bool,
1146 thin_haves: Option<&[Oid<'_>]>,
1147 ) -> Result<Vec<crate::pack_walk::EmitEntry>> {
1148 use crate::index_layout::{ObjType, ObjectIndex as _};
1149 use std::collections::hash_map::Entry;
1150
1151 self.absorb_pending()
1152 .context("absorbing pending index work before emitting a pack")?;
1153
1154 // Deduplicate, preserving the caller's order. A repeated oid in the
1155 // request must not become a repeated entry in the pack.
1156 //
1157 // **A map and not a set**, keyed oid → position in `asked` (which is the
1158 // position in `rows`). The membership question is the one it always was;
1159 // what the position buys is the base's *archive offset* for a
1160 // `REF_DELTA`, which names its base by oid and can learn where that base
1161 // sits no other way. It is a `usize` per requested object on a vector
1162 // this function already builds — not a second pass and not a second
1163 // index lookup.
1164 let mut asked: Vec<&[u8]> = Vec::with_capacity(oids.len());
1165 let mut seen: HashMap<&[u8], usize> = HashMap::with_capacity(oids.len());
1166 for oid in oids {
1167 if let Entry::Vacant(slot) = seen.entry(*oid) {
1168 slot.insert(asked.len());
1169 asked.push(*oid);
1170 }
1171 }
1172
1173 let rows = self.index().lookup_batch(&asked);
1174
1175 // **One mapping for the whole emit set.** Only the header probes below
1176 // read through it — the entries themselves are recorded as extents and
1177 // resolved by [`Self::resolve_emit_payloads`], which takes its own
1178 // snapshot of the same [`ArchiveMap`](crate::archive_map::ArchiveMap) and
1179 // therefore the same `Arc` unless a push landed in between.
1180 let snap = self.archive_snapshot()?;
1181
1182 // The two questions "is this base in the request" are asked in the two
1183 // coordinate systems the two delta forms use, and both are answered off
1184 // the request itself rather than off the index.
1185 //
1186 // **Offset → position in the request**, not a bare set of offsets. The
1187 // membership question is the same one it always was; what the position
1188 // buys is the base's *oid*, which the re-delta arm below needs in order
1189 // to read the base's content. It is the same single pass over `rows`
1190 // that built the set, so it costs a `usize` per requested object and no
1191 // extra work — and it is emphatically not the whole-repository
1192 // offset→oid scan the old delta-base closure used to do.
1193 let in_set_offsets: HashMap<u64, usize> = rows
1194 .iter()
1195 .enumerate()
1196 .filter_map(|(i, r)| r.as_ref().map(|r| (r.offset, i)))
1197 .collect();
1198 let oid_len = self.hash_kind().oid_len();
1199
1200 // The base of the last boundary entry that was re-deltified, kept so a
1201 // run of entries hanging off one ancestor inflates it once. One slot and
1202 // not an LRU: measured on `h2h-linear-sha1-2048c-1024f-16k`, 961
1203 // boundary entries name 855 distinct stored bases, so there is no reuse
1204 // worth a cache that could hold a repository's worth of objects — and
1205 // this crate is under a live RSS budget.
1206 let mut base_content: Option<(u64, Vec<u8>)> = None;
1207
1208 // What the client already holds, built **lazily and at most once**. A
1209 // clone never asks for it (`thin_haves` is `None`), and a fetch every one
1210 // of whose bases is inside the request never reaches the line that builds
1211 // it either — only a request that actually cuts a delta chain pays, which
1212 // is the only request it can help.
1213 let mut client_bases: Option<ExternalBases> = None;
1214
1215 let mut out = Vec::with_capacity(asked.len());
1216 for (oid, row) in asked.iter().zip(rows.iter()) {
1217 // **An oid this repository does not hold is REFUSED, not dropped.**
1218 //
1219 // This used to `continue`, and that is the same silent-under-send
1220 // shape as the closure defect above: the request asks for N objects,
1221 // the pack carries N-1, the receipt says success and the client is
1222 // the first to find out. It also made the two engines disagree about
1223 // what a missing object *means* — gunnar's in-memory arm has always
1224 // said *"emit_pack was asked for X, which this store does not hold;
1225 // a pack emitter that skipped it would send a pack no client can
1226 // close"* — so a benchmark across the two was comparing two
1227 // contracts.
1228 //
1229 // No legitimate caller relies on the skip. Every set that reaches
1230 // here is built out of this store's own index: `select` answers from
1231 // the reachability bitmaps, and gunnar's own walk resolves each id
1232 // against the store before it selects it. A partial-clone filter is
1233 // no exception and is the case worth naming, because it is the one
1234 // that *looks* like it might be: `--filter=blob:none` **removes**
1235 // oids from the selection, it never adds one the store lacks, so a
1236 // filtered request is a smaller set of objects that are all present
1237 // and this refusal never fires on it.
1238 let Some(row) = row.as_ref() else {
1239 bail!(
1240 "this pack was asked for {}, which this repository does not hold; emitting \
1241 the rest would be a short pack reported as a success, so this refuses \
1242 instead",
1243 hex::encode(oid)
1244 );
1245 };
1246
1247 // Which base this entry names, if any, and whether the request holds
1248 // it. A `REF_DELTA`'s `delta_base` column is 0 by construction —
1249 // `DeltaBase::as_offset` reports 0 for a ref base and 0 is also the
1250 // no-base sentinel — so its base is read out of the entry's own
1251 // bytes, directly after the type/size varint.
1252 //
1253 // A base that is *outside* the request is carried in **both**
1254 // coordinate systems rather than as a bare `false`, because that is
1255 // what the thin arm below needs: an `OFS_DELTA` knows only where its
1256 // base sits, a `REF_DELTA` knows only what it is called, and
1257 // "does the client hold it" has to be asked in whichever one the
1258 // entry speaks.
1259 let base = match row.obj_type {
1260 ObjType::OfsDelta => match in_set_offsets.get(&row.delta_base) {
1261 Some(_) => BaseOf::Inside {
1262 at: row.delta_base,
1263 },
1264 None => BaseOf::Outside {
1265 offset: row.delta_base,
1266 oid: None,
1267 },
1268 },
1269 ObjType::RefDelta => {
1270 let head = self.extent(&snap, row.offset, row.len.min(64))?;
1271 let (_, _, n) = crate::pack_walk::type_and_size_of(&head)?;
1272 match head.get(n..n + oid_len) {
1273 // In the request — **and where**. `rows[i]` is `None`
1274 // only for an oid this store does not hold, which the
1275 // refusal above turns into an error the moment the loop
1276 // reaches it; until then the base is treated as outside,
1277 // which is the answer that cannot be wrong.
1278 Some(b) => match seen
1279 .get(b)
1280 .and_then(|i| rows[*i].as_ref())
1281 .map(|r| r.offset)
1282 {
1283 Some(at) => BaseOf::Inside { at },
1284 None => BaseOf::Outside {
1285 offset: 0,
1286 oid: Some(b.to_vec()),
1287 },
1288 },
1289 // A truncated entry names no base this can read. It is
1290 // not "inside", and it is not offered to the client
1291 // either — it falls through to the whole rebuild, which
1292 // is the answer that cannot be wrong.
1293 None => BaseOf::Outside {
1294 offset: 0,
1295 oid: None,
1296 },
1297 }
1298 }
1299 _ => BaseOf::Whole,
1300 };
1301
1302 if let BaseOf::Whole | BaseOf::Inside { .. } = base {
1303 let inside_at = match base {
1304 BaseOf::Inside { at } => Some(at),
1305 _ => None,
1306 };
1307 // **The whole-clone arm, and it reads not one byte here.** The
1308 // entry is recorded as its address in the archive; the bytes are
1309 // resolved against the mapping at emit time and go straight from
1310 // the page cache to the wire. This line used to be a `pread`
1311 // into a fresh `Vec` — 13.8 M of each on a `linux.git` clone, all
1312 // of them held at once, which is the 2314 MB peak RSS
1313 // [`crate::pack_walk::EntryBytes`] documents.
1314 let extent = crate::pack_walk::EntryBytes::Extent {
1315 offset: row.offset,
1316 len: row.len,
1317 };
1318 let (stored, obj_type, delta_base) = match (row.obj_type, inside_at) {
1319 // An `OFS_DELTA` for a client that cannot read one: re-head it
1320 // as a `REF_DELTA` naming the same base by oid. The compressed
1321 // payload is the delta stream either way, so this is still a
1322 // copy in the `recompressed` sense — nothing is inflated — but
1323 // the concatenation of a new header and an old payload exists
1324 // nowhere on disk, so it cannot be an extent and is `Owned`.
1325 (ObjType::OfsDelta, Some(_)) if !ofs_delta_ok => {
1326 let base_oid = self.oid_at_offset(row.delta_base, &asked, &rows)?;
1327 let bytes = self.extent(&snap, row.offset, row.len)?;
1328 (
1329 crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(
1330 &bytes, &base_oid,
1331 )?),
1332 ObjType::RefDelta,
1333 0,
1334 )
1335 }
1336 // 🚨 **A stored `REF_DELTA` whose base is IN THIS PACK is
1337 // re-headed as an `OFS_DELTA`, and that is not an
1338 // optimisation.**
1339 //
1340 // gitoxide — which is `gunnar_client`, `gunnar`'s own
1341 // `receive-pack`, and every other gix-based receiver — reads
1342 // every `OBJ_REF_DELTA` in an incoming pack as naming an
1343 // object *the receiver already has*.
1344 // `gix_pack::data::input::LookupRefDeltaObjectsIter` consults
1345 // the local object database and the bases it has already
1346 // spliced in, and **nowhere else**: never inside the pack it
1347 // is reading. A base that is in the pack and not in the
1348 // receiver's store is `input::Error::NotFound`, and the fetch
1349 // dies with `The object <base> could not be decoded or wasn't
1350 // found` — with the whole pack rejected, on a clone whose
1351 // target is empty by definition, so *every* such entry is
1352 // fatal. That is `S-003` / gitoxide#2882, and `gunnar-wire`'s
1353 // `Selection::client_has` carries the same rule for gunnar's
1354 // own emitter.
1355 //
1356 // Stock git resolves it happily, which is why this survived
1357 // five green `git clone` runs against the store that could
1358 // not serve one `gunnar_client` fetch (`gunnar.rewrite`
1359 // round 0 and `gunnar.year` week 0, all four znippy columns,
1360 // 2026-08-14 — the server logged the same request `served`).
1361 //
1362 // The entry keeps its compressed delta stream byte for byte;
1363 // only the header changes, from "base named by oid" to "base
1364 // named by distance", which is the naming every receiver in
1365 // existence resolves inside the pack. This is where such an
1366 // entry comes from in the first place: `git index-pack
1367 // --fix-thin` appends the base into the pack and leaves the
1368 // ref-delta naming it by oid, so any repository pushed to
1369 // more than once holds them.
1370 //
1371 // # What it costs, stated rather than glossed
1372 //
1373 // `Owned`, so this is the one arm that gives up the zero-copy
1374 // property for an entry it does **not** rebuild: the
1375 // compressed delta stream is memcpy'd once, because a new
1376 // header concatenated with an old payload exists nowhere on
1377 // disk and no extent can address it. Bounded by the
1378 // compressed size of the ref-deltas whose base is in the
1379 // request — not by the repository, and not by the pack.
1380 //
1381 // It could be avoided: `emit_pack` re-encodes every header
1382 // from `EmitEntry::obj_type` while `header_len` finds the
1383 // payload from the STORED bytes' own type, so an `Extent`
1384 // over the untouched ref-delta declared as an `OfsDelta`
1385 // would emit the identical pack with no copy at all. That is
1386 // deliberately not done: it makes an `EmitEntry` disagree
1387 // with the bytes it points at, and the entry model being
1388 // trustworthy is worth more than the memcpy. If the RSS
1389 // budget ever says otherwise, the honest form is a third
1390 // `EntryBytes` variant that carries the new header beside
1391 // the extent, not a silent disagreement.
1392 (ObjType::RefDelta, Some(at)) if ofs_delta_ok => {
1393 let bytes = self.extent(&snap, row.offset, row.len)?;
1394 (
1395 crate::pack_walk::EntryBytes::Owned(Self::as_ofs_delta(
1396 &bytes, oid_len, row.offset, at,
1397 )?),
1398 ObjType::OfsDelta,
1399 at,
1400 )
1401 }
1402 // Everything else copies. A `REF_DELTA` left as one here is
1403 // the `!ofs_delta_ok` client, which cannot be handed a
1404 // distance at all — and which is stock git older than 1.4.4,
1405 // never a gix receiver, since gix has always advertised
1406 // `ofs-delta`.
1407 _ => (extent, row.obj_type, row.delta_base),
1408 };
1409 out.push(crate::pack_walk::EmitEntry {
1410 oid: oid.to_vec(),
1411 stored,
1412 obj_type,
1413 uncompressed_size: row.uncompressed_size,
1414 delta_base,
1415 offset: row.offset,
1416 recompressed: false,
1417 deltified: false,
1418 });
1419 continue;
1420 }
1421
1422 // ── the base is outside the request ──────────────────────────────
1423 //
1424 // **Thin first.** If the client consented to a thin pack and the
1425 // negotiation vouches that it already holds this base, the entry
1426 // goes out as a `REF_DELTA` naming that base by oid, carrying the
1427 // **same compressed delta stream**. Nothing is inflated, nothing is
1428 // deflated, and the pack is smaller than the whole object by exactly
1429 // the margin the delta was worth. That is what `caps.thin` has always
1430 // been the allowance for.
1431 //
1432 // An `OFS_DELTA` cannot survive this: it names its base by a
1433 // backwards distance *within the pack*, and the base is not in the
1434 // pack. Re-heading it as a `REF_DELTA` is the same header swap the
1435 // no-`ofs-delta` client path does, over an unchanged payload.
1436 if let Some(haves) = thin_haves {
1437 if client_bases.is_none() {
1438 client_bases = Some(self.client_bases(haves)?);
1439 }
1440 let ext = client_bases
1441 .as_ref()
1442 .expect("the client's bases were just built");
1443 let base_oid: Option<Vec<u8>> = match &base {
1444 // Unreachable: the arm above `continue`s on both. Matched
1445 // rather than `unreachable!()` so that a future arm falling
1446 // through here declines the thin path instead of panicking
1447 // inside a serving thread.
1448 BaseOf::Whole | BaseOf::Inside { .. } => None,
1449 BaseOf::Outside {
1450 oid: Some(named), ..
1451 } => ext
1452 .held
1453 .contains(&OidKey::new(named)?)
1454 .then(|| named.clone()),
1455 BaseOf::Outside {
1456 offset,
1457 oid: None,
1458 } => ext.by_offset.get(offset).map(|k| k.as_slice().to_vec()),
1459 };
1460 if let Some(base_oid) = base_oid {
1461 // A stored `REF_DELTA` already names its base by oid, so its
1462 // bytes go out untouched and stay an extent. An `OFS_DELTA`
1463 // has to be re-headed, and the result is new bytes.
1464 let stored = if row.obj_type == ObjType::OfsDelta {
1465 let bytes = self.extent(&snap, row.offset, row.len)?;
1466 crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(&bytes, &base_oid)?)
1467 } else {
1468 crate::pack_walk::EntryBytes::Extent {
1469 offset: row.offset,
1470 len: row.len,
1471 }
1472 };
1473 out.push(crate::pack_walk::EmitEntry {
1474 oid: oid.to_vec(),
1475 stored,
1476 obj_type: ObjType::RefDelta,
1477 uncompressed_size: row.uncompressed_size,
1478 // The base is NOT in this pack, so nothing may be ordered
1479 // after it: `topological_order` reads 0 as "names no base
1480 // in here", which for an external base is the truth.
1481 delta_base: 0,
1482 offset: row.offset,
1483 // Copied. The payload never left the archive's bytes.
1484 recompressed: false,
1485 deltified: false,
1486 });
1487 continue;
1488 }
1489 }
1490
1491 // The base is not being sent and the client cannot supply it, so the
1492 // stored bytes cannot go out as they are. Two answers remain, and
1493 // the first is worth ~4× the second on the wire.
1494 let (kind, body) = self.content(oid)?.ok_or_else(|| {
1495 anyhow!(
1496 "{} is a delta whose base is outside this request, so it must be sent whole, \
1497 and this repository cannot produce its content — neither §14's exploded \
1498 table nor a re-derivation from the verbatim packs answered",
1499 hex::encode(oid)
1500 )
1501 })?;
1502
1503 // ── re-delta against a base the request DOES carry ────────────────
1504 //
1505 // The candidate comes from the entry's own stored delta chain: the
1506 // nearest ancestor that is inside the request. That is the packer's
1507 // original similarity judgement, already made and already written
1508 // down — no window, no sort, no scan of the repository.
1509 let candidate = if crate::delta::enabled() {
1510 self.in_request_chain_ancestor(row.delta_base, row.offset, &in_set_offsets)?
1511 } else {
1512 None
1513 };
1514 if let Some(base_at) = candidate {
1515 let at = *in_set_offsets
1516 .get(&base_at)
1517 .expect("the ancestor was found by lookup in this very map");
1518 // Reuse the last base if this entry hangs off the same one.
1519 if base_content.as_ref().is_none_or(|(o, _)| *o != base_at) {
1520 let base_oid = asked[at];
1521 let (_, base_body) = self.content(base_oid)?.ok_or_else(|| {
1522 anyhow!(
1523 "{} is in this request and is the delta base chosen for {}, and this \
1524 repository cannot produce its content",
1525 hex::encode(base_oid),
1526 hex::encode(oid)
1527 )
1528 })?;
1529 base_content = Some((base_at, base_body));
1530 }
1531 let (_, base_body) = base_content.as_ref().expect("just filled");
1532
1533 // `None` is the encoder declining — an unrelated base, or a
1534 // delta that came out no smaller than the object. Falling
1535 // through to the whole rebuild is then strictly right.
1536 if let Some(d) = crate::delta::delta(base_body, &body)? {
1537 let stored = Self::ofs_delta_entry(&d, row.offset, base_at)?;
1538 // A client with no `ofs-delta` is served the same computed
1539 // delta under a `REF_DELTA` head naming the base by oid —
1540 // the identical header swap the base-inside arm above does,
1541 // over a payload that is already built.
1542 // **The `Owned` exception, in its primary shape.** These
1543 // bytes were computed here and exist on no disk, so no
1544 // extent addresses them; carrying them on the entry is
1545 // correct and is bounded by the boundary a narrowed request
1546 // cuts through (961 of 33 773 entries on the fixture below),
1547 // never by the repository.
1548 let (stored, obj_type, delta_base) = if ofs_delta_ok {
1549 (
1550 crate::pack_walk::EntryBytes::Owned(stored),
1551 ObjType::OfsDelta,
1552 base_at,
1553 )
1554 } else {
1555 (
1556 crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(
1557 &stored, asked[at],
1558 )?),
1559 ObjType::RefDelta,
1560 0,
1561 )
1562 };
1563 out.push(crate::pack_walk::EmitEntry {
1564 oid: oid.to_vec(),
1565 stored,
1566 obj_type,
1567 uncompressed_size: row.uncompressed_size,
1568 delta_base,
1569 offset: row.offset,
1570 // Inflated and re-deflated, so NOT a copy — and the
1571 // receipt says which of the two rebuilds ran.
1572 recompressed: true,
1573 deltified: true,
1574 });
1575 continue;
1576 }
1577 }
1578
1579 // Ship the object whole rather than dragging an unreachable base —
1580 // and its children — into a clone.
1581 out.push(crate::pack_walk::EmitEntry {
1582 oid: oid.to_vec(),
1583 // Inflated and re-deflated from §14's resolved content: the
1584 // `Owned` exception again, and again bounded by the boundary.
1585 stored: crate::pack_walk::EntryBytes::Owned(Self::whole_entry(kind, &body)?),
1586 obj_type: crate::serve::resolved_type(kind),
1587 uncompressed_size: body.len() as u64,
1588 // Whole: it names no base, and nothing may order it after one.
1589 delta_base: 0,
1590 offset: row.offset,
1591 recompressed: true,
1592 deltified: false,
1593 });
1594 }
1595 Ok(out)
1596 }
1597
1598 /// **PHASE 1 of emitting: resolve every entry's stored bytes, in parallel.**
1599 ///
1600 /// Emission splits in two, and the split is not a preference — it is where
1601 /// the ordering dependency actually is:
1602 ///
1603 /// | phase | what it does | why |
1604 /// |---|---|---|
1605 /// | 1, **here**, parallel | address → bytes for every entry | the archive is immutable, so N workers over disjoint (or shared) extents need **no lock at all** |
1606 /// | 2, [`crate::pack_walk::emit_pack`], serial | assign output offsets, encode `OFS_DELTA` distances, write, hash | an `OFS_DELTA` names its base by **distance back in the output pack**, so entry *n* cannot be encoded until every earlier entry's byte length is known |
1607 ///
1608 /// Phase 2 is not parallelisable and no attempt is made: the distance chain
1609 /// is a genuine serial dependency, and a running sha1 is another.
1610 ///
1611 /// # No lock, and why that is a fact rather than an intention
1612 ///
1613 /// [`Self::adopt_journal`] states the storage property this rests on — *"the
1614 /// blob file is append-only and `gc` truncates nothing, so a retired pack's
1615 /// bytes are still exactly where the journal says they are"*. Every byte a
1616 /// snapshot covers is therefore immutable for the life of the file, so the
1617 /// workers below share one `&Mapped` and read it concurrently with no
1618 /// synchronisation of any kind. There is no mutex here; if one were needed
1619 /// the data model would be wrong.
1620 ///
1621 /// # 🚨 This is a fan-out inside the serving tier, and `crate::serve`'s
1622 /// header says that tier is serial
1623 ///
1624 /// It did, and the reason it gave is still true and still respected: the
1625 /// constellation's fan-out primitive has no reentrancy detection, so an
1626 /// unbounded fan-out per request inside an admission gate that already runs
1627 /// 32 transfers spawns W² threads. That is why this is **bounded twice**:
1628 ///
1629 /// * it does not fan out at all below [`FAN_OUT_AT`] entries, so every fetch
1630 /// small enough for thread-spawn to dominate runs on the calling thread
1631 /// exactly as before, and
1632 /// * it never exceeds [`MAX_RESOLVE_WORKERS`] (override with
1633 /// `ZNIPPY_GIT_EMIT_WORKERS`), so the worst case is 32 × 4 and not
1634 /// 32 × ncores.
1635 ///
1636 /// # What the fan-out is worth — stated honestly
1637 ///
1638 /// Resolving a mapped extent is a bounds check and a pointer, so the work
1639 /// this parallelises is **not** the resolve itself. It is the two things
1640 /// beside it: the first-touch **page fault** on each entry's header (the
1641 /// grammar is parsed here, which is what forces that fault, and it validates
1642 /// the entry before phase 2 commits a byte to the wire), and the `pread`
1643 /// fallback for any extent the mapping does not cover. On a warm page cache
1644 /// with the whole archive mapped, phase 1 is nearly free and the fan-out
1645 /// buys correspondingly little — the 20 % of cycles this change is aimed at
1646 /// is recovered by *not copying*, not by threads. I could not measure the
1647 /// fan-out's own contribution separately: oden was at loadavg 25.9 while
1648 /// this was written (2026-08-14), and a shared box cannot answer a question
1649 /// that fine.
1650 pub(crate) fn resolve_emit_payloads<'a>(
1651 &self,
1652 entries: &'a [crate::pack_walk::EmitEntry],
1653 snap: &'a crate::archive_map::Mapped,
1654 ) -> Result<Vec<std::borrow::Cow<'a, [u8]>>>
1655 where
1656 Self: Sync,
1657 {
1658 let workers = if entries.len() < FAN_OUT_AT {
1659 1
1660 } else {
1661 emit_workers()
1662 };
1663 self.resolve_with(entries, snap, workers)
1664 }
1665
1666 /// [`Self::resolve_emit_payloads`] with the worker count named rather than
1667 /// decided — `1` runs inline on the calling thread.
1668 ///
1669 /// Split out so the fan-out is **testable at a size a test fixture can
1670 /// reach**. The production threshold is 4096 entries and the corpus these
1671 /// guards run on holds 2687, so a test that went through the front door
1672 /// would exercise the serial arm every time and the gatling arm never;
1673 /// `the_fan_out_resolves_the_identical_bytes_the_serial_pass_does` calls
1674 /// this with 1 and with 4 over the same set and requires the two to agree
1675 /// byte for byte.
1676 pub(crate) fn resolve_with<'a>(
1677 &self,
1678 entries: &'a [crate::pack_walk::EmitEntry],
1679 snap: &'a crate::archive_map::Mapped,
1680 workers: usize,
1681 ) -> Result<Vec<std::borrow::Cow<'a, [u8]>>>
1682 where
1683 Self: Sync,
1684 {
1685 use znippy_zoomies::gatling_forkjoin::gatling_for_each;
1686
1687 let one = |i: usize| -> Result<std::borrow::Cow<'a, [u8]>> {
1688 let e = &entries[i];
1689 let bytes = match &e.stored {
1690 // The bytes live on the entry — the rebuilt-delta exception.
1691 // Borrowed, never cloned: `EntryBytes::Owned` is already the
1692 // only allocation this path makes and copying it would double it.
1693 crate::pack_walk::EntryBytes::Owned(v) => std::borrow::Cow::Borrowed(v.as_slice()),
1694 crate::pack_walk::EntryBytes::Extent { offset, len } => {
1695 self.extent(snap, *offset, *len)?
1696 }
1697 };
1698 // Parse the entry's own header now. Two things fall out: a malformed
1699 // entry is refused **before** phase 2 has written any of the pack,
1700 // and the first cache line of every entry is touched here — on
1701 // whichever worker got it — rather than one at a time on the serial
1702 // writer.
1703 crate::pack_walk::type_and_size_of(&bytes).with_context(|| {
1704 format!(
1705 "resolving the stored bytes of {} for emission",
1706 hex::encode(&e.oid)
1707 )
1708 })?;
1709 Ok(bytes)
1710 };
1711
1712 if workers <= 1 {
1713 return (0..entries.len()).map(one).collect();
1714 }
1715 gatling_for_each(entries.len(), workers, one)
1716 .into_iter()
1717 .collect()
1718 }
1719
1720 /// **Both phases, one call** — the only way a caller in this crate turns an
1721 /// ordered emit set into pack bytes.
1722 ///
1723 /// It is three lines, and it is a function rather than three lines because
1724 /// LAW 5's *fix by construction* applies exactly here: the snapshot has to
1725 /// outlive the slices taken from it, and a caller that took its own snapshot
1726 /// per entry, or dropped it between the phases, would be writing a pack out
1727 /// of a mapping that no longer exists. Routing `emit_oids` and every test
1728 /// through one writer makes that impossible to get wrong twice.
1729 pub(crate) fn emit_ordered(
1730 &self,
1731 ordered: &[crate::pack_walk::EmitEntry],
1732 out: &mut dyn std::io::Write,
1733 ) -> Result<crate::pack_walk::EmitReport>
1734 where
1735 Self: Sync,
1736 {
1737 let snap = self
1738 .archive_snapshot()
1739 .context("mapping the archive to emit")?;
1740 let payloads = self
1741 .resolve_emit_payloads(ordered, &snap)
1742 .context("resolving the emit set's stored bytes")?;
1743 crate::pack_walk::emit_pack(
1744 ordered,
1745 self.hash_kind(),
1746 out,
1747 &|i| Ok(payloads[i].as_ref()),
1748 )
1749 }
1750
1751 /// **The nearest ancestor of `from`'s stored delta chain that is inside the
1752 /// request**, or `None` when the chain leaves the request and never comes
1753 /// back.
1754 ///
1755 /// `from` is the archive offset of a base already known to be *outside* the
1756 /// request; `entry_at` is the offset of the entry that named it.
1757 ///
1758 /// # Why this cannot make a cycle, and why that is not a comment but an
1759 /// invariant
1760 ///
1761 /// The answer is required to satisfy `answer < entry_at`. In a well-formed
1762 /// pack a base always precedes the delta that names it, so every edge in the
1763 /// stored graph already points from a higher offset to a lower one; a new
1764 /// edge with the same property keeps the emitted graph a DAG **by
1765 /// construction**, whatever else the request contains. That matters because
1766 /// the obvious better candidate — the entry's own delta *children*, one
1767 /// revision away instead of two to nine — is exactly the set that would
1768 /// close a cycle: a child's stored base is this entry, so pointing this
1769 /// entry at the child makes `topological_order` unable to order either, and
1770 /// `emit_pack` refuses the set. Measured cost of the restriction on
1771 /// `h2h-linear-sha1-2048c-1024f-16k`: ~1 470 bytes per boundary entry
1772 /// against ~314 for the cyclic candidate, and 5 800 for shipping it whole.
1773 ///
1774 /// An `OFS_DELTA` decreases the offset every hop, so the bound is free
1775 /// there; a `REF_DELTA` names its base by oid and could point anywhere, so
1776 /// the comparison is made rather than assumed.
1777 ///
1778 /// # Cost
1779 ///
1780 /// One 64-byte `pread` per hop, no inflate, no allocation sized by an
1781 /// object — the same walk and the same ceiling as
1782 /// [`crate::serve::GitStore::resolved_type_at`], for the same reason: a
1783 /// pushed pack's back-references are not under this server's control, so a
1784 /// cycle must terminate rather than spin. Measured hop counts on the fixture
1785 /// above: 2 for 392 of the 859 that resolve, 3 for 260, and 9 at the worst.
1786 fn in_request_chain_ancestor(
1787 &self,
1788 from: u64,
1789 entry_at: u64,
1790 in_request: &HashMap<u64, usize>,
1791 ) -> Result<Option<u64>> {
1792 use crate::index_layout::{ObjType, ObjectIndex as _};
1793
1794 /// git's own `pack.depth` ceiling, doubled — the same bound
1795 /// `resolved_type_at` carries and for the same reason.
1796 const MAX_LINKS: usize = 100;
1797 /// Enough of an entry to read its type/size varint plus either an
1798 /// `OFS_DELTA` distance or a `REF_DELTA` oid.
1799 const HEADER_PROBE: u64 = 64;
1800
1801 let oid_len = self.hash_kind().oid_len();
1802 let mut at = from;
1803 for _ in 0..MAX_LINKS {
1804 if at != 0 && at < entry_at && in_request.contains_key(&at) {
1805 return Ok(Some(at));
1806 }
1807 if at == 0 {
1808 return Ok(None);
1809 }
1810 let head = self.read_extent(at, HEADER_PROBE)?;
1811 let (t, _, n) = crate::pack_walk::type_and_size_of(&head)?;
1812 at = match t {
1813 // A whole object with no base: the chain ends here, and it was
1814 // not in the request.
1815 ObjType::Commit | ObjType::Tree | ObjType::Blob | ObjType::Tag => {
1816 return Ok(None)
1817 }
1818 ObjType::OfsDelta => {
1819 let (distance, _) = crate::pack_walk::ofs_distance_of(&head[n..])?;
1820 match at.checked_sub(distance) {
1821 Some(next) if next != 0 => next,
1822 // A base before the start of the archive is a corrupt
1823 // back-reference. The whole rebuild is always a correct
1824 // answer, so this declines rather than failing a clone
1825 // over an entry it was only ever going to optimise.
1826 _ => return Ok(None),
1827 }
1828 }
1829 ObjType::RefDelta => {
1830 let Some(base_oid) = head.get(n..n + oid_len) else {
1831 return Ok(None);
1832 };
1833 match self.index().lookup(base_oid) {
1834 Some(row) => row.offset,
1835 None => return Ok(None),
1836 }
1837 }
1838 };
1839 }
1840 Ok(None)
1841 }
1842
1843 /// An `OFS_DELTA` pack entry carrying `d`, deflated.
1844 ///
1845 /// The distance written here is the **archive** distance, which is the same
1846 /// coordinate every stored entry's distance is in;
1847 /// [`crate::pack_walk::emit_pack`] rewrites it into output coordinates for
1848 /// every entry it emits, copied or not, so this only has to be well-formed
1849 /// and self-consistent — [`crate::pack_walk::header_len`] parses it back to
1850 /// find where the payload starts.
1851 ///
1852 /// `Compression::fast` for the same reason [`Self::whole_entry`] uses it,
1853 /// and the trade is far better here: the buffer being deflated is the delta,
1854 /// which on the measured fixture is ~1.4 KB against the object's 16 KiB, so
1855 /// this path deflates roughly a **tenth** of the bytes the whole rebuild it
1856 /// replaces does.
1857 fn ofs_delta_entry(d: &[u8], entry_at: u64, base_at: u64) -> Result<Vec<u8>> {
1858 use crate::index_layout::ObjType;
1859 use std::io::Write as _;
1860
1861 let distance = entry_at.checked_sub(base_at).ok_or_else(|| {
1862 anyhow!(
1863 "a computed delta at archive offset {entry_at} names a base at {base_at}, which \
1864 is after it — an ofs-delta distance is backwards and this would not encode"
1865 )
1866 })?;
1867 let mut out = Vec::with_capacity(d.len() / 2 + 32);
1868 crate::pack_walk::encode_type_and_size(&mut out, ObjType::OfsDelta, d.len() as u64);
1869 crate::pack_walk::encode_ofs_distance(&mut out, distance);
1870 let mut z = flate2::write::ZlibEncoder::new(out, flate2::Compression::fast());
1871 z.write_all(d)?;
1872 Ok(z.finish()?)
1873 }
1874
1875 /// **The objects the receiver already held before this transfer**, in both
1876 /// coordinate systems a delta can name a base in.
1877 ///
1878 /// `haves` is the negotiated common **tips**, exactly as
1879 /// [`git_storage_trait::GitServe::emit_pack`] defines them, and the closure
1880 /// over them is this store's own — the contract deliberately passes tips
1881 /// rather than the walk's exclusion set, because the closure is
1882 /// repository-sized and the engine can rebuild it from its own bitmaps.
1883 ///
1884 /// # This is a VOUCHER, and the caller has to be able to stand behind it
1885 ///
1886 /// Every object named here may be pointed at by a delta the pack does not
1887 /// carry. If the receiver turns out **not** to hold one, its
1888 /// `index-pack --fix-thin` dies with `pack has N unresolved deltas`, so a
1889 /// caller that cannot vouch must not pass `caps.thin`. The case that matters
1890 /// is a **partial clone**: a `blob:none` client's `have` tips reach blobs it
1891 /// was deliberately never sent, so the closure over-states what it holds and
1892 /// a filtered fetch must not be marked thin.
1893 ///
1894 /// # Cost
1895 ///
1896 /// One bitmap union plus one point lookup per held object, and it happens at
1897 /// most once per request — and only on a request that has a base outside
1898 /// itself to place. The `by_offset` map exists because an `OFS_DELTA` knows
1899 /// its base only as an archive offset; building it over what the *client*
1900 /// holds rather than over the repository is what keeps this off the
1901 /// whole-repository scan the closure used to do.
1902 ///
1903 /// **And it is spent in raw bytes now, end to end.** This used to ask
1904 /// [`GitOps::reachable`] — hex out of the ordinal space, `hex::decode`d
1905 /// straight back to bytes, then a third `to_vec` per object to key the
1906 /// offset map, three heap allocations per held object for a set the size of
1907 /// the client's history. It asks [`GitStore::reachable_raw`] instead: one
1908 /// buffer for the closure, one hash table of [`OidKey`]s over it, and
1909 /// nothing per object.
1910 fn client_bases(&self, haves: &[Oid<'_>]) -> Result<ExternalBases> {
1911 use crate::index_layout::ObjectIndex as _;
1912
1913 let held = self.reachable_raw(haves, &[]).context(
1914 "closing over the negotiated common tips, to learn which bases the client can supply \
1915 for itself",
1916 )?;
1917 let refs: Vec<&[u8]> = held.iter().collect();
1918 let rows = self.index().lookup_batch(&refs);
1919 let mut by_offset = HashMap::with_capacity(refs.len());
1920 let mut keys = std::collections::HashSet::with_capacity(refs.len());
1921 for (oid, row) in refs.iter().zip(rows.iter()) {
1922 let key = OidKey::new(oid)?;
1923 if let Some(row) = row {
1924 by_offset.insert(row.offset, key);
1925 }
1926 keys.insert(key);
1927 }
1928 Ok(ExternalBases {
1929 by_offset,
1930 held: keys,
1931 })
1932 }
1933
1934 /// The oid of the entry at `offset`, looked up **in the request** — which is
1935 /// the only place it can be, because this is called for a base the request
1936 /// was just shown to contain.
1937 ///
1938 /// Linear over the request rather than a map, because it runs only on the
1939 /// `ofs-delta`-less client path: a capability no git since 1.4.4 omits, and
1940 /// one that has to be *advertised absent* to reach here at all.
1941 fn oid_at_offset(
1942 &self,
1943 offset: u64,
1944 asked: &[&[u8]],
1945 rows: &[Option<crate::index_layout::IndexRow>],
1946 ) -> Result<Vec<u8>> {
1947 asked
1948 .iter()
1949 .zip(rows)
1950 .find(|(_, r)| r.as_ref().is_some_and(|r| r.offset == offset))
1951 .map(|(oid, _)| oid.to_vec())
1952 .ok_or_else(|| {
1953 anyhow!(
1954 "the entry at archive offset {offset} was checked to be in this request and \
1955 then could not be found in it"
1956 )
1957 })
1958 }
1959
1960 /// Re-head a stored `OFS_DELTA` as a `REF_DELTA` naming `base_oid`.
1961 ///
1962 /// The compressed delta stream is copied unchanged; only the header differs,
1963 /// because the two forms differ in **how the base is named** and in nothing
1964 /// else. This is why a client without `ofs-delta` costs bytes and not CPU.
1965 fn as_ref_delta(stored: &[u8], base_oid: &[u8]) -> Result<Vec<u8>> {
1966 use crate::index_layout::ObjType;
1967 let (t, stated_size, n) = crate::pack_walk::type_and_size_of(stored)?;
1968 if t != ObjType::OfsDelta {
1969 bail!("only an ofs-delta can be re-headed as a ref-delta, this entry is {t:?}");
1970 }
1971 let (_, d) = crate::pack_walk::ofs_distance_of(stored.get(n..).unwrap_or(&[]))?;
1972 let mut out = Vec::with_capacity(stored.len() + base_oid.len());
1973 crate::pack_walk::encode_type_and_size(&mut out, ObjType::RefDelta, stated_size);
1974 out.extend_from_slice(base_oid);
1975 out.extend_from_slice(stored.get(n + d..).unwrap_or(&[]));
1976 Ok(out)
1977 }
1978
1979 /// Re-head a stored `REF_DELTA` as an `OFS_DELTA` naming the base that sits
1980 /// at `base_at` — **the inverse of [`Self::as_ref_delta`]**, and the fix for
1981 /// the one pack shape gitoxide cannot read.
1982 ///
1983 /// The compressed delta stream is copied unchanged; only the header differs,
1984 /// because the two forms differ in how the base is *named* and in nothing
1985 /// else. The reason this has to exist is in the caller: a `REF_DELTA` whose
1986 /// base is inside the same pack is resolved by stock git and refused by
1987 /// every gix-based receiver.
1988 ///
1989 /// # The distance written here is a placeholder, and that is by design
1990 ///
1991 /// Exactly as [`Self::ofs_delta_entry`] documents: the distance an
1992 /// `OFS_DELTA` carries is relative to a position in the pack it is *in*, so
1993 /// [`crate::pack_walk::emit_pack`] re-encodes it for every entry from the
1994 /// output offsets it is assigning, and [`crate::pack_walk::header_len`]
1995 /// parses whatever is written here only to find where the payload starts. It
1996 /// therefore has to be well-formed and self-consistent, and nothing more.
1997 ///
1998 /// It cannot simply be `entry_at - base_at`: a `REF_DELTA` is allowed to
1999 /// name a base that comes **after** it in the pack, and `git index-pack
2000 /// --fix-thin` produces exactly that — it appends the base at the end. So a
2001 /// forward reference encodes as `1`, and `emit_pack` writes the real
2002 /// backwards distance once `topological_order` has put the base first.
2003 fn as_ofs_delta(
2004 stored: &[u8],
2005 oid_len: usize,
2006 entry_at: u64,
2007 base_at: u64,
2008 ) -> Result<Vec<u8>> {
2009 use crate::index_layout::ObjType;
2010 let (t, stated_size, n) = crate::pack_walk::type_and_size_of(stored)?;
2011 if t != ObjType::RefDelta {
2012 bail!("only a ref-delta can be re-headed as an ofs-delta, this entry is {t:?}");
2013 }
2014 let payload = stored.get(n + oid_len..).ok_or_else(|| {
2015 anyhow!(
2016 "a ref-delta entry at archive offset {entry_at} is {} bytes, which is not even its \
2017 header plus a {oid_len}-byte base oid",
2018 stored.len()
2019 )
2020 })?;
2021 let mut out = Vec::with_capacity(payload.len() + 32);
2022 crate::pack_walk::encode_type_and_size(&mut out, ObjType::OfsDelta, stated_size);
2023 crate::pack_walk::encode_ofs_distance(
2024 &mut out,
2025 entry_at.checked_sub(base_at).filter(|d| *d > 0).unwrap_or(1),
2026 );
2027 out.extend_from_slice(payload);
2028 Ok(out)
2029 }
2030
2031 /// A whole (non-delta) pack entry for `body`: type/size header, then the
2032 /// body deflated.
2033 ///
2034 /// The **only** place in this crate that deflates on a serving path, and the
2035 /// only reason it exists is a delta whose base the request does not contain.
2036 /// `Compression::fast` deliberately: these bytes are re-compressed because
2037 /// they *cannot* be copied, so the trade is wire size against a clone's
2038 /// latency, and the entries this runs on are a small boundary fraction of any
2039 /// real request.
2040 fn whole_entry(kind: crate::object::GitObjectKind, body: &[u8]) -> Result<Vec<u8>> {
2041 use std::io::Write as _;
2042 let mut out = Vec::with_capacity(body.len() / 2 + 32);
2043 crate::pack_walk::encode_type_and_size(
2044 &mut out,
2045 crate::serve::resolved_type(kind),
2046 body.len() as u64,
2047 );
2048 let mut z = flate2::write::ZlibEncoder::new(out, flate2::Compression::fast());
2049 z.write_all(body)?;
2050 Ok(z.finish()?)
2051 }
2052
2053 /// Take the absorb gate and hold it.
2054 ///
2055 /// The only way to observe "a read that arrived before the drain" as a
2056 /// **state** rather than as a race: with this held, the worker is parked at
2057 /// the top of its absorb, so the pack is provably durable, provably queued
2058 /// and provably not indexed. Its guard is in this module's tests, and nothing
2059 /// on a serving path takes it.
2060 #[cfg(test)]
2061 pub(crate) fn hold_absorb_gate(&self) -> std::sync::MutexGuard<'_, ()> {
2062 self.absorber.gate.lock().expect("absorb gate poisoned")
2063 }
2064
2065 /// `pread` an extent out of the verbatim archive, into a fresh `Vec`.
2066 ///
2067 /// **The fallback and the owning path.** A caller that can hold a borrow
2068 /// wants [`Self::archive_snapshot`] plus [`Self::extent`] instead: this one
2069 /// costs a syscall, an allocation and a kernel→user copy, which is the shape
2070 /// [`crate::archive_map`] exists to get off the emit path.
2071 pub(crate) fn read_extent(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
2072 self.absorber.read_extent(offset, len)
2073 }
2074
2075 /// One mapping of the verbatim archive, to be taken **once per operation**
2076 /// and sliced N times. See [`crate::archive_map::ArchiveMap::snapshot`].
2077 pub(crate) fn archive_snapshot(&self) -> Result<Arc<crate::archive_map::Mapped>> {
2078 self.absorber.map.snapshot()
2079 }
2080
2081 /// **The one resolver**: an extent's bytes, borrowed from `snap` when it
2082 /// covers them and `pread` into a fresh buffer when it does not.
2083 ///
2084 /// `Cow::Borrowed` is the ordinary answer and the whole point — no syscall,
2085 /// no allocation, no copy, just a bounds-checked slice of the page cache.
2086 /// `Cow::Owned` happens for an extent past the mapped end, which means the
2087 /// blob file grew after the snapshot was taken (a push landed mid-clone).
2088 /// `Mapped::get` returning `None` is *"go and pread it"* and never *"there
2089 /// are no bytes"*, so the two arms return the **same bytes** and differ only
2090 /// in what they cost — which is the property
2091 /// `an_extent_past_the_mapping_falls_back_and_returns_the_same_bytes`
2092 /// asserts.
2093 pub(crate) fn extent<'m>(
2094 &self,
2095 snap: &'m crate::archive_map::Mapped,
2096 offset: u64,
2097 len: u64,
2098 ) -> Result<std::borrow::Cow<'m, [u8]>> {
2099 match snap.get(offset, len) {
2100 Some(b) => Ok(std::borrow::Cow::Borrowed(b)),
2101 None => Ok(std::borrow::Cow::Owned(self.read_extent(offset, len)?)),
2102 }
2103 }
2104
2105 /// Record a durable pack as pending index work.
2106 ///
2107 /// The **bit going clear**, and the ack path's last act before it returns.
2108 pub(crate) fn queue(&self, pack_id: u64, extent: Extent) -> Result<()> {
2109 self.absorber.queue(pack_id, extent)
2110 }
2111
2112 fn reach_bitmaps(&self) -> Result<Arc<Vec<ReachEntry>>> {
2113 self.absorber.reach_bitmaps()
2114 }
2115
2116 /// [`Absorber::reach_bitmaps_with`], for the differential guard.
2117 pub(crate) fn reach_bitmaps_with(
2118 &self,
2119 policy: crate::reach::ReachPolicy,
2120 cache: bool,
2121 ) -> Result<Arc<Vec<ReachEntry>>> {
2122 self.absorber.reach_bitmaps_with(policy, cache)
2123 }
2124
2125 /// The commit oid set, raw, shared. See [`Derived::commit_raw`].
2126 pub(crate) fn commit_oids_raw(
2127 &self,
2128 ) -> Result<Option<Arc<std::collections::HashSet<Vec<u8>>>>> {
2129 self.absorber.commit_oids_raw()
2130 }
2131
2132 fn refold(&self) -> Result<()> {
2133 self.absorber.refold()
2134 }
2135}
2136
2137impl<S: ObjectIndex> Absorber<S> {
2138 fn open(
2139 blobs: &Path,
2140 hash: GitHashKind,
2141 objects: ObjectReadStack<S>,
2142 exploded: ExplodedArchive,
2143 ) -> Result<Self> {
2144 let reader =
2145 File::open(blobs).with_context(|| format!("opening {} for reads", blobs.display()))?;
2146 Ok(Self {
2147 blobs: blobs.to_path_buf(),
2148 reader,
2149 map: crate::archive_map::ArchiveMap::new(blobs),
2150 hash,
2151 objects,
2152 exploded,
2153 derived: RwLock::new(Derived::default()),
2154 packs: Mutex::new(PackState::default()),
2155 gate: Mutex::new(()),
2156 })
2157 }
2158
2159 /// Packs whose objects are not in the index yet — §13.12's bit, counted.
2160 fn unindexed_packs(&self) -> usize {
2161 self.packs.lock().expect("pack state lock").unabsorbed
2162 }
2163
2164 /// The bit going clear: this pack is durable and its objects are not in.
2165 ///
2166 /// Skips a pack the drain already absorbed — the ack path queues *after* it
2167 /// has handed the extent to the channel, so the worker can legitimately be
2168 /// finished before this runs, and clearing a bit for a pack that is already
2169 /// indexed would leave it clear for ever.
2170 fn queue(&self, pack_id: u64, extent: Extent) -> Result<()> {
2171 let mut s = self
2172 .packs
2173 .lock()
2174 .map_err(|_| anyhow!("pack state poisoned"))?;
2175 s.note(pack_id, extent);
2176 Ok(())
2177 }
2178
2179 /// **The bit, derived from the two durable facts, on open.**
2180 ///
2181 /// `journal` is every row this archive's journal holds, in append order.
2182 /// [`acked_packs`](crate::archive_write::acked_packs) is what turns that into
2183 /// the packs: its index *is* the pack ordinal — dense by construction, and
2184 /// re-derived the same way by every later open. For each one, the question is
2185 /// only *does this pack have any rows in the index*, which
2186 /// [`ObjectReadStack::extents_with_rows`] answers in one early-terminating
2187 /// scan rather than per pack.
2188 ///
2189 /// Returns the packs whose bit came up **clear**: durable bytes, no rows. The
2190 /// caller re-queues them on the account's channel, which is what makes a
2191 /// pack that was mid-absorb when the process died get absorbed by the next
2192 /// one instead of being durable-but-invisible for ever.
2193 ///
2194 /// The verbatim extents are also handed to [`Derived::packs`] — *all* of
2195 /// them, not only the unabsorbed ones. That is what lets [`BaseSource`]
2196 /// re-derive a delta base's content from the truth after a restart; without
2197 /// it a reopened store knows the rows of a pack it can no longer find the
2198 /// bytes of.
2199 ///
2200 /// # The one case where "no rows" does not mean "never absorbed" — and the
2201 /// third durable fact that now tells them apart
2202 ///
2203 /// [`GitOps::gc`] is the only thing that removes rows. A pack **all** of
2204 /// whose objects it finds dead therefore ends with a journal extent and no
2205 /// rows, which is byte for byte the state a pack that was in flight when the
2206 /// machine died leaves behind. This diff cannot tell them apart, and it must
2207 /// not guess: re-queueing the first resurrects every dead object, and *not*
2208 /// re-queueing the second loses a pack whose bytes were acked.
2209 ///
2210 /// So `gc` says which it is, durably, before it drops a row:
2211 /// [`retire_packs`](crate::archive_write::retire_packs) appends a tombstone
2212 /// naming the retired pack's offset, and a pack named by one is never
2213 /// re-queued here. The bit is still derived from durable facts only — there
2214 /// are now three of them (the journal's pack rows, the journal's tombstones,
2215 /// the index's rows) and none of them is a cache of another.
2216 ///
2217 /// A **partly** dead pack is not tombstoned and still has rows, so it takes
2218 /// the `absorbed` branch exactly as it always did.
2219 ///
2220 /// A tombstoned pack that *does* still have rows is a `gc` that was killed
2221 /// between the tombstone and the drop. It is treated as absorbed, which is
2222 /// what its rows say: nothing is lost, and the next `gc` finishes the job.
2223 ///
2224 /// Its extent still goes to [`Derived::packs`] either way. The blob file is
2225 /// append-only and `gc` truncates nothing, so a retired pack's bytes are
2226 /// still exactly where the journal says they are — and a live object may
2227 /// legitimately be a delta against a base that is itself unreachable, so
2228 /// dropping the extent from that list would break the resolve rather than
2229 /// tidy it.
2230 ///
2231 /// # The second durable fact: §14's exploded table
2232 ///
2233 /// A pack counts as absorbed only if the exploded table is **whole**, which
2234 /// is `exploded.rows() >= objects.len()` — one row per object, because §14's
2235 /// table is eager and skips nothing. If it is short, the table has been
2236 /// dropped (or a process died between the two commits) and **no** pack counts
2237 /// as absorbed: every one is re-queued and re-exploded from the verbatim
2238 /// truth. That is the whole of "droppable": deleting the file is a supported
2239 /// operation whose only cost is one re-ingest, and §13.12's rule holds
2240 /// unchanged over it — absent means fall back, never means wrong.
2241 ///
2242 /// `>=` and not `==` on purpose. A resolve that failed part way leaves rows
2243 /// for the objects it got to, and a `gc` retires both tables together, so the
2244 /// table can legitimately hold *more* than the index names; only being
2245 /// **short** is evidence of a gap.
2246 fn adopt_journal(&self, journal: &[Extent]) -> Result<Vec<PendingPack>> {
2247 if journal.is_empty() {
2248 return Ok(Vec::new());
2249 }
2250 // The packs, in ordinal order, and the offsets a `gc` retired. A
2251 // tombstone is a row and not a pack, so it holds no ordinal.
2252 let packs = crate::archive_write::acked_packs(journal);
2253 let retired = crate::archive_write::retired_offsets(journal);
2254 if packs.is_empty() {
2255 return Ok(Vec::new());
2256 }
2257 // Under `Graph` the table is short **by design** — commits and trees
2258 // only — so the row-count comparison would call every store dropped and
2259 // re-queue every pack on every open. It is weakened to "not empty",
2260 // which still catches the case this rule exists for (the file deleted),
2261 // and is stated rather than hidden: making it exact needs a durable
2262 // commit-and-tree count, which the index does not publish today.
2263 let exploded_is_whole = match self.exploded.policy() {
2264 crate::exploded_arrow::ExplodePolicy::Graph => {
2265 self.objects.len() == 0 || self.exploded.rows()? > 0
2266 }
2267 _ => self.exploded.rows()? >= self.objects.len() as u64,
2268 };
2269 let has_rows = if exploded_is_whole {
2270 self.objects.extents_with_rows(&packs)?
2271 } else {
2272 vec![false; packs.len()]
2273 };
2274 {
2275 let mut d = self
2276 .derived
2277 .write()
2278 .map_err(|_| anyhow!("derived poisoned"))?;
2279 d.packs = packs.clone();
2280 }
2281 let mut s = self
2282 .packs
2283 .lock()
2284 .map_err(|_| anyhow!("pack state poisoned"))?;
2285 let mut requeue = Vec::new();
2286 for (i, (&extent, &absorbed)) in packs.iter().zip(&has_rows).enumerate() {
2287 let pack_id = i as u64;
2288 if absorbed {
2289 s.mark_absorbed(pack_id, extent);
2290 } else if retired.contains(&extent.0) {
2291 // A `gc` dropped every row this pack had and said so durably
2292 // *before* dropping them. There is no index work owed for it,
2293 // ever: re-queueing it is precisely how its dead objects would
2294 // come back. The bit goes **up** rather than being left clear,
2295 // because a clear bit is what makes a read fall back to
2296 // absorbing — which is the same resurrection by another door.
2297 s.mark_absorbed(pack_id, extent);
2298 } else {
2299 s.note(pack_id, extent);
2300 requeue.push(PendingPack {
2301 pack_id,
2302 offset: extent.0,
2303 len: extent.1,
2304 });
2305 }
2306 }
2307 Ok(requeue)
2308 }
2309
2310 /// **The indexer's half of the split**, and not one of the twelve.
2311 ///
2312 /// Reads each pending pack's verbatim bytes back out of the archive,
2313 /// resolves them to oids ([`crate::resolve`]), appends the rows to the
2314 /// `objects` table and re-folds the derived tables. This is what §13.9-12
2315 /// puts behind the channel; it runs after the ack, never before one.
2316 ///
2317 /// Two callers, one function: the account indexer's worker (through
2318 /// [`ObjectAbsorb`]) and a read that arrived before that worker did. The
2319 /// second is §13.12's fallback in its literal form — *slower, never wrong*.
2320 fn absorb_pending(&self) -> Result<usize> {
2321 // Nothing to do, and — this is the point — **no gate taken**: the common
2322 // case is a store whose drain has kept up, and it must not queue behind
2323 // one that is running.
2324 if self.unindexed_packs() == 0 {
2325 return Ok(0);
2326 }
2327 let _gate = self
2328 .gate
2329 .lock()
2330 .map_err(|_| anyhow!("absorb gate poisoned"))?;
2331 let jobs: Vec<PendingPack> = {
2332 let s = self
2333 .packs
2334 .lock()
2335 .map_err(|_| anyhow!("pack state poisoned"))?;
2336 s.pending()
2337 };
2338 let mut absorbed = 0usize;
2339 for job in &jobs {
2340 self.absorb_gated(job)?;
2341 absorbed += 1;
2342 }
2343 Ok(absorbed)
2344 }
2345
2346 /// One pack, **with [`Absorber::gate`] already held**.
2347 ///
2348 /// The order at the end is the correctness argument: the rows are in the
2349 /// index *before* the bit is cleared. A read that finds the bit still clear
2350 /// blocks on the gate and then finds the rows; a read that finds it set finds
2351 /// the rows too. There is no interleaving in which a durable object answers
2352 /// "absent" — which during negotiation would make a client withhold objects
2353 /// and lose data.
2354 fn absorb_gated(&self, job: &PendingPack) -> Result<()> {
2355 {
2356 let s = self
2357 .packs
2358 .lock()
2359 .map_err(|_| anyhow!("pack state poisoned"))?;
2360 if s.is_absorbed(job.pack_id) {
2361 // Already in, by whoever reached it first. Re-resolving is now
2362 // *correct* — the exploded table is keyed by oid and the graph is
2363 // folded from it rather than pushed at, so a second absorb of one
2364 // pack changes nothing (that idempotence is what makes dropping
2365 // the table and re-queueing every pack a safe operation). It is
2366 // still a whole pack re-read, re-inflated and re-hashed for
2367 // nothing, so it is skipped.
2368 return Ok(());
2369 }
2370 }
2371 self.absorb_one(job).map_err(|e| {
2372 // The entry stays pending by construction — it was never removed —
2373 // so the bytes are durable, the pack is still un-indexed, and reads
2374 // keep falling back rather than answering absent.
2375 e.context(format!(
2376 "absorbing pack {} at ({}, {}) — its bytes are durable and it stays \
2377 un-indexed; reads will keep falling back rather than answer absent",
2378 job.pack_id, job.offset, job.len
2379 ))
2380 })?;
2381 let mut s = self
2382 .packs
2383 .lock()
2384 .map_err(|_| anyhow!("pack state poisoned"))?;
2385 s.mark_absorbed(job.pack_id, (job.offset, job.len));
2386 Ok(())
2387 }
2388
2389 /// One pack: verbatim bytes in, `objects` rows **and** §14's exploded rows
2390 /// out, from **one** walk and **one** resolve.
2391 ///
2392 /// The exploded rows are not a second pass and not a second resolver (LAW 5):
2393 /// [`crate::resolve::resolve_walked`] hands every object it produces to the
2394 /// sink as it produces it, including the blob payloads it is about to drop,
2395 /// so the side table falls out of the pass that was already being made.
2396 ///
2397 /// Ordering is the correctness argument, and it is the same one the drain
2398 /// makes one level up: **the exploded rows are committed before the fold that
2399 /// reads them**, and the fold runs before this returns, so there is no
2400 /// interleaving in which the graph is folded over a table that does not yet
2401 /// hold this pack's commits.
2402 fn absorb_one(&self, job: &PendingPack) -> Result<()> {
2403 let bytes = self.read_extent(job.offset, job.len)?;
2404 let walked = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
2405 let rows = crate::resolve::resolve_walked(
2406 &bytes,
2407 &walked,
2408 self.hash,
2409 job.offset,
2410 self,
2411 &self.exploded,
2412 )?;
2413 // Down before anything folds over it.
2414 self.exploded.flush()?;
2415
2416 let entries: Vec<crate::index_layout::IndexEntry> =
2417 rows.iter().map(|r| r.index_entry()).collect();
2418 self.objects.append(&entries)?;
2419
2420 {
2421 let mut d = self
2422 .derived
2423 .write()
2424 .map_err(|_| anyhow!("derived poisoned"))?;
2425 // A reopen already listed every acked extent (`adopt_journal`), so
2426 // this is the first absorb of a pack pushed in *this* process — or a
2427 // re-queued one that is already listed. `find` is what reads this
2428 // list, so a duplicate would not be wrong, only unbounded.
2429 if !d.packs.contains(&(job.offset, job.len)) {
2430 d.packs.push((job.offset, job.len));
2431 }
2432 }
2433 // The graph and the tree payloads are folded from the exploded table, not
2434 // pushed at from here. That is what makes a re-absorb idempotent — which
2435 // it has to be, because dropping the table re-queues every pack.
2436 self.refold()?;
2437 let _ = self.objects.maybe_rebuild()?;
2438 Ok(())
2439 }
2440
2441 /// Recompute everything derived from the object set: the commit graph, the
2442 /// tree payloads, the ordinal space, the generation numbers, and (lazily) the
2443 /// bitmaps.
2444 ///
2445 /// **The graph is folded from §14's exploded table, not accumulated.** That
2446 /// is the whole fix for the clean-reopen hole: commit and tree payloads are
2447 /// on disk, so this call produces the same graph after a restart as before
2448 /// one, and it is called at the end of [`GitStore::open_with_arms`] for
2449 /// exactly that reason. It also makes a re-absorb idempotent — the old
2450 /// accumulate-into-a-`Vec` shape doubled every commit if a pack was absorbed
2451 /// twice, which is why dropping the table and re-queueing everything was not
2452 /// a safe operation before this.
2453 ///
2454 /// **Generations are stored and must be recomputed in any fold** (§13,
2455 /// decided). A generation number is `1 + max(parents)`, so it is invalidated
2456 /// the moment history's shape changes — which is exactly what absorbing a
2457 /// pack does. [`assign_generations`] is the one writer of that number and it
2458 /// is called here, over the whole graph, rather than incrementally: an
2459 /// incremental update would have to know which descendants moved, and being
2460 /// wrong about that produces an ancestry cutoff that skips real ancestors.
2461 ///
2462 /// # Cost
2463 ///
2464 /// One range scan of the exploded table's commits and one of its trees, per
2465 /// absorbed pack. It is the same order as the `tail_oids` scan below, which
2466 /// this function already made on every absorb, and it reads **no blob
2467 /// payload** — that is what the kind index in
2468 /// [`crate::exploded`] is for.
2469 ///
2470 /// ## And that order is the whole repository, on every push
2471 ///
2472 /// Said plainly, because "the same order as the scan below" reads like a
2473 /// reassurance and is not one: **nothing here is proportional to the push.**
2474 /// Every call rebuilds every tree payload, every commit node, every
2475 /// generation number, a `HashMap<String, u32>` over every oid in the
2476 /// repository and a sorted `Vec<String>` of the same — for a push of one
2477 /// commit exactly as for a push of ten thousand.
2478 ///
2479 /// MEASURED on oden 2026-08-11, release, quiet box (`some avg10` 0.00). The
2480 /// **same** 100-commit increment, pushed onto the same history seeded to
2481 /// four different depths, server CPU for the push:
2482 ///
2483 /// | objects already in the repository | CPU |
2484 /// |---:|---:|
2485 /// | 9 614 | 0.12 s |
2486 /// | 35 350 | 0.23 s |
2487 /// | 103 988 | 0.45 s |
2488 /// | 206 925 | 0.77 s |
2489 ///
2490 /// That is a straight line: **≈88 ms fixed, plus ≈3.3 µs per object already
2491 /// stored**, and the fit predicts the four points to within 20 ms. It is
2492 /// also why a push is slower than the ack path suggests — §13.9 puts the
2493 /// index behind a channel, but `put_refs` checks every ref target with
2494 /// `has`, `has` calls `lookup_one`, and `lookup_one` absorbs pending packs
2495 /// first. The drain is off the *ack* and squarely on the *push*.
2496 ///
2497 /// This is not fixed here and the numbers are recorded so the next attempt
2498 /// starts from them rather than from a guess. The obvious lever — fold
2499 /// incrementally — is the one the paragraph above rules out for
2500 /// generations, so it needs its own argument and its own gate.
2501 fn refold(&self) -> Result<()> {
2502 // Read out of redb *before* taking the write lock: the fold is the long
2503 // part and nothing that reads `derived` should queue behind it.
2504 let commits = self.exploded.of_kind(GitObjectKind::Commit)?;
2505 let trees = self.exploded.of_kind(GitObjectKind::Tree)?;
2506
2507 let mut d = self
2508 .derived
2509 .write()
2510 .map_err(|_| anyhow!("derived poisoned"))?;
2511 d.trees = trees
2512 .into_iter()
2513 .map(|(oid, payload)| (hex::encode(oid), payload))
2514 .collect();
2515 let graph: Vec<CommitNode> = commits
2516 .iter()
2517 .map(|(oid, payload)| {
2518 let h = crate::object::parse_commit(payload);
2519 CommitNode {
2520 oid: hex::encode(oid),
2521 parents: h.parents,
2522 tree: h.tree,
2523 committer_time: h.committer_time,
2524 generation: 0,
2525 }
2526 })
2527 .collect();
2528 d.graph = assign_generations(graph);
2529 // The same commit set `select` refuses a `want` against, in the raw form
2530 // it tests. Folded once here instead of once per request; see the field.
2531 // A row that does not decode makes the whole set `None`, which is the
2532 // decline `select` used to reach by returning early out of its own loop.
2533 d.commit_raw = d
2534 .graph
2535 .iter()
2536 .map(|n| hex::decode(&n.oid).ok())
2537 .collect::<Option<std::collections::HashSet<Vec<u8>>>>()
2538 .map(Arc::new);
2539
2540 // Ordinals: oid-lexicographic rank over every object in the index. Same
2541 // ordering rule the Arrow projection uses, computed here so the bitmaps
2542 // and the ordinals they address are produced by one call.
2543 //
2544 // **Sorted as bytes, then hex-encoded — not hex-encoded and then
2545 // sorted.** The two orders are the same order (hex is monotone over
2546 // fixed-width oids, which is why this is not a behaviour change), and
2547 // taking it in this direction is what lets the raw and the hex form come
2548 // out of ONE sort: `d.oids_raw` is the concatenation of the same
2549 // sequence `d.oids` spells in hex, so ordinal `o` names the same object
2550 // in both by construction rather than by two agreeing computations
2551 // (LAW 5).
2552 let mut raw = self.tail_oids()?;
2553 raw.sort_unstable();
2554 let mut oids_raw = Vec::with_capacity(raw.iter().map(Vec::len).sum());
2555 for oid in &raw {
2556 oids_raw.extend_from_slice(oid);
2557 }
2558 let oids: Vec<String> = raw.iter().map(hex::encode).collect();
2559 d.ordinal = oids
2560 .iter()
2561 .enumerate()
2562 .map(|(i, o)| (o.clone(), i as u32))
2563 .collect();
2564 d.oids = oids;
2565 d.oids_raw = oids_raw;
2566 // The bitmaps are over the ordinal space that just changed.
2567 d.reach = Arc::default();
2568 Ok(())
2569 }
2570
2571 /// Every oid in the index, from the tail — which redb keeps in oid order, so
2572 /// this is already the ordinal order.
2573 fn tail_oids(&self) -> Result<Vec<Vec<u8>>> {
2574 self.objects.oids_in_order()
2575 }
2576
2577 /// The bitmaps, built on first use after a fold and cached until the next.
2578 fn reach_bitmaps(&self) -> Result<Arc<Vec<ReachEntry>>> {
2579 self.reach_bitmaps_with(live_reach_policy(), true)
2580 }
2581
2582 /// [`Absorber::reach_bitmaps`] with the policy **named** and the cache
2583 /// optional.
2584 ///
2585 /// `cache` is false for the differential guard, which builds the table at two
2586 /// very different caps over one store and requires the two to answer
2587 /// identically; letting either poison `d.reach` would make the second arm
2588 /// read the first one's table and the comparison would be of a thing against
2589 /// itself.
2590 fn reach_bitmaps_with(
2591 &self,
2592 policy: crate::reach::ReachPolicy,
2593 cache: bool,
2594 ) -> Result<Arc<Vec<ReachEntry>>> {
2595 {
2596 let d = self
2597 .derived
2598 .read()
2599 .map_err(|_| anyhow!("derived poisoned"))?;
2600 if cache && !d.reach.is_empty() {
2601 // A refcount bump. This was `d.reach.clone()` — one deep copy of
2602 // every commit's roaring bitmap, per call, twice per request.
2603 return Ok(Arc::clone(&d.reach));
2604 }
2605 if d.graph.is_empty() {
2606 return Ok(Arc::default());
2607 }
2608 }
2609 let mut d = self
2610 .derived
2611 .write()
2612 .map_err(|_| anyhow!("derived poisoned"))?;
2613 let facts = crate::reach::ObjectFacts {
2614 ordinal: &d.ordinal,
2615 trees: &d.trees,
2616 oid_len: self.hash.oid_len(),
2617 };
2618 let built = Arc::new(crate::reach::build_reach(&d.graph, &facts, policy));
2619 if cache {
2620 // `Arc::new` once, then a refcount bump into the cache — the second
2621 // full copy this line used to make (`d.reach = built.clone()`) is
2622 // gone too.
2623 d.reach = Arc::clone(&built);
2624 }
2625 Ok(built)
2626 }
2627
2628 /// The commit oid set, raw, as folded. See [`Derived::commit_raw`].
2629 fn commit_oids_raw(&self) -> Result<Option<Arc<std::collections::HashSet<Vec<u8>>>>> {
2630 Ok(self
2631 .derived
2632 .read()
2633 .map_err(|_| anyhow!("derived poisoned"))?
2634 .commit_raw
2635 .clone())
2636 }
2637
2638 // `ordinal_space()` lived here: `d.oids.clone()`, a fresh `Vec<String>` of
2639 // every oid in the store, handed to `reachable_oids` twice per request. It
2640 // is gone rather than left unused — its one caller now borrows `d.oids`
2641 // under the read guard it was already going to take.
2642
2643 /// `pread` an extent out of the verbatim archive, **into uninitialised
2644 /// capacity** — no zero-fill before the read.
2645 ///
2646 /// This used to be `vec![0u8; len]` + `read_exact_at`, and under emit
2647 /// concurrency the zeroing was not a detail: profiled on oden 2026-08-12
2648 /// (gunnar serving 32 concurrent clones, perf on the server pid), ~44 % of
2649 /// serve CPU was memory zeroing/copying — `memset` 18.1 % plus kernel
2650 /// page-zeroing 7.8 % — against ~12 % for one clone, because 32 threads
2651 /// churning pack-scale buffers recycle mimalloc freelist blocks that must
2652 /// each be memset before `pread` immediately overwrites every byte.
2653 ///
2654 /// Soundness: no reference to uninitialised memory is ever formed. The
2655 /// bytes are written through a raw pointer into the `Vec`'s spare
2656 /// capacity, `EINTR` retries, a short read refuses, and `set_len` runs
2657 /// only after every one of `len` bytes has been written.
2658 fn read_extent(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
2659 use std::os::fd::AsRawFd;
2660 let want = usize::try_from(len).context("extent length overflows usize")?;
2661 let mut buf: Vec<u8> = Vec::with_capacity(want);
2662 let fd = self.reader.as_raw_fd();
2663 let mut filled = 0usize;
2664 while filled < want {
2665 let n = unsafe {
2666 libc::pread(
2667 fd,
2668 buf.as_mut_ptr().add(filled).cast(),
2669 want - filled,
2670 i64::try_from(offset + filled as u64).context("extent offset overflows off_t")?,
2671 )
2672 };
2673 if n < 0 {
2674 let err = std::io::Error::last_os_error();
2675 if err.kind() == std::io::ErrorKind::Interrupted {
2676 continue;
2677 }
2678 return Err(err).with_context(|| {
2679 format!("reading ({offset}, {len}) out of {}", self.blobs.display())
2680 });
2681 }
2682 if n == 0 {
2683 bail!(
2684 "short read at ({offset}, {len}) out of {}: {filled} of {want} bytes before \
2685 EOF — the archive is truncated relative to its index",
2686 self.blobs.display()
2687 );
2688 }
2689 filled += n as usize;
2690 }
2691 // Every byte of `want` is now initialised.
2692 unsafe { buf.set_len(want) };
2693 Ok(buf)
2694 }
2695}
2696
2697/// **The drain's object-level ingress**, and the reason a push ends in object
2698/// rows without anybody reading first.
2699///
2700/// The account indexer holds this as an `Arc<dyn ObjectAbsorb>` and calls it once
2701/// per drained job, on its own thread, after the ack. It is the *same* absorb a
2702/// falling-back read takes, gate and all (LAW 5): there is no background copy of
2703/// this logic to drift.
2704impl<S: ObjectIndex> ObjectAbsorb for Absorber<S> {
2705 fn absorb(&self, job: IndexJob) -> Result<()> {
2706 let _gate = self
2707 .gate
2708 .lock()
2709 .map_err(|_| anyhow!("absorb gate poisoned"))?;
2710 self.absorb_gated(&PendingPack::from_job(job))
2711 }
2712}
2713
2714impl<S: ObjectIndex + 'static> GitStore<S> {
2715 /// The push path, for the two `put` halves.
2716 pub(crate) fn push_path(&self) -> &PushPath {
2717 &self.push
2718 }
2719
2720 pub(crate) fn account(&self) -> &str {
2721 &self.account
2722 }
2723
2724 pub(crate) fn ref_log(&self) -> &RefLog {
2725 &self.refs
2726 }
2727
2728 pub(crate) fn ref_gate(&self) -> &Mutex<()> {
2729 &self.ref_gate
2730 }
2731
2732 /// The closure check's index half: a `REF_DELTA`'s base has to exist —
2733 /// **in the store, or in this very pack**.
2734 ///
2735 /// This is the one part of receive-pack that reads `objects.oid` (§13's
2736 /// table), and on the path a real push takes it reads nothing else.
2737 ///
2738 /// # The second half of that sentence used to be missing, and it refused
2739 /// packs git itself writes
2740 ///
2741 /// [`crate::pack_walk::PackWalk::closure`] cannot do better on its own:
2742 /// a walk reports what the bytes say, and the bytes of a `REF_DELTA` name a
2743 /// base by **oid**, which is a fact about resolved content that no walk can
2744 /// know. So every ref base came back on `external_refs` and every one of
2745 /// them was required to be in the store already.
2746 ///
2747 /// A `REF_DELTA` naming a base *inside the same pack* is ordinary and git
2748 /// writes it constantly: `git index-pack --fix-thin` completes a pushed thin
2749 /// pack by **appending the base object to the pack** and leaving the delta
2750 /// naming it by oid. Every such pack was refused with *"this pack deltas
2751 /// against X, which this repository does not have"*, on a store that was
2752 /// being handed a perfectly ordinary, self-contained packfile — the base was
2753 /// entry 1 of 25 in the pack's own `.idx` (`gunnar.multi_pack_serve`,
2754 /// 2026-08-14). The same missing check is `S-003` / gitoxide#2882 in
2755 /// `LookupRefDeltaObjectsIter`; it is the same defect wearing the receiving
2756 /// hat instead of the sending one.
2757 ///
2758 /// # What it costs, and where
2759 ///
2760 /// **Nothing on any push that was going to succeed before.** Every base
2761 /// found in the store short-circuits exactly as it did, and a pack with no
2762 /// ref-delta at all never gets past the first loop.
2763 ///
2764 /// The pack is resolved only when the alternative is *refusing it*, and then
2765 /// the resolve is the cheapest honest answer available: an entry's oid is
2766 /// knowable only by applying its delta chain, which is precisely what
2767 /// [`crate::resolve::resolve_walked`] does — the same resolver the absorb is
2768 /// about to run over the same pack anyway, reused rather than twinned
2769 /// (LAW 5). A push that is genuinely thin against a base nobody has still
2770 /// fails, one resolve later, and still names the oid.
2771 pub(crate) fn external_bases_exist(&self, bytes: &[u8], w: &PackWalk) -> Result<()> {
2772 let c = w.closure();
2773 if !c.broken_offsets.is_empty() {
2774 bail!(
2775 "this pack is corrupt: {} delta base offset(s) do not land on an entry — first {}",
2776 c.broken_offsets.len(),
2777 c.broken_offsets[0]
2778 );
2779 }
2780 // The fast path, unchanged: a base the store already holds is settled
2781 // without reading one byte of the pack.
2782 let mut unheld: Vec<&[u8]> = Vec::new();
2783 for oid in &c.external_refs {
2784 if self.lookup_one(oid)?.is_none() {
2785 unheld.push(oid.as_slice());
2786 }
2787 }
2788 if unheld.is_empty() {
2789 return Ok(());
2790 }
2791
2792 // Only now, and only because the alternative is refusing the push: does
2793 // the pack supply these itself? `resolve_walked` answers with the
2794 // store behind it, so a base that really is external is still found the
2795 // cheap way and only a genuinely absent one fails.
2796 let resolved = crate::resolve::resolve_walked(
2797 bytes,
2798 w,
2799 self.hash_kind(),
2800 0,
2801 &*self.absorber,
2802 &crate::exploded::NoSink,
2803 )
2804 .with_context(|| {
2805 format!(
2806 "this pack deltas against {} object(s) this repository does not have, so it was \
2807 resolved to find out whether the pack carries them itself — first {}",
2808 unheld.len(),
2809 hex::encode(unheld[0])
2810 )
2811 })?;
2812 let in_pack: std::collections::HashSet<&[u8]> =
2813 resolved.iter().map(|r| r.oid.as_slice()).collect();
2814 for oid in unheld {
2815 if !in_pack.contains(oid) {
2816 bail!(
2817 "this pack deltas against {}, which this repository does not have and which \
2818 the pack does not carry either — the push is refused rather than stored with \
2819 a dangling base",
2820 hex::encode(oid)
2821 );
2822 }
2823 }
2824 Ok(())
2825 }
2826
2827 /// One oid, on the **serial** path. See [`lookup_path`].
2828 pub(crate) fn lookup_one(&self, oid: Oid<'_>) -> Result<Option<crate::index_layout::IndexRow>> {
2829 if self.unindexed_packs() > 0 {
2830 self.absorb_pending()?;
2831 }
2832 Ok(self.absorber.objects.lookup(oid))
2833 }
2834
2835 /// The current ref namespace, as the log folds to it.
2836 pub(crate) fn ref_state(&self) -> Result<BTreeMap<String, crate::refs::RefState>> {
2837 self.refs.current()
2838 }
2839
2840 /// Reachability over the graph, as roaring bitmaps, in **this store's**
2841 /// ordinal space.
2842 ///
2843 /// Returns `(want ∪ closure) − (have ∪ closure)` as oid hex. Every object —
2844 /// commit, tree and blob — is in the bitmaps, because
2845 /// [`crate::reach::build_reach`] walks the trees; that is why this can answer
2846 /// with objects rather than only with commits.
2847 ///
2848 /// **Hex is for the maintenance path, not the serving one.** The answer is
2849 /// computed in raw bytes by [`GitStore::reachable_raw_with`] and encoded
2850 /// here, at one `String` per object; every caller that serves a request asks
2851 /// for the raw form instead. See [`Derived::oids_raw`] for what that cost.
2852 pub(crate) fn reachable_oids(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> Result<Vec<String>> {
2853 Ok(self
2854 .reachable_raw_with(want, have, live_reach_policy(), true)?
2855 .iter()
2856 .map(hex::encode)
2857 .collect())
2858 }
2859
2860 /// [`GitStore::reachable_oids`] with the bitmap policy **named** and the
2861 /// cache optional.
2862 ///
2863 /// One body, not two (LAW 5): both are [`GitStore::reachable_raw_with`], so
2864 /// the guarded path and the shipped path are the same code.
2865 #[cfg(test)]
2866 pub(crate) fn reachable_oids_with(
2867 &self,
2868 want: &[Oid<'_>],
2869 have: &[Oid<'_>],
2870 policy: crate::reach::ReachPolicy,
2871 cache: bool,
2872 ) -> Result<Vec<String>> {
2873 Ok(self
2874 .reachable_raw_with(want, have, policy, cache)?
2875 .iter()
2876 .map(hex::encode)
2877 .collect())
2878 }
2879
2880 /// `(want ∪ closure) − (have ∪ closure)`, in **raw oid bytes**, in one
2881 /// allocation.
2882 ///
2883 /// The answer every serving caller wants and the one the bitmaps naturally
2884 /// produce: an ordinal indexes straight into [`Derived::oids_raw`], so
2885 /// nothing is encoded, decoded or allocated per object on the way out.
2886 pub(crate) fn reachable_raw(
2887 &self,
2888 want: &[Oid<'_>],
2889 have: &[Oid<'_>],
2890 ) -> Result<git_storage_trait::OidList> {
2891 self.reachable_raw_with(want, have, live_reach_policy(), true)
2892 }
2893
2894 /// [`GitStore::reachable_raw`] with the bitmap policy **named** and the
2895 /// cache optional.
2896 ///
2897 /// It exists because the answer is defined to be **independent of the
2898 /// policy** — [`crate::reach::accumulate`] covers whatever the table does
2899 /// not — and a property like that can only be tested by driving the policy.
2900 /// A test that could not reach it would be asserting that one code path
2901 /// equals itself. `cache: false` keeps one arm's table from being read by
2902 /// the next, which is the same trap in a different hat.
2903 pub(crate) fn reachable_raw_with(
2904 &self,
2905 want: &[Oid<'_>],
2906 have: &[Oid<'_>],
2907 policy: crate::reach::ReachPolicy,
2908 cache: bool,
2909 ) -> Result<git_storage_trait::OidList> {
2910 if self.unindexed_packs() > 0 {
2911 self.absorb_pending()?;
2912 }
2913 // `reach_bitmaps` may take the derived WRITE lock to build, so it runs
2914 // to completion before the read guard below is taken. Getting that order
2915 // wrong is a self-deadlock, not a slow path.
2916 let bitmaps = self.reach_bitmaps_with(policy, cache)?;
2917 let by_commit: HashMap<&str, &roaring::RoaringBitmap> = bitmaps
2918 .iter()
2919 .map(|e| (e.commit.as_str(), &e.bitmap))
2920 .collect();
2921 // **One read guard, held across the whole answer, and nothing cloned out
2922 // of it.** This used to be `ordinal_space()?` (a `Vec<String>` of every
2923 // oid in the store) plus `.ordinal.clone()` (a `HashMap<String, u32>` of
2924 // the same, again) — ~2 × 12 455 `String` allocations per call and two
2925 // calls per request, for two tables that cannot change between folds.
2926 // Borrowing them is not a lock held longer than the work: the guard is a
2927 // reader, folds take the writer, and a fold that lands mid-answer would
2928 // invalidate the ordinals this loop is translating.
2929 let d = self
2930 .absorber
2931 .derived
2932 .read()
2933 .map_err(|_| anyhow!("derived poisoned"))?;
2934 let space = &d.oids_raw;
2935 let ordinal = &d.ordinal;
2936 let oid_len = self.hash.oid_len();
2937
2938 // The graph by oid, for [`crate::reach::accumulate`]'s commit walk. Built
2939 // per call and not cached: it is one `HashMap` of borrows over rows the
2940 // read guard is already holding, and caching it would be a third
2941 // derivation to keep in step with a fold.
2942 let graph: HashMap<&str, &CommitNode> =
2943 d.graph.iter().map(|n| (n.oid.as_str(), n)).collect();
2944 let facts = crate::reach::ObjectFacts {
2945 ordinal,
2946 trees: &d.trees,
2947 oid_len,
2948 };
2949
2950 let mut union = roaring::RoaringBitmap::new();
2951 for oid in want {
2952 let hex = hex::encode(oid);
2953 match by_commit.get(hex.as_str()) {
2954 // A commit with a bitmap: its whole closure, in one OR.
2955 Some(bm) => union |= *bm,
2956 // **No bitmap. This is the case that used to be a wrong answer.**
2957 //
2958 // It contributed the object itself and nothing else — see
2959 // `crate::reach::accumulate`'s header for what that cost. Now a
2960 // commit the graph holds is WALKED, bounded by the first
2961 // bitmapped commit behind it, and only a `want` the store does
2962 // not hold at all is still refused.
2963 None => {
2964 if graph.contains_key(hex.as_str()) {
2965 crate::reach::accumulate(
2966 hex.as_str(),
2967 &by_commit,
2968 &graph,
2969 &facts,
2970 &mut union,
2971 );
2972 } else if let Some(&o) = ordinal.get(hex.as_str()) {
2973 // Not a commit at all — a tag or a blob asked for
2974 // directly. Just it, as before.
2975 union.insert(o);
2976 } else {
2977 bail!(
2978 "want {hex} is not in this repository — a negotiation must not be \
2979 answered from a partial set"
2980 );
2981 }
2982 }
2983 }
2984 }
2985 let mut had = roaring::RoaringBitmap::new();
2986 for oid in have {
2987 let hex = hex::encode(oid);
2988 if let Some(bm) = by_commit.get(hex.as_str()) {
2989 had |= *bm;
2990 } else if graph.contains_key(hex.as_str()) {
2991 // The same walk on the exclude side, and it must be the same
2992 // walk: an under-counted `have` over-sends (harmless, wasteful)
2993 // but an under-counted `want` under-sends, and answering the two
2994 // sides by different rules is how a `want − have` stops being a
2995 // subtraction of like for like.
2996 crate::reach::accumulate(hex.as_str(), &by_commit, &graph, &facts, &mut had);
2997 } else if let Some(&o) = ordinal.get(hex.as_str()) {
2998 had.insert(o);
2999 }
3000 // An unknown `have` is normal: the client may have objects we do not.
3001 // It contributes nothing, which is the safe direction — we send more,
3002 // never less.
3003 }
3004
3005 let delta = union - had;
3006 // One allocation for the whole answer, sized before the loop. The
3007 // ordinal space is a flat buffer, so each object is a `copy_from_slice`
3008 // of `oid_len` bytes into it and nothing else.
3009 let mut out = git_storage_trait::OidList::with_capacity(delta.len() as usize, oid_len);
3010 for o in delta {
3011 let at = o as usize * oid_len;
3012 let raw = space
3013 .get(at..at + oid_len)
3014 .ok_or_else(|| anyhow!("ordinal {o} is outside this store's ordinal space"))?;
3015 out.push(raw)?;
3016 }
3017 Ok(out)
3018 }
3019
3020 /// Every object reachable from every ref: what a GC must keep.
3021 pub(crate) fn live_set(&self) -> Result<std::collections::HashSet<String>> {
3022 let refs = self.ref_state()?;
3023 let tips: Vec<Vec<u8>> = refs
3024 .values()
3025 .filter_map(|s| s.target.as_deref())
3026 .filter_map(|t| hex::decode(t).ok())
3027 .collect();
3028 if tips.is_empty() {
3029 bail!(
3030 "this repository has no ref pointing at an object, so every object in it would be \
3031 dead. A GC that would delete everything is refused: name a ref first"
3032 );
3033 }
3034 let borrowed: Vec<Oid<'_>> = tips.iter().map(|t| t.as_slice()).collect();
3035 let mut live: std::collections::HashSet<String> =
3036 self.reachable_oids(&borrowed, &[])?.into_iter().collect();
3037 // The peeled targets of annotated tags are reachable too, and a tag
3038 // object is not a commit so it has no bitmap of its own.
3039 for s in refs.values() {
3040 if let Some(p) = &s.peeled {
3041 live.insert(p.clone());
3042 }
3043 if let Some(t) = &s.target {
3044 live.insert(t.clone());
3045 }
3046 }
3047 Ok(live)
3048 }
3049
3050 /// Drop every row the live set does not name. The index is append-only for
3051 /// everything except this.
3052 ///
3053 /// **Both tables, in one call.** §14's exploded table is retired with the
3054 /// `objects` table because its row count against that one's is what says
3055 /// whether it is whole ([`Absorber::adopt_journal`]): dropping from one and
3056 /// not the other would leave every later open re-exploding the entire
3057 /// repository. The count returned is still the index's — that is what a
3058 /// [`GcReport`] means by a dropped row.
3059 pub(crate) fn drop_dead_rows(&self, live: &std::collections::HashSet<String>) -> Result<u64> {
3060 let dropped = self
3061 .absorber
3062 .objects
3063 .retain(&|oid: &[u8]| live.contains(&hex::encode(oid)))?;
3064 self.absorber
3065 .exploded
3066 .retain(&|oid: &[u8]| live.contains(&hex::encode(oid)))?;
3067 // **Everything folded from those two tables, refolded.** The commit
3068 // graph, the tree payloads, the ordinal space and the bitmaps are all
3069 // derivations of the rows that just went, and a derivation that still
3070 // names a dropped oid is *wrong* rather than merely stale — a `reachable`
3071 // over a graph holding dead commits selects objects the index can no
3072 // longer serve. This is the same [`Absorber::refold`] every absorb and
3073 // every open already calls, over the table this call just changed;
3074 // nothing about how the graph is rebuilt is touched here.
3075 //
3076 // The bitmaps go with it: they address **ordinals**, and dropping rows
3077 // renumbers the ordinal space, so `refold` clears them and the next
3078 // `reach_bitmaps()` builds them over the new numbering. Carrying them
3079 // across would be carrying an index into a different array.
3080 self.absorber.refold()?;
3081 Ok(dropped)
3082 }
3083
3084 /// **The journal's half of a GC: retire every pack whose objects are now all
3085 /// dead, before a single row is dropped.**
3086 ///
3087 /// Returns the offsets retired.
3088 ///
3089 /// # Why this exists
3090 ///
3091 /// §13.12's `indexed` bit is derived on open as *extent in the journal, rows
3092 /// not in the index* ([`Absorber::adopt_journal`]). Dropping every row of a
3093 /// pack produces that state exactly, so without this call the next open
3094 /// re-queues the pack and re-absorbs the objects a GC just decided were dead.
3095 /// A **partly** dead pack keeps rows and was never affected; this is only
3096 /// about the all-dead case.
3097 ///
3098 /// # What it does not do
3099 ///
3100 /// It does not truncate `objects.pack`. The blob file is append-only and a
3101 /// pack in the middle of it cannot be cut out without moving every extent
3102 /// after it — which would invalidate every index row in the archive. The dead
3103 /// pack's bytes stay; what changes is that nothing will ever index them
3104 /// again. Reclaiming those bytes is a rewrite of the blob file, and it is not
3105 /// this function.
3106 ///
3107 /// An arm with no journal ([`FastWriter`](crate::archive_write::FastWriter))
3108 /// has nothing to retire and nothing that re-queues, so it returns empty.
3109 ///
3110 /// # Cost
3111 ///
3112 /// One [`ObjectReadStack::extents_with_rows`] scan plus one batch lookup of
3113 /// the live set. Both are already the shape a GC pays elsewhere, and both run
3114 /// once per `gc()`, not per pack.
3115 pub(crate) fn retire_dead_packs(
3116 &self,
3117 live: &std::collections::HashSet<String>,
3118 ) -> Result<Vec<u64>> {
3119 let Some(journal) = self.arms.writer.journal(&self.blobs) else {
3120 return Ok(Vec::new());
3121 };
3122 if !journal.exists() {
3123 return Ok(Vec::new());
3124 }
3125 let rows = crate::archive_write::read_journal(&journal)?;
3126 let packs = crate::archive_write::acked_packs(&rows);
3127 let already = crate::archive_write::retired_offsets(&rows);
3128 if packs.is_empty() {
3129 return Ok(Vec::new());
3130 }
3131
3132 // Which packs hold any row at all right now. A pack with none is either
3133 // already retired or was never absorbed — neither is this call's
3134 // business, and calling the second one dead would lose it.
3135 let occupied = self.absorber.objects.extents_with_rows(&packs)?;
3136
3137 // Which packs hold a row that survives. Placed by binary search over the
3138 // pack starts, the same way the scan above places a row.
3139 let mut order: Vec<usize> = (0..packs.len()).filter(|&i| packs[i].1 > 0).collect();
3140 order.sort_unstable_by_key(|&i| packs[i].0);
3141 let starts: Vec<u64> = order.iter().map(|&i| packs[i].0).collect();
3142 let live_oids: Vec<Vec<u8>> = live.iter().filter_map(|h| hex::decode(h).ok()).collect();
3143 let borrowed: Vec<&[u8]> = live_oids.iter().map(|o| o.as_slice()).collect();
3144 let mut has_live = vec![false; packs.len()];
3145 for (offset, _) in self
3146 .absorber
3147 .objects
3148 .extents_batch(&borrowed)
3149 .into_iter()
3150 .flatten()
3151 {
3152 let p = starts.partition_point(|&s| s <= offset);
3153 if p == 0 {
3154 continue;
3155 }
3156 let i = order[p - 1];
3157 if offset < packs[i].0 + packs[i].1 {
3158 has_live[i] = true;
3159 }
3160 }
3161
3162 let dead: Vec<u64> = (0..packs.len())
3163 .filter(|&i| occupied[i] && !has_live[i] && !already.contains(&packs[i].0))
3164 .map(|i| packs[i].0)
3165 .collect();
3166 // Durable before the rows go. The one ordering this function has.
3167 crate::archive_write::retire_packs(&journal, &dead)?;
3168 Ok(dead)
3169 }
3170
3171 /// **Write generation 0** at [`archive_path`](Self::archive_path), carrying
3172 /// the verbatim packs this store has acked plus `reserved`.
3173 ///
3174 /// The whole of the decision-making lives in
3175 /// [`seal_generation_zero`](crate::archive_write::seal_generation_zero); this
3176 /// is the three paths it needs, taken off the same fields the push path and
3177 /// the GC read, so a seal cannot address a different blob file or a different
3178 /// journal from the one that acked the packs.
3179 pub(crate) fn seal_archive(
3180 &self,
3181 reserved: Vec<ReservedSection>,
3182 ) -> Result<crate::archive_write::SealReport> {
3183 crate::archive_write::seal_generation_zero(
3184 &self.blobs,
3185 self.arms.writer.journal(&self.blobs).as_deref(),
3186 &self.archive,
3187 reserved,
3188 )
3189 }
3190
3191 /// The sections a sealed archive carries: the ref log, the commit graph and
3192 /// the bitmaps.
3193 pub(crate) fn reserved_sections(&self) -> Result<Vec<ReservedSection>> {
3194 use znippy_common::{GUNNAR_GRAPH_MODULE, GUNNAR_REACH_MODULE};
3195 let mut out = vec![self.refs.seal_section()?];
3196 let d = self
3197 .absorber
3198 .derived
3199 .read()
3200 .map_err(|_| anyhow!("derived poisoned"))?;
3201 if !d.graph.is_empty() {
3202 out.push(ReservedSection::arrow(
3203 GUNNAR_GRAPH_MODULE,
3204 crate::graph::graph_schema(),
3205 vec![crate::graph::build_graph_batch(&d.graph)?],
3206 ));
3207 }
3208 drop(d);
3209 let reach = self.reach_bitmaps()?;
3210 if !reach.is_empty() {
3211 out.push(ReservedSection::arrow(
3212 GUNNAR_REACH_MODULE,
3213 crate::reach::reach_schema(),
3214 vec![crate::reach::build_reach_batch(&reach)?],
3215 ));
3216 }
3217 Ok(out)
3218 }
3219}
3220
3221// ── the runtime selector ──────────────────────────────────────────────────────
3222
3223/// **A [`GitStore`] whose index arm was chosen at run time.**
3224///
3225/// The [`ObjectIndex`] arm is a *type*, so a value can only choose it by naming
3226/// every monomorphisation. This enum is that list, and it is an enum rather than
3227/// a `Box<dyn GitOps>` for two reasons:
3228///
3229/// * a boxed trait object can only offer the twelve, and the arms differ in
3230/// things the twelve deliberately do not expose — how many Arrow bytes the
3231/// projection actually materialised, which writer is on the ack path. A bench
3232/// or an operator dump needs those, and a guard needs them to assert that the
3233/// selector selected **on applied output** rather than on a label;
3234/// * there is no allocation and no vtable: the `match` is one predictable branch
3235/// at the top of a call, and everything under it — the `stree` probe, the
3236/// Arrow gather, the redb tail — is the same fully static code
3237/// [`GitStore::<S>::open_with_arms`](GitStore::open_with_arms) produces,
3238/// because `S` is known inside each arm.
3239///
3240/// # What it costs, stated rather than hidden
3241///
3242/// One branch per [`GitOps`] call — **not per object**. `extents(&[1000 oids])`
3243/// is one branch and a thousand monomorphised lookups behind it. The binary
3244/// carries three copies of the store, which is the price of picking a type at
3245/// run time and there is no version of this that does not pay it.
3246///
3247/// A caller that knows its arm at compile time should **not** come through here:
3248/// [`GitStore::open`] and [`GitStore::open_with_arms`] hand back a concrete
3249/// store with no branch at all, and that includes every caller that wants the
3250/// default.
3251pub enum SelectedStore {
3252 OneTableFourColumns(GitStore<OneTableFourColumns>),
3253 FourTables(GitStore<crate::index_layout::FourTables>),
3254 PackedPayload(GitStore<crate::index_layout::PackedPayload>),
3255}
3256
3257/// One expression, evaluated against whichever concrete store this is. The
3258/// delegation below is generated from it so there is no second copy of any
3259/// method body (LAW 5).
3260macro_rules! on_arm {
3261 ($self:ident, $s:ident => $body:expr) => {
3262 match $self {
3263 SelectedStore::OneTableFourColumns($s) => $body,
3264 SelectedStore::FourTables($s) => $body,
3265 SelectedStore::PackedPayload($s) => $body,
3266 }
3267 };
3268}
3269
3270impl SelectedStore {
3271 /// The arms this store was built with.
3272 pub fn arms(&self) -> StoreConfig {
3273 on_arm!(self, s => s.arms())
3274 }
3275
3276 /// The selected writer's name, as it goes on a bench row.
3277 pub fn writer_name(&self) -> &'static str {
3278 on_arm!(self, s => s.writer_name())
3279 }
3280
3281 /// What the selected writer's `append` promises.
3282 pub fn writer_durability(&self) -> &'static str {
3283 on_arm!(self, s => s.writer_durability())
3284 }
3285
3286 /// **Arrow IPC bytes the selected index arm actually materialised.**
3287 ///
3288 /// The applied output that distinguishes one layout from another: a packed
3289 /// 25-byte column, four columns in one section and four independent
3290 /// sections are three different numbers for the same objects. This is what
3291 /// a guard reads to prove the selector selected, because a *name* would
3292 /// prove only that a name was copied.
3293 pub fn index_ipc_bytes(&self) -> usize {
3294 on_arm!(self, s => s.index().ipc_bytes())
3295 }
3296
3297 /// **The name the built projection reports about itself** —
3298 /// `OneTableFourColumns`, `FourTables` or `PackedPayload`.
3299 ///
3300 /// The other half of the applied-output pair [`index_ipc_bytes`] starts:
3301 /// the byte count separates the arms once objects are in the store, and
3302 /// this separates them from the instant it is built, including on an empty
3303 /// store where all three materialise nothing. It is the value
3304 /// [`crate::arms::IndexArm::projection_name`] exists to be compared with —
3305 /// what the layout calls itself, never what the selector asked for — so a
3306 /// server logging it is stating what it built rather than repeating its own
3307 /// environment back.
3308 ///
3309 /// [`index_ipc_bytes`]: SelectedStore::index_ipc_bytes
3310 pub fn index_name(&self) -> &'static str {
3311 on_arm!(self, s => s.index().projection_name())
3312 }
3313
3314 /// The oids a pushed extent introduced — [`GitStore::oids_in_extent`],
3315 /// delegated.
3316 ///
3317 /// Not one of the twelve (a pack's extent is a znippy concept), so it does
3318 /// not arrive with the [`GitOps`] impl below and has to be forwarded by
3319 /// hand. A server that reports the objects a push introduced needs it, and
3320 /// a server that picked its index arm at run time still needs it.
3321 pub fn oids_in_extent(&self, extent: Extent) -> Result<Vec<Vec<u8>>> {
3322 on_arm!(self, s => s.oids_in_extent(extent))
3323 }
3324
3325 /// Objects in the `objects` table.
3326 pub fn object_count(&self) -> usize {
3327 on_arm!(self, s => s.object_count())
3328 }
3329
3330 /// Block until the background drain has absorbed everything pushed.
3331 pub fn wait_indexed(&self) {
3332 on_arm!(self, s => s.wait_indexed())
3333 }
3334
3335 /// Rebuild the projection, so the layout under test is what answers rather
3336 /// than the redb tail.
3337 pub fn rebuild_projection(&self) -> Result<()> {
3338 on_arm!(self, s => s.index().rebuild())
3339 }
3340
3341 /// The twelfth method — inherent, not on [`GitOps`], because it returns
3342 /// Arrow `ReservedSection`s a gix backend has no analog for.
3343 pub fn seal(&self) -> Result<Vec<ReservedSection>> {
3344 on_arm!(self, s => s.seal())
3345 }
3346}
3347
3348impl GitOps for SelectedStore {
3349 fn put(&self, pack: &[u8], refs: &[RefUpdate]) -> Result<TxId> {
3350 on_arm!(self, s => s.put(pack, refs))
3351 }
3352 fn put_pack(&self, bytes: &[u8]) -> Result<TxId> {
3353 on_arm!(self, s => s.put_pack(bytes))
3354 }
3355 fn put_refs(&self, updates: &[RefUpdate]) -> Result<TxId> {
3356 on_arm!(self, s => s.put_refs(updates))
3357 }
3358 fn get(&self, oid: Oid<'_>) -> Result<Option<Stored>> {
3359 on_arm!(self, s => s.get(oid))
3360 }
3361 fn has(&self, oid: Oid<'_>) -> Result<bool> {
3362 on_arm!(self, s => s.has(oid))
3363 }
3364 fn size(&self, oid: Oid<'_>) -> Result<Option<u64>> {
3365 on_arm!(self, s => s.size(oid))
3366 }
3367 fn extents(&self, oids: &[Oid<'_>]) -> Result<Vec<Option<Extent>>> {
3368 on_arm!(self, s => s.extents(oids))
3369 }
3370 fn refs(&self) -> Result<Vec<RefRow>> {
3371 on_arm!(self, s => s.refs())
3372 }
3373 fn update_ref(&self, name: &str, old: Option<Oid<'_>>, new: Option<Oid<'_>>) -> Result<TxId> {
3374 on_arm!(self, s => s.update_ref(name, old, new))
3375 }
3376 fn put_refs_cas(&self, edits: &[git_storage_trait::RefCas<'_>]) -> Result<TxId> {
3377 on_arm!(self, s => s.put_refs_cas(edits))
3378 }
3379 fn reachable(&self, want: &[Oid<'_>], have: &[Oid<'_>]) -> Result<Vec<Vec<u8>>> {
3380 on_arm!(self, s => s.reachable(want, have))
3381 }
3382 fn gc(&self) -> Result<GcReport> {
3383 on_arm!(self, s => s.gc())
3384 }
3385}
3386
3387/// The reading contract, delegated the same way [`GitOps`] is.
3388///
3389/// Every method is one predictable branch over a monomorphised body — the same
3390/// arrangement, and the same cost argument, [`SelectedStore`] makes for the
3391/// eleven. It is here rather than in [`crate::serve`] only because the macro
3392/// that generates it lives in this module.
3393impl crate::serve::GitServe for SelectedStore {
3394 fn read(&self, oid: Oid<'_>) -> Result<Option<(crate::index_layout::ObjType, Vec<u8>)>> {
3395 on_arm!(self, s => crate::serve::GitServe::read(s, oid))
3396 }
3397 fn header(&self, oid: Oid<'_>) -> Result<Option<(crate::index_layout::ObjType, u64)>> {
3398 on_arm!(self, s => crate::serve::GitServe::header(s, oid))
3399 }
3400 fn sizes(&self, oids: &[Oid<'_>]) -> Result<Vec<Option<u64>>> {
3401 on_arm!(self, s => crate::serve::GitServe::sizes(s, oids))
3402 }
3403 fn head(&self) -> Result<Option<RefRow>> {
3404 on_arm!(self, s => crate::serve::GitServe::head(s))
3405 }
3406 fn set_head(&self, target: &str) -> Result<TxId> {
3407 on_arm!(self, s => crate::serve::GitServe::set_head(s, target))
3408 }
3409 /// `objects` is the **exact set to emit**, not tips to close over — the
3410 /// contract's word, and this forwarder uses it so that the name cannot drift
3411 /// back into `want` one arm at a time. See [`crate::serve::GitServe::emit_pack`].
3412 fn emit_pack(
3413 &self,
3414 objects: &[Oid<'_>],
3415 have: &[Oid<'_>],
3416 caps: &git_storage_trait::Caps,
3417 out: &mut dyn std::io::Write,
3418 ) -> Result<git_storage_trait::PackStats> {
3419 on_arm!(self, s => crate::serve::GitServe::emit_pack(s, objects, have, caps, out))
3420 }
3421 fn select(
3422 &self,
3423 want: &[Oid<'_>],
3424 have: &[Oid<'_>],
3425 ) -> Result<Option<git_storage_trait::ReachSet>> {
3426 on_arm!(self, s => crate::serve::GitServe::select(s, want, have))
3427 }
3428}
3429
3430/// **Open a store with all three arms chosen at run time.**
3431///
3432/// The one place [`crate::arms::IndexArm`] — a value — is turned into `S` — a
3433/// type. See [`SelectedStore`] for what the resulting handle costs.
3434pub fn open_selected(
3435 root: &Path,
3436 account: &str,
3437 hash: GitHashKind,
3438 arms: StoreConfig,
3439) -> Result<SelectedStore> {
3440 use crate::arms::IndexArm;
3441 use crate::index_layout::{FourTables, PackedPayload};
3442 Ok(match arms.index {
3443 IndexArm::OneTableFourColumns => {
3444 SelectedStore::OneTableFourColumns(GitStore::<OneTableFourColumns>::open_with_arms(
3445 root, account, hash, arms,
3446 )?)
3447 }
3448 IndexArm::FourTables => SelectedStore::FourTables(GitStore::<FourTables>::open_with_arms(
3449 root, account, hash, arms,
3450 )?),
3451 IndexArm::PackedPayload => SelectedStore::PackedPayload(
3452 GitStore::<PackedPayload>::open_with_arms(root, account, hash, arms)?,
3453 ),
3454 })
3455}
3456
3457/// **The env-driven front door**: read the three variables **once**, here, and
3458/// build the store they name.
3459///
3460/// This is the whole runtime selector a server or a bench needs — one call, one
3461/// triple of `getenv`s, and from then on the store holds built objects and reads
3462/// nothing. There is deliberately no per-operation lookup of any kind: a
3463/// `std::env::var` inside a serving loop would be one syscall per object served,
3464/// which is a real defect that was found and fixed in gunnar the day before this
3465/// was written. [`crate::arms::env_reads`] counts every read this crate makes so
3466/// that "once, at construction" is an assertion rather than a claim.
3467///
3468/// An unset variable takes the shipping default, so a process with a clean
3469/// environment gets exactly [`GitStore::open`]'s combination.
3470pub fn open_from_env(root: &Path, account: &str, hash: GitHashKind) -> Result<SelectedStore> {
3471 let arms = StoreConfig::from_env()?;
3472 open_selected(root, account, hash, arms)
3473}
3474
3475impl<S: ObjectIndex> Absorber<S> {
3476 /// **One object's resolved content, and §14's table is what answers.**
3477 ///
3478 /// Two paths, and they are not interchangeable even though they return
3479 /// identical bytes:
3480 ///
3481 /// 1. **the exploded table** — one point lookup, no inflate, no delta chain;
3482 /// 2. **re-derived from the verbatim truth** — find which pack holds the
3483 /// object, read the whole pack back, resolve it. Correct and slow, and it
3484 /// exists because the table is droppable: a store whose table has been
3485 /// deleted, or whose re-explosion has not caught up, must still answer.
3486 ///
3487 /// The counters are the only way to tell them apart. Identical bytes come
3488 /// back either way, so a byte comparison proves nothing about which path ran
3489 /// — that is the identity-value trap in its exact form, and it is why
3490 /// [`crate::exploded::ExplodedStats::rederived`] is bumped here rather than
3491 /// inferred.
3492 ///
3493 /// On the [`Absorber`] rather than on [`GitStore`] because that is where the
3494 /// absorb runs: the background worker has no store handle, and a base lookup
3495 /// that needed one would put the whole store behind the drain.
3496 fn resolved(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
3497 // (1) §14's table.
3498 if let Some(hit) = self.exploded.content(oid)? {
3499 return Ok(Some(hit));
3500 }
3501 // (2) the verbatim truth, re-derived.
3502 let Some(row) = self.objects.lookup(oid) else {
3503 return Ok(None);
3504 };
3505 let found = {
3506 let d = self
3507 .derived
3508 .read()
3509 .map_err(|_| anyhow!("derived poisoned"))?;
3510 d.packs
3511 .iter()
3512 .find(|(o, l)| row.offset >= *o && row.offset < o + l)
3513 .copied()
3514 };
3515 let Some((pack_offset, pack_len)) = found else {
3516 return Ok(None);
3517 };
3518 self.exploded.note_rederived();
3519 let bytes = self.read_extent(pack_offset, pack_len)?;
3520 let walked = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
3521 // Through the sink, not through `Resolved::payload`: the resolver drops a
3522 // blob payload as soon as nothing in the pack deltas against it, so
3523 // reading that field answered `None` for most blobs — the commonest
3524 // object there is. The sink sees it before it is dropped.
3525 let capture = crate::exploded::CaptureOne::new(oid);
3526 crate::resolve::resolve_walked(
3527 &bytes,
3528 &walked,
3529 self.hash,
3530 pack_offset,
3531 &crate::resolve::NoBases,
3532 &capture,
3533 )?;
3534 Ok(capture.take())
3535 }
3536}
3537
3538/// A thin pack's external delta base, answered from §14's exploded table.
3539///
3540/// Before that table existed this had to read and re-resolve a whole pack per
3541/// base, and it deliberately cached nothing — because a cache here *would* have
3542/// been the undecided table. It is decided now (eager, 2026-08-08), so the cache
3543/// is the table and this is one point lookup.
3544///
3545/// **The refusal is unchanged**: a base this cannot produce still comes back
3546/// `None`, and [`crate::resolve::ResolveError::missing_base`] refuses the pack by
3547/// name rather than half-resolving it.
3548impl<S: ObjectIndex> BaseSource for Absorber<S> {
3549 fn content(&self, oid: &[u8]) -> Option<(GitObjectKind, Vec<u8>)> {
3550 self.resolved(oid).ok().flatten()
3551 }
3552}
3553
3554#[cfg(test)]
3555mod tests {
3556 use super::*;
3557 use crate::archive_write::{read_journal, SafeWriter};
3558 use crate::object::GitObjectKind;
3559 use crate::store::tests::{one_blob_pack, real_pack, tmpdir};
3560 use std::collections::HashSet;
3561 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomicOrdering};
3562 use std::time::{Duration, Instant};
3563 use znippy_zoomies::background::Job;
3564 use znippy_zoomies::gatling_forkjoin::gatling_for_each;
3565
3566 fn loadavg() -> String {
3567 std::fs::read_to_string("/proc/loadavg")
3568 .unwrap_or_default()
3569 .split_whitespace()
3570 .take(3)
3571 .collect::<Vec<_>>()
3572 .join(" ")
3573 }
3574
3575 /// **The ordinal space has two spellings and they must be ONE sequence.**
3576 ///
3577 /// `Derived::oids` (hex) is what the ordinal map, the commit graph and the
3578 /// tree payloads are keyed by; `Derived::oids_raw` (flat bytes) is what
3579 /// every serving answer is built out of, because it is what the bitmaps'
3580 /// ordinals index into. They are folded from one sorted vector so they
3581 /// cannot disagree — this asserts the property that argument claims, at
3582 /// every ordinal, because an off-by-one between them is a server that names
3583 /// the wrong object in a pack and no exit code can see it.
3584 ///
3585 /// Seen RED by folding the raw form in reverse: 3 341 of 3 341 ordinals
3586 /// named a different object, and the closure comparison below disagreed as
3587 /// well.
3588 #[test]
3589 fn the_raw_ordinal_space_and_the_hex_one_are_the_same_sequence() {
3590 let dir = tmpdir("ordinal-space-two-spellings");
3591 let (pack, rows) = real_pack();
3592 let store = GitStore::open(&dir, "rickard").unwrap();
3593 store.put_pack(&pack).unwrap();
3594 store.wait_indexed();
3595
3596 let oid_len = store.hash_kind().oid_len();
3597 let (hex, raw) = {
3598 let d = store.absorber.derived.read().unwrap();
3599 (d.oids.clone(), d.oids_raw.clone())
3600 };
3601 assert!(
3602 !hex.is_empty(),
3603 "the fixture pack ({} entries) folded to an EMPTY ordinal space, so every \
3604 comparison below would hold vacuously",
3605 rows.len()
3606 );
3607 assert_eq!(
3608 raw.len(),
3609 hex.len() * oid_len,
3610 "the raw ordinal space is {} bytes for {} oids of {oid_len} bytes",
3611 raw.len(),
3612 hex.len()
3613 );
3614 let mut checked = 0usize;
3615 for (o, h) in hex.iter().enumerate() {
3616 assert_eq!(
3617 &raw[o * oid_len..(o + 1) * oid_len],
3618 hex::decode(h).unwrap().as_slice(),
3619 "ordinal {o} is {h} in the hex space and something else in the raw one"
3620 );
3621 checked += 1;
3622 }
3623 assert_eq!(checked, hex.len());
3624
3625 // And the answer built out of each: the raw closure IS the hex closure,
3626 // for a real tip rather than for an empty request.
3627 let tip = store
3628 .graph_snapshot()
3629 .into_iter()
3630 .max_by_key(|c| c.generation)
3631 .expect("a graph");
3632 let tip_raw = hex::decode(&tip.oid).unwrap();
3633 let flat = store.reachable_raw(&[&tip_raw], &[]).unwrap();
3634 let hexed = store.reachable_oids(&[&tip_raw], &[]).unwrap();
3635 assert!(
3636 flat.len() > 1,
3637 "the tip's closure is {} object(s) — too small to tell an ordering bug from a \
3638 coincidence",
3639 flat.len()
3640 );
3641 assert_eq!(
3642 flat.iter().map(hex::encode).collect::<Vec<_>>(),
3643 hexed,
3644 "the serving answer and the maintenance answer name different objects"
3645 );
3646 assert!(flat.contains(&tip_raw), "a closure without its own tip");
3647 eprintln!(
3648 "load {}; {} ordinals agree in both spellings, tip closure {} objects",
3649 loadavg(),
3650 checked,
3651 flat.len()
3652 );
3653 }
3654
3655 /// Distinct commits in a resolved pack — what the graph must hold exactly
3656 /// once.
3657 fn commits_in(rows: &[crate::resolve::Resolved]) -> usize {
3658 rows.iter()
3659 .filter(|r| r.kind == GitObjectKind::Commit)
3660 .map(|r| r.oid.clone())
3661 .collect::<HashSet<Vec<u8>>>()
3662 .len()
3663 }
3664
3665 /// **A push ends in object rows, and no read asked for them.**
3666 ///
3667 /// This is the seam this file wires: before it, `put` produced a *pack* row
3668 /// on the account indexer's channel and the object rows appeared only when
3669 /// somebody read (or called `absorb_pending` by hand). Here nothing is read
3670 /// between `put_pack` and `wait_indexed` — the only thing that could have
3671 /// built the rows is the drain.
3672 ///
3673 /// Asserted on applied output: one row per pack entry, every oid resolving
3674 /// out of the index, and the pack's indexed bit up. `absorb_failures` is
3675 /// asserted too, because a drain that absorbed nothing and recorded the
3676 /// failure would otherwise look like a drain that was never called.
3677 ///
3678 /// Seen RED by making the drain skip the absorber
3679 /// (`match absorber.as_deref()` → `match None::<&dyn ObjectAbsorb>` in
3680 /// `indexer::index_worker`): "the drain built no object rows: 0 of 2687".
3681 /// Restored.
3682 ///
3683 /// MEASURED on oden, 2026-08-08, release, three runs at **1-minute loadavg
3684 /// 5.77** (another tenant's work — oden is shared, so this is not an idle
3685 /// box): 2687 objects out of a 5 653 302-byte real pack absorbed in
3686 /// 161–164 ms, **16 421–16 673 object rows/s**, behind an ack of 29.1–29.4 ms.
3687 /// An earlier set at loadavg 16.94 gave 15 528 rows/s, so the figure moves
3688 /// about 7% with the box.
3689 ///
3690 /// Read it as a *floor* rather than as the engine's speed: the drain here is
3691 /// one pack, so it never uses more than one absorb, and the 162 ms covers a
3692 /// full inflate-and-delta-apply of every entry, a SHA-1 per object, the redb
3693 /// append and a whole-tail re-fold. Nothing about it has been optimised and
3694 /// no arm of it has been isolated. The rate is printed with the loadavg of
3695 /// its own run so a later number is never compared against a different
3696 /// machine state by accident.
3697 #[test]
3698 fn a_push_ends_in_object_rows_and_no_read_asked_for_them() {
3699 let dir = tmpdir("drain-rows");
3700 let store = GitStore::open(&dir, "rickard").unwrap();
3701 let (pack, rows) = real_pack();
3702
3703 let t = Instant::now();
3704 let tx = store.put_pack(&pack).unwrap();
3705 let ack = t.elapsed();
3706 let pack_id = tx.pack_id.expect("a pack push assigns an id");
3707
3708 // Nothing is read here. The drain is the only thing that can move.
3709 store.wait_indexed();
3710 let drained = t.elapsed();
3711
3712 assert_eq!(
3713 store.object_count(),
3714 rows.len(),
3715 "the drain built no object rows: {} of {}",
3716 store.object_count(),
3717 rows.len()
3718 );
3719 assert_eq!(store.unindexed_packs(), 0, "the bit is still clear");
3720 assert!(
3721 store.indexer().is_indexed(pack_id),
3722 "the pack's indexed bit is not up after the drain"
3723 );
3724 assert_eq!(
3725 store.indexer().absorb_failures(),
3726 0,
3727 "the drain recorded an absorb failure: {:?}",
3728 store.indexer().last_absorb_error()
3729 );
3730
3731 // Every oid answers **out of the rows**, not out of a fallback: the
3732 // index is asked directly, so nothing can absorb behind this loop.
3733 for (i, r) in rows.iter().enumerate() {
3734 let row = store
3735 .index()
3736 .lookup(&r.oid)
3737 .unwrap_or_else(|| panic!("object {i} {} has no row", hex::encode(&r.oid)));
3738 assert_eq!(
3739 (row.offset, row.len),
3740 (tx.extent.unwrap().0 + r.offset, r.len)
3741 );
3742 assert_eq!(row.uncompressed_size, r.uncompressed_size);
3743 }
3744 assert_eq!(
3745 store.absorb_pending().unwrap(),
3746 0,
3747 "the drain left work for a read to do"
3748 );
3749 assert_eq!(
3750 store.commit_count(),
3751 commits_in(&rows),
3752 "the commit graph does not match the pack"
3753 );
3754
3755 let secs = (drained - ack).as_secs_f64();
3756 eprintln!(
3757 "load {}; {} objects in {} bytes: ack {:.0} µs, drain {:.1} ms, {:.0} object rows/s",
3758 loadavg(),
3759 rows.len(),
3760 pack.len(),
3761 ack.as_secs_f64() * 1e6,
3762 (drained - ack).as_secs_f64() * 1e3,
3763 rows.len() as f64 / secs.max(1e-9),
3764 );
3765 }
3766
3767 // ── the bit across a restart ─────────────────────────────────────────────
3768
3769 /// Where the child of the kill guard finds its fixture and leaves its marker.
3770 const KILL_DIR: &str = "GUNNAR_KILL_MID_ABSORB_DIR";
3771
3772 /// **The other half of [`a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack`]:
3773 /// the process that dies.**
3774 ///
3775 /// It runs as a child of that test and it **aborts on purpose**, which is why
3776 /// it is `#[ignore]`d: `SIGABRT` from inside a test binary is a failing test
3777 /// run, and the point is for the *parent* to see the signal.
3778 ///
3779 /// The kill lands where it has to land — after `put_pack` has returned, so
3780 /// the bytes and the journal row are on the platter, and before any row is in
3781 /// the index, which the held absorb gate makes a **state** rather than a
3782 /// hope. Nothing is simulated: no injected error, no fault flag, no early
3783 /// return. The process is gone.
3784 #[test]
3785 #[ignore = "spawned by a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack; it aborts on purpose"]
3786 fn the_child_that_dies_between_durable_bytes_and_indexed_rows() {
3787 let Ok(dir) = std::env::var(KILL_DIR) else {
3788 // Run by hand with `--ignored` and no fixture: there is nothing to
3789 // kill over, so do nothing rather than abort somebody's test run.
3790 return;
3791 };
3792 let dir = PathBuf::from(dir);
3793 let pack = std::fs::read(dir.join("fixture.pack")).expect("the parent's fixture");
3794
3795 let store = GitStore::open(&dir, "rickard").expect("open");
3796 // The drain is parked in front of its absorb for as long as this lives,
3797 // so no row can land before the abort below.
3798 let _gate = store.hold_absorb_gate();
3799 let tx = store.put_pack(&pack).expect("the ack path");
3800 assert_eq!(tx.pack_id, Some(0), "a fresh archive starts at ordinal 0");
3801 assert_eq!(store.unindexed_packs(), 1, "the pack is not queued");
3802 assert_eq!(store.index().len(), 0, "a row landed before the kill");
3803 std::fs::write(dir.join("ready"), b"durable, not indexed").expect("marker");
3804
3805 // Not `panic!`, not `exit`: the process stops here with a live gate, a
3806 // live worker and a live redb handle, which is what a machine that dies
3807 // mid-absorb leaves behind.
3808 std::process::abort();
3809 }
3810
3811 /// **A store reopened over a pack that was still being absorbed re-queues
3812 /// it, answers correctly the whole way through, and gets its derived tables
3813 /// back.**
3814 ///
3815 /// The interruption is a **real** one: a child process is spawned, pushes the
3816 /// pack, and is killed by `SIGABRT` between "bytes durable" and "rows
3817 /// indexed". The parent then asserts what the kill left on disk — one journal
3818 /// row, the pack verbatim, **zero** index rows — before opening a store over
3819 /// it, so the recovery is being tested against the state it claims to
3820 /// recover from and not against a lucky one.
3821 ///
3822 /// Three things are then asserted, all on applied output:
3823 ///
3824 /// 1. **the pack is re-queued** — the account indexer of the *new* process
3825 /// publishes ordinal 0, which it can only do for a job somebody submitted;
3826 /// 2. **a read is correct throughout** — `has` is asked the instant the store
3827 /// is open, before any drain has been waited for, and must answer `true`;
3828 /// 3. **`commit_count()` and `reachable()` come back** — the derived tables
3829 /// are rebuilt by the same absorb, so the graph holds exactly the pack's
3830 /// commits and a tip's closure is non-empty.
3831 ///
3832 /// Seen RED by making `open_with` re-queue nothing (`absorber
3833 /// .adopt_journal(&acked)?` → an empty `Vec`, which is what this store did
3834 /// before this change): "a durable object read `absent` from a store reopened
3835 /// over the pack that holds it — the crash-recovery bit did not re-queue
3836 /// anything". Restored.
3837 ///
3838 /// Seen RED a second time by inverting the diff in
3839 /// [`Absorber::adopt_journal`] (`if absorbed` → `if !absorbed`), so a pack
3840 /// with **no** rows has its bit set and a pack with rows is re-queued: the
3841 /// same first assertion fires, because a set bit over an empty index is
3842 /// exactly the wrong answer a bloom filter's false positive would have
3843 /// produced — the read stops falling back and says `absent` about an object
3844 /// whose bytes are durable. Restored.
3845 ///
3846 /// **Why the assertions are not immediately after the reopen.** The trap the
3847 /// drain guards already record applies here too, mirrored: right after
3848 /// `GitStore::open` the drain may legitimately have finished the re-queued
3849 /// pack already, so `unindexed_packs() == 1` is not assertable and is not
3850 /// asserted. What *is* assertable at that instant is the answer to a read,
3851 /// which must be `true` whichever side of the drain it lands on — and that is
3852 /// exactly the property §13.12 promises.
3853 ///
3854 /// **And it still holds now that a `gc` can retire a pack's journal row.**
3855 /// The tombstone tells "indexed then garbage-collected" from "never
3856 /// indexed", and this pack is the second: nothing retired it, so it must
3857 /// still be re-queued. Seen RED by making `adopt_journal` treat *every* pack
3858 /// as retired (`retired.contains(&extent.0)` → `true`): "a durable object
3859 /// read `absent` from a store reopened over the pack that holds it — the
3860 /// crash-recovery bit did not re-queue anything". Restored.
3861 #[test]
3862 fn a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack() {
3863 use std::os::unix::process::ExitStatusExt;
3864
3865 let dir = tmpdir("kill-mid-absorb");
3866 let (pack, rows) = real_pack();
3867 std::fs::write(dir.join("fixture.pack"), &pack).unwrap();
3868
3869 let status = std::process::Command::new(std::env::current_exe().unwrap())
3870 .args([
3871 "--exact",
3872 "--ignored",
3873 "--nocapture",
3874 "git_ops::tests::the_child_that_dies_between_durable_bytes_and_indexed_rows",
3875 ])
3876 .env(KILL_DIR, &dir)
3877 .status()
3878 .expect("spawn the child that dies");
3879 assert_eq!(
3880 status.signal(),
3881 Some(6),
3882 "the child did not die by SIGABRT — it {status:?}, so nothing was interrupted"
3883 );
3884 assert!(
3885 dir.join("ready").exists(),
3886 "the child never reached the kill point"
3887 );
3888
3889 // ── what the kill left on disk ───────────────────────────────────────
3890 let blobs = dir.join("objects.pack");
3891 let acked = read_journal(&SafeWriter::journal_path(&blobs)).unwrap();
3892 assert_eq!(
3893 acked.len(),
3894 1,
3895 "the pack's extent is not durable: {acked:?}"
3896 );
3897 assert_eq!(acked[0].1, pack.len() as u64);
3898 let on_disk = std::fs::read(&blobs).unwrap();
3899 assert_eq!(
3900 &on_disk[acked[0].0 as usize..(acked[0].0 + acked[0].1) as usize],
3901 &pack[..],
3902 "the verbatim bytes did not survive the kill"
3903 );
3904 {
3905 // Opened and dropped before the store takes redb's file lock.
3906 let tail = ObjectReadStack::<OneTableFourColumns>::open(
3907 &dir.join("objects.tail"),
3908 RebuildTriggers::default(),
3909 crate::arms::DEFAULT_REDB_CACHE_BYTES,
3910 )
3911 .unwrap();
3912 assert_eq!(
3913 tail.len(),
3914 0,
3915 "the kill landed after the rows were indexed, not between the bytes and the rows \
3916 — this guard would then be testing nothing"
3917 );
3918 }
3919
3920 // ── the reopen ───────────────────────────────────────────────────────
3921 let store = GitStore::open(&dir, "rickard").unwrap();
3922 assert!(
3923 store.has(&rows[0].oid).unwrap(),
3924 "a durable object read `absent` from a store reopened over the pack that holds it — \
3925 the crash-recovery bit did not re-queue anything"
3926 );
3927
3928 store.wait_indexed();
3929 assert_eq!(
3930 store.object_count(),
3931 rows.len(),
3932 "the re-queued pack's objects never reached the index — left: {}, right: {}",
3933 store.object_count(),
3934 rows.len()
3935 );
3936 assert!(
3937 store.indexer().is_indexed(0),
3938 "the re-queued pack's indexed bit never went up in the new process"
3939 );
3940 assert_eq!(store.unindexed_packs(), 0);
3941 assert_eq!(store.absorb_pending().unwrap(), 0);
3942 for r in rows.iter().take(512) {
3943 let row = store
3944 .index()
3945 .lookup(&r.oid)
3946 .unwrap_or_else(|| panic!("{} has no row after recovery", hex::encode(&r.oid)));
3947 assert_eq!((row.offset, row.len), (acked[0].0 + r.offset, r.len));
3948 }
3949
3950 // ── the derived tables, which were the second symptom ────────────────
3951 assert_eq!(
3952 store.commit_count(),
3953 commits_in(&rows),
3954 "the commit graph did not come back: {} of {}",
3955 store.commit_count(),
3956 commits_in(&rows)
3957 );
3958 let tip = store
3959 .graph_snapshot()
3960 .into_iter()
3961 .max_by_key(|c| c.generation)
3962 .expect("a graph");
3963 let tip_raw = hex::decode(&tip.oid).unwrap();
3964 let closure = store.reachable(&[&tip_raw], &[]).unwrap();
3965 assert!(
3966 !closure.is_empty(),
3967 "reachable() came back empty after the reopen"
3968 );
3969 assert!(closure.contains(&tip_raw));
3970 eprintln!(
3971 "load {}; killed mid-absorb: {} journal extent(s) recovered {} object rows, {} \
3972 commits, closure of the tip {} objects",
3973 loadavg(),
3974 acked.len(),
3975 store.object_count(),
3976 store.commit_count(),
3977 closure.len(),
3978 );
3979 }
3980
3981 /// **A clean reopen re-queues nothing, and the next push gets a fresh
3982 /// ordinal.**
3983 ///
3984 /// The other side of the diff. A store whose pack was fully absorbed before
3985 /// it closed must come back with the bit **set**: re-absorbing would be a
3986 /// whole pack re-resolved for nothing, and — because the commit graph is a
3987 /// `Vec` — it is the same doubling the drain's own guard exists to prevent.
3988 ///
3989 /// "Re-queued nothing" is asserted deterministically rather than by a timing
3990 /// window: the account indexer in this process publishes a pack only for a
3991 /// job somebody submitted, so `is_indexed(0)` being **false** after
3992 /// `wait_indexed` is proof that open submitted none.
3993 ///
3994 /// The second half is the ordinal. Pack ids are the journal's row numbers, so
3995 /// a reopened store must resume after them. Asserted on applied output rather
3996 /// than on the id alone: the newly pushed pack's object has to be findable,
3997 /// and it is exactly what goes missing when a fresh pack is handed an
3998 /// absorbed pack's ordinal — `queue` skips it, the drain skips it, and its
3999 /// durable bytes are never indexed by anybody.
4000 ///
4001 /// Seen RED by making `Absorber::adopt_journal` treat every journal extent as
4002 /// unabsorbed (`let has_rows = vec![false; journal.len()]`): "a fully
4003 /// absorbed pack was re-queued on open — the diff is not exact".
4004 ///
4005 /// Seen RED a second time by `AtomicU64::new(packs_already_acked(...)?)` →
4006 /// `AtomicU64::new(0)` in `PushPath::with_absorber`: "a reopened store handed
4007 /// a fresh pack the ordinal of an absorbed one — left: Some(0), right:
4008 /// Some(1)". With that same edit **and the id assertion deleted**, the
4009 /// applied-output one goes too — "the second push's object never reached the
4010 /// index — its ordinal collided with an absorbed pack's, so every writer
4011 /// skipped it" — which is what says this guard is about the objects and not
4012 /// about a counter. Restored.
4013 #[test]
4014 fn a_clean_reopen_requeues_nothing_and_the_next_push_gets_a_fresh_ordinal() {
4015 let dir = tmpdir("clean-reopen");
4016 let (pack, rows) = real_pack();
4017 let (second, blob_oid) = one_blob_pack(b"a push that arrives after the restart");
4018
4019 {
4020 let store = GitStore::open(&dir, "rickard").unwrap();
4021 store.put_pack(&pack).unwrap();
4022 store.wait_indexed();
4023 assert_eq!(
4024 store.object_count(),
4025 rows.len(),
4026 "the first run did not index"
4027 );
4028 }
4029
4030 let store = GitStore::open(&dir, "rickard").unwrap();
4031 store.wait_indexed();
4032 assert!(
4033 !store.indexer().is_indexed(0),
4034 "a fully absorbed pack was re-queued on open — the diff is not exact"
4035 );
4036 assert_eq!(store.unindexed_packs(), 0);
4037 assert_eq!(
4038 store.absorb_pending().unwrap(),
4039 0,
4040 "the reopen left index work"
4041 );
4042 assert_eq!(
4043 store.object_count(),
4044 rows.len(),
4045 "the reopened store lost rows"
4046 );
4047 assert!(store.has(&rows[0].oid).unwrap());
4048
4049 let tx = store.put_pack(&second).unwrap();
4050 assert_eq!(
4051 tx.pack_id,
4052 Some(1),
4053 "a reopened store handed a fresh pack the ordinal of an absorbed one"
4054 );
4055 store.wait_indexed();
4056 assert!(
4057 store.has(&blob_oid).unwrap(),
4058 "the second push's object never reached the index — its ordinal collided with an \
4059 absorbed pack's, so every writer skipped it"
4060 );
4061 assert_eq!(store.object_count(), rows.len() + 1);
4062 assert!(store.indexer().is_indexed(1));
4063 // And the journal is a log: both packs' extents are in it, in order.
4064 let acked = read_journal(&SafeWriter::journal_path(store.blobs_path())).unwrap();
4065 assert_eq!(
4066 acked.len(),
4067 2,
4068 "the reopen truncated the journal: {acked:?}"
4069 );
4070 assert_eq!(acked[1], tx.extent.unwrap());
4071 }
4072
4073 /// **What the derive-on-open diff costs at a realistic pack count.**
4074 ///
4075 /// Not a guard — a measurement, with two assertions on it that keep it from
4076 /// measuring the wrong thing: every absorbed pack must come back `true`, and
4077 /// the one pack that has no rows must come back `false`.
4078 ///
4079 /// Both cases are timed because they are different algorithms in practice.
4080 /// The **clean reopen** — every pack absorbed — terminates as soon as each
4081 /// extent has been hit once, which happens after a few rows per pack because
4082 /// the tail is in oid order and an oid says nothing about which pack its
4083 /// object came from. The **crash** case has one pack with no rows at all, and
4084 /// proving an absence means reading the tail to the end.
4085 ///
4086 /// MEASURED on oden 2026-08-08, release, 1000 real pushed packs / 3686 object
4087 /// rows, 1-minute loadavg 2.77: **0.38 ms** clean and 0.39 ms with one pack
4088 /// un-absorbed (0.40 / 0.39 ms at loadavg 13.06, so the figure barely moves
4089 /// with the box). At this row count the two cases are the same, because a
4090 /// 3686-row tail is scanned to the end faster than the early exit saves
4091 /// anything; they diverge with the object count, and the table on
4092 /// [`ObjectReadStack::extents_with_rows`] carries that at a million rows
4093 /// (0.74 ms against 109 ms).
4094 ///
4095 /// The printed line carries the loadavg of its own run, the pack count and
4096 /// the row count, because none of the three is comparable across machines or
4097 /// across a busy box.
4098 #[test]
4099 fn the_derive_on_open_diff_costs_milliseconds_at_a_realistic_pack_count() {
4100 const PACKS: usize = 1000;
4101 let dir = tmpdir("diff-cost");
4102 let (real, rows) = real_pack();
4103
4104 {
4105 let store = GitStore::open(&dir, "rickard").unwrap();
4106 store.put_pack(&real).unwrap();
4107 for i in 0..PACKS - 1 {
4108 let (p, _) = one_blob_pack(format!("pack number {i}").as_bytes());
4109 store.put_pack(&p).unwrap();
4110 }
4111 store.wait_indexed();
4112 assert_eq!(
4113 store.object_count(),
4114 rows.len() + PACKS - 1,
4115 "the fixture did not absorb"
4116 );
4117 }
4118
4119 let acked = read_journal(&SafeWriter::journal_path(&dir.join("objects.pack"))).unwrap();
4120 assert_eq!(acked.len(), PACKS, "one journal row per pack");
4121 // The tail on its own, so what is timed is the diff rather than redb's
4122 // open and the projection build that a store's open also pays for.
4123 let tail = ObjectReadStack::<OneTableFourColumns>::open(
4124 &dir.join("objects.tail"),
4125 RebuildTriggers::default(),
4126 crate::arms::DEFAULT_REDB_CACHE_BYTES,
4127 )
4128 .unwrap();
4129
4130 let t = Instant::now();
4131 let hit = tail.extents_with_rows(&acked).unwrap();
4132 let clean = t.elapsed();
4133 assert!(
4134 hit.iter().all(|&h| h),
4135 "an absorbed pack was reported unabsorbed"
4136 );
4137
4138 // One pack in flight when the machine died: durable bytes, no rows.
4139 let mut in_flight = acked.clone();
4140 let end = acked[PACKS - 1].0 + acked[PACKS - 1].1;
4141 in_flight.push((end, 4096));
4142 let t = Instant::now();
4143 let hit = tail.extents_with_rows(&in_flight).unwrap();
4144 let crashed = t.elapsed();
4145 assert_eq!(
4146 hit.iter().filter(|h| !**h).count(),
4147 1,
4148 "the pack with no rows is not the only one reported unabsorbed"
4149 );
4150
4151 eprintln!(
4152 "load {}; {PACKS} packs / {} object rows: derive-on-open diff {:.2} ms clean \
4153 (early exit), {:.2} ms with one pack un-absorbed (full tail scan)",
4154 loadavg(),
4155 tail.len(),
4156 clean.as_secs_f64() * 1e3,
4157 crashed.as_secs_f64() * 1e3,
4158 );
4159 }
4160
4161 /// **The bit is a bitset over dense ordinals, and one bit is one pack.**
4162 ///
4163 /// Ordinals that straddle word boundaries (63, 64, 65) are the whole point:
4164 /// a shift that ignores the word split, or a word index that does not, sets
4165 /// a neighbour's bit — and a neighbour's bit set means a durable pack is
4166 /// declared absorbed with no rows behind it.
4167 ///
4168 /// Seen RED by `>> (i % 64) & 1` → `>> (i % 63) & 1` in
4169 /// `PackState::is_absorbed`: "the bit for ordinal 64 did not go up". Restored.
4170 ///
4171 /// Seen RED a second time by `if self.extent[i].is_none()` → `if
4172 /// self.extent[i].is_none() || true` in `PackState::note`, i.e. letting a
4173 /// re-queue of a pack that is already recorded count as new work — which is
4174 /// the race the ack path really produces, since the drain can absorb before
4175 /// `queue` runs: "an absorbed pack was queued again — left: 6, right: 5".
4176 /// Restored.
4177 #[test]
4178 fn the_indexed_bit_is_a_bitset_over_dense_pack_ordinals() {
4179 let ids = [0u64, 1, 63, 64, 65, 4095];
4180 let mut s = PackState::default();
4181 for id in ids {
4182 s.note(id, (id * 4096 + 12, 4096));
4183 }
4184 assert_eq!(s.unabsorbed, ids.len());
4185 assert!(
4186 ids.iter().all(|&i| !s.is_absorbed(i)),
4187 "a bit is up already"
4188 );
4189
4190 s.mark_absorbed(64, (64 * 4096 + 12, 4096));
4191 assert!(s.is_absorbed(64), "the bit for ordinal 64 did not go up");
4192 for other in [0u64, 1, 63, 65, 4095] {
4193 assert!(
4194 !s.is_absorbed(other),
4195 "setting ordinal 64 also set ordinal {other} — one bit spilled onto a neighbour"
4196 );
4197 }
4198 assert_eq!(s.unabsorbed, ids.len() - 1);
4199 assert_eq!(
4200 s.pending().iter().map(|p| p.pack_id).collect::<Vec<_>>(),
4201 vec![0, 1, 63, 65, 4095],
4202 "the pending list is not the clear bits"
4203 );
4204 assert_eq!(
4205 s.pending()[0],
4206 PendingPack {
4207 pack_id: 0,
4208 offset: 12,
4209 len: 4096
4210 },
4211 "a pending pack lost the extent it must be absorbed from"
4212 );
4213
4214 // Both writers are idempotent, because both of them race.
4215 s.mark_absorbed(64, (64 * 4096 + 12, 4096));
4216 s.note(64, (64 * 4096 + 12, 4096));
4217 assert!(s.is_absorbed(64));
4218 assert_eq!(
4219 s.unabsorbed,
4220 ids.len() - 1,
4221 "an absorbed pack was queued again"
4222 );
4223
4224 // One bit per pack, and that is the whole cost: 4096 ordinals is 64
4225 // words. A hash set of the same ordinals is an order of magnitude more,
4226 // and a bloom would be smaller and wrong.
4227 assert_eq!(s.words.len(), 64);
4228 assert_eq!(s.words.len() * 8, 512, "4096 packs is 512 bytes of bits");
4229 }
4230
4231 /// **A read that arrives before the drain waits; it never answers absent.**
4232 ///
4233 /// The "before" is a *state*, not a race: the absorb gate is held by this
4234 /// thread, so the drain is provably parked at the top of its absorb and the
4235 /// index provably holds no row for the pack that `put_pack` just made
4236 /// durable.
4237 ///
4238 /// Two things are then asserted about a read issued into that state, and
4239 /// they are the two halves of §13.12. It must not **finish** — an answer
4240 /// while the index is empty could only have been "absent" — and when the
4241 /// gate is released it must answer **true**. After the drain, the same
4242 /// question is answered from the rows with nothing left to absorb.
4243 ///
4244 /// Seen RED **twice**, for the two halves.
4245 ///
4246 /// 1. The fallback: dropping `if self.unindexed_packs() > 0 {
4247 /// self.absorb_pending()?; }` from `lookup_one` gave "a read that arrived
4248 /// before the drain answered in 150 ms with an index holding 0 rows — the
4249 /// only answer it can have given is `absent`". Restored.
4250 /// 2. The bit's ordering: publishing the pack rows in `index_worker`
4251 /// *before* the absorb loop instead of after it gave "the pack's indexed
4252 /// bit went up while the drain is still parked in front of its absorb —
4253 /// the bit does not gate the object rows". Restored.
4254 ///
4255 /// **The second break is the reason the two assertions sit after the 150 ms
4256 /// sleep and not before it.** Asserted immediately after `put_pack` they were
4257 /// blind: the worker is still in its `pread`-and-SHA-1 fan-out at that
4258 /// instant, so the bit is legitimately down and the mutation stayed green. A
4259 /// guard for an ordering has to be read at a point where the wrong order has
4260 /// already had its chance.
4261 #[test]
4262 fn a_read_before_the_drain_waits_and_never_answers_absent() {
4263 let dir = tmpdir("drain-before");
4264 let store = Arc::new(GitStore::open(&dir, "rickard").unwrap());
4265 let (pack, rows) = real_pack();
4266
4267 // The drain is parked here, before it can absorb anything.
4268 let gate = store.hold_absorb_gate();
4269 let tx = store.put_pack(&pack).unwrap();
4270 let pack_id = tx.pack_id.unwrap();
4271
4272 assert_eq!(store.unindexed_packs(), 1, "the pack is not queued");
4273 assert_eq!(
4274 store.index().len(),
4275 0,
4276 "the index already holds rows — the 'before the drain' state is not the one under test"
4277 );
4278 assert!(
4279 !store.indexer().is_indexed(pack_id),
4280 "the bit is already up"
4281 );
4282
4283 // A read into exactly that state.
4284 let oid = rows[0].oid.clone();
4285 let answered = Arc::new(AtomicBool::new(false));
4286 let (s, a) = (store.clone(), answered.clone());
4287 // `Job` is this constellation's one sanctioned background thread (LAW 3);
4288 // a bare `std::thread::spawn` here would trip `rayon_free_law`.
4289 let reader = Job::spawn(move || {
4290 let got = s.has(&oid).unwrap();
4291 a.store(true, AtomicOrdering::Release);
4292 got
4293 });
4294 std::thread::sleep(Duration::from_millis(150));
4295 assert!(
4296 !answered.load(AtomicOrdering::Acquire),
4297 "a read that arrived before the drain answered in 150 ms with an index holding {} \
4298 rows — the only answer it can have given is `absent`",
4299 store.index().len()
4300 );
4301 // The drain has had 150 ms and is parked *before* its absorb, which is
4302 // where the bit's ordering is observable: it must still be down, and the
4303 // index must still be empty.
4304 assert_eq!(
4305 store.index().len(),
4306 0,
4307 "the index gained rows while the absorb gate was held"
4308 );
4309 assert!(
4310 !store.indexer().is_indexed(pack_id),
4311 "the pack's indexed bit went up while the drain is still parked in front of its \
4312 absorb — the bit does not gate the object rows"
4313 );
4314
4315 drop(gate);
4316 assert!(
4317 reader.join().unwrap(),
4318 "a read that waited for the drain still answered absent"
4319 );
4320
4321 // And afterwards it is served from the rows.
4322 store.wait_indexed();
4323 assert_eq!(store.unindexed_packs(), 0);
4324 assert!(store.indexer().is_indexed(pack_id));
4325 assert_eq!(store.index().len(), rows.len());
4326 assert_eq!(store.absorb_pending().unwrap(), 0);
4327 for r in rows.iter().take(256) {
4328 assert!(
4329 store.index().lookup(&r.oid).is_some(),
4330 "{} is not in the rows after the drain",
4331 hex::encode(&r.oid)
4332 );
4333 }
4334 }
4335
4336 /// **The bit is cleared after the rows go in, never before.**
4337 ///
4338 /// The ordering inside `absorb_gated` is the whole reason a read cannot be
4339 /// told "absent" while the drain is running, and the only way to see it is to
4340 /// read *during* one. Four gatling workers (LAW 3 — never rayon, and never a
4341 /// raw thread pool) hammer `has` over the pack's oids from the moment
4342 /// `put_pack` returns until the pack's bit is up.
4343 ///
4344 /// Asserted on applied output: not one of those reads may answer absent for
4345 /// an object whose bytes are already durable.
4346 ///
4347 /// Seen RED by clearing the bit *before* absorbing (a
4348 /// `s.pending.retain(|p| p.pack_id != job.pack_id)` added in `absorb_gated`
4349 /// above the `self.absorb_one(job)` call): "4 reader(s) were told a durable
4350 /// object was absent — **32 865 593** absent answers over 2687 objects".
4351 /// Restored. The count is that large because the readers spin until the bit
4352 /// goes up and, with the bit cleared first, every pass over the sample missed
4353 /// for the whole length of the absorb.
4354 ///
4355 /// The honest limit: this is a race, so a green run proves the window was not
4356 /// hit, not that no window exists. That is why the printed line carries how
4357 /// many of the reads actually landed while the pack was still un-indexed —
4358 /// **a run where that number is 0 proved nothing**, and it is reported rather
4359 /// than asserted because forcing it would mean pausing the drain, which is
4360 /// the one thing this test must not do. MEASURED 2026-08-08 on oden at
4361 /// loadavg 16.94: 336 of 336 reads landed early.
4362 #[test]
4363 fn no_read_is_told_absent_while_the_drain_is_running() {
4364 let dir = tmpdir("drain-during");
4365 let store = GitStore::open(&dir, "rickard").unwrap();
4366 let (pack, rows) = real_pack();
4367 let oids: Vec<&[u8]> = rows.iter().map(|r| r.oid.as_slice()).collect();
4368
4369 let tx = store.put_pack(&pack).unwrap();
4370 let pack_id = tx.pack_id.unwrap();
4371
4372 let early = AtomicU64::new(0);
4373 let absent = AtomicU64::new(0);
4374 let reads = AtomicU64::new(0);
4375 let workers = 4usize;
4376 gatling_for_each(workers, workers, |w| loop {
4377 let still_pending = store.unindexed_packs() > 0;
4378 for oid in oids.iter().skip(w).step_by(workers * 8) {
4379 if still_pending {
4380 early.fetch_add(1, AtomicOrdering::Relaxed);
4381 }
4382 reads.fetch_add(1, AtomicOrdering::Relaxed);
4383 if !store.has(oid).unwrap() {
4384 absent.fetch_add(1, AtomicOrdering::Relaxed);
4385 }
4386 }
4387 if store.indexer().is_indexed(pack_id) {
4388 break;
4389 }
4390 });
4391
4392 let absent = absent.load(AtomicOrdering::Acquire);
4393 assert_eq!(
4394 absent,
4395 0,
4396 "{workers} reader(s) were told a durable object was absent — {absent} absent answers \
4397 over {} objects",
4398 rows.len()
4399 );
4400 assert_eq!(store.object_count(), rows.len());
4401 eprintln!(
4402 "load {}; {} reads across {workers} gatling workers, {} of them while the pack was \
4403 still un-indexed",
4404 loadavg(),
4405 reads.load(AtomicOrdering::Acquire),
4406 early.load(AtomicOrdering::Acquire),
4407 );
4408 }
4409
4410 /// **One pack is absorbed once, whichever of the two callers gets there
4411 /// first.**
4412 ///
4413 /// Both callers are made to attempt it, deterministically: with the gate
4414 /// held, the drain is parked inside `ObjectAbsorb::absorb` and a reader is
4415 /// parked inside `absorb_pending`, both past their own early exits. Releasing
4416 /// the gate lets exactly one do the work; the other must take the skip.
4417 ///
4418 /// The observable is the **commit graph**, not the object table:
4419 /// `ObjectReadStack::append` is keyed by oid and would swallow a second
4420 /// identical row, but the graph is a `Vec` and a second absorb pushes every
4421 /// commit into it again — which silently doubles the fold every reachability
4422 /// bitmap is built from.
4423 ///
4424 /// Seen RED by disabling the `s.absorbed.contains(&job.pack_id)` skip in
4425 /// `absorb_gated`: "the commit graph carries **1102** commits, the pack has
4426 /// **551**" — exactly double, both callers having done the whole fold.
4427 /// Restored.
4428 #[test]
4429 fn a_pack_is_absorbed_once_even_when_both_callers_race_for_it() {
4430 let dir = tmpdir("drain-once");
4431 let store = Arc::new(GitStore::open(&dir, "rickard").unwrap());
4432 let (pack, rows) = real_pack();
4433
4434 let gate = store.hold_absorb_gate();
4435 store.put_pack(&pack).unwrap();
4436
4437 let s = store.clone();
4438 let reader = Job::spawn(move || s.absorb_pending().unwrap());
4439 // Both are now blocked on the gate: the drain inside `absorb`, the reader
4440 // inside `absorb_pending`, each having already decided there is work.
4441 std::thread::sleep(Duration::from_millis(100));
4442 drop(gate);
4443
4444 let by_read = reader.join().unwrap();
4445 store.wait_indexed();
4446
4447 assert_eq!(
4448 store.commit_count(),
4449 commits_in(&rows),
4450 "the commit graph carries {} commits, the pack has {}",
4451 store.commit_count(),
4452 commits_in(&rows)
4453 );
4454 assert_eq!(store.object_count(), rows.len());
4455 assert_eq!(store.unindexed_packs(), 0);
4456 eprintln!(
4457 "load {}; the read absorbed {by_read} pack(s), the drain absorbed the rest",
4458 loadavg()
4459 );
4460 }
4461
4462 // ── §14's exploded table ─────────────────────────────────────────────────
4463
4464 /// **A push produces an exploded row for every object in the pack, and each
4465 /// one hashes back to its own oid.**
4466 ///
4467 /// Eager, decided 2026-08-08: no threshold, no size cap, no type skipped —
4468 /// so the row count is the *object* count and not some subset of it. Nothing
4469 /// is read between `put_pack` and `wait_indexed`, so the only thing that
4470 /// could have built the rows is the background drain, which is the half of
4471 /// §14 that says "built by the same background indexer".
4472 ///
4473 /// The assertion is applied output in its strongest available form: every
4474 /// object's content is read back out of the table, re-serialised canonically
4475 /// and **re-hashed**, and the hash has to be the oid it was filed under. A
4476 /// row count alone could be satisfied by 2687 rows of the wrong bytes; a
4477 /// re-hash cannot.
4478 ///
4479 /// Seen RED by deleting the `sink.explode(&oid, kind, &payload)?` call in
4480 /// `resolve::resolve_walked`: "the drain built no exploded rows: 0 of 2687".
4481 /// Restored.
4482 ///
4483 /// Seen RED a second time by making the sink skip blobs in
4484 /// `ExplodedTable::explode` (`if kind == GitObjectKind::Blob { return Ok(()) }`
4485 /// at the top — which is what a *lazy* or size-capped table would look like
4486 /// from here): "the drain built no exploded rows: 1420 of 2687". Restored,
4487 /// because §14's open question was closed EAGER and disk is explicitly not a
4488 /// consideration — and because the 1267 rows that edit removes are exactly
4489 /// the blobs a content read wants most.
4490 #[test]
4491 fn a_push_produces_an_exploded_row_for_every_object_in_the_pack() {
4492 let dir = tmpdir("exploded-every-object");
4493 let store = GitStore::open(&dir, "rickard").unwrap();
4494 let (pack, rows) = real_pack();
4495
4496 let t = Instant::now();
4497 store.put_pack(&pack).unwrap();
4498 let ack = t.elapsed();
4499 store.wait_indexed();
4500 // **Drain, printed beside ack**, because the two answer different
4501 // questions and only one of them was ever in doubt. Ack is unchanged by
4502 // construction — nothing on it touches this table. Drain is where the
4503 // eager resolve and the sink both live, and it is the column redb's
4504 // 2026-08-08 measurement recorded as 316.2–329.8 ms against 166.7–168.6
4505 // ms with the sink stubbed out. Without it here, "did the medium buy
4506 // anything but disk" has no answer in this repository.
4507 let drain = t.elapsed();
4508
4509 let distinct: HashSet<Vec<u8>> = rows.iter().map(|r| r.oid.clone()).collect();
4510 let stats = store.exploded_stats();
4511 assert_eq!(
4512 stats.rows,
4513 distinct.len() as u64,
4514 "the drain built no exploded rows: {} of {}",
4515 stats.rows,
4516 distinct.len()
4517 );
4518 assert_eq!(
4519 stats.written, stats.rows,
4520 "rows exist that this process never committed"
4521 );
4522
4523 // Every one of them, re-hashed. Bytes that came back wrong cannot pass.
4524 let hash = store.hash_kind();
4525 let mut checked = 0usize;
4526 for oid in &distinct {
4527 let (kind, payload) = store
4528 .content(oid)
4529 .unwrap()
4530 .unwrap_or_else(|| panic!("{} has no exploded row", hex::encode(oid)));
4531 let rehashed = hash.oid_of(&crate::object::canonical(kind, &payload));
4532 assert_eq!(
4533 &rehashed,
4534 oid,
4535 "the exploded row filed under {} contains an object that hashes to {}",
4536 hex::encode(oid),
4537 hex::encode(&rehashed)
4538 );
4539 checked += 1;
4540 }
4541 assert_eq!(checked, distinct.len());
4542 // Nothing re-derived: the table answered all of them.
4543 assert_eq!(
4544 store.exploded_stats().rederived,
4545 0,
4546 "a content read re-resolved a pack even though the table was whole"
4547 );
4548 // **What it costs on disk, measured rather than assumed.** §14 says the
4549 // price is "both copies are held" and the decision was taken with "we
4550 // don't care if disk is tripled" — so the real multiple belongs in the
4551 // output where anybody can read it, not in a sentence. It is not 3x on a
4552 // real pack: the verbatim bytes are deflated *and* delta-encoded, the
4553 // table holds raw inflated content, and redb pages it.
4554 let inflated: u64 = rows.iter().map(|r| r.uncompressed_size).sum();
4555 let table = std::fs::metadata(store.exploded_path())
4556 .map(|m| m.len())
4557 .unwrap_or(0);
4558 eprintln!(
4559 "load {}; {} pack entries → {} exploded rows, all {} re-hashed to their own oid; \
4560 ack {:.1} ms, drain {:.1} ms, {:.0} rows/s; disk: pack {:.1} MiB verbatim, objects \
4561 inflate to {:.1} MiB, table file {:.1} MiB ({:.0}x the pack, {:.3}x the payload)",
4562 loadavg(),
4563 rows.len(),
4564 stats.rows,
4565 checked,
4566 ack.as_secs_f64() * 1e3,
4567 drain.as_secs_f64() * 1e3,
4568 stats.rows as f64 / drain.as_secs_f64(),
4569 pack.len() as f64 / (1 << 20) as f64,
4570 inflated as f64 / (1 << 20) as f64,
4571 table as f64 / (1 << 20) as f64,
4572 table as f64 / pack.len() as f64,
4573 table as f64 / inflated.max(1) as f64,
4574 );
4575 }
4576
4577 /// **The amplification measurement, at whatever scale you point it at.**
4578 ///
4579 /// [`a_push_produces_an_exploded_row_for_every_object_in_the_pack`] prints
4580 /// the same ratio on a 5.4 MiB fixture and is a *guard*, so it has to stay
4581 /// fast. This is the same arithmetic with no assertion about size and no
4582 /// fixture of its own: give it a pack and it reports what the table cost.
4583 ///
4584 /// ```text
4585 /// ZNIPPY_EXPLODE_BENCH_PACK=/path/to/pack-….pack \
4586 /// cargo test -p znippy-plugin-git --lib the_table_costs_what_it_holds \
4587 /// -- --ignored --nocapture
4588 /// ```
4589 ///
4590 /// It exists because the 12× figure that killed redb — 16.8 GB of resolved
4591 /// `linux.git` content in a 204 GB file — was taken by hand off a bench run
4592 /// that no longer exists, and a number nobody can re-take is a number that
4593 /// rots. `#[ignore]`d and pack-less by default: with no pack named it
4594 /// returns, so it never fails a suite for the absence of a corpus nobody
4595 /// promised to keep.
4596 #[test]
4597 #[ignore = "needs a pack: ZNIPPY_EXPLODE_BENCH_PACK=<path> cargo test … -- --ignored --nocapture"]
4598 fn the_table_costs_what_it_holds() {
4599 let Ok(path) = std::env::var("ZNIPPY_EXPLODE_BENCH_PACK") else {
4600 eprintln!("no ZNIPPY_EXPLODE_BENCH_PACK named; nothing measured");
4601 return;
4602 };
4603 let pack = std::fs::read(&path).unwrap_or_else(|e| panic!("reading {path}: {e}"));
4604 let dir = tmpdir("exploded-cost");
4605 let store = GitStore::open(&dir, "rickard").unwrap();
4606
4607 let t = Instant::now();
4608 store.put_pack(&pack).unwrap();
4609 let ack = t.elapsed();
4610 store.wait_indexed();
4611 let drain = t.elapsed();
4612
4613 let stats = store.exploded_stats();
4614 let table = std::fs::metadata(store.exploded_path())
4615 .map(|m| m.len())
4616 .unwrap_or(0);
4617 // The payload the table actually holds, asked of the table rather than
4618 // of the pack — a pack entry's `uncompressed_size` is what git recorded,
4619 // and this is what we stored. Only the second one can be divided into
4620 // the file size and mean anything.
4621 let mut held = 0u64;
4622 for kind in [
4623 GitObjectKind::Commit,
4624 GitObjectKind::Tree,
4625 GitObjectKind::Blob,
4626 GitObjectKind::Tag,
4627 ] {
4628 held += store
4629 .exploded_of_kind(kind)
4630 .unwrap()
4631 .iter()
4632 .map(|(_, p)| p.len() as u64)
4633 .sum::<u64>();
4634 }
4635 let mib = |n: u64| n as f64 / (1 << 20) as f64;
4636 eprintln!(
4637 "load {}; pack {} — {:.1} MiB verbatim → {} rows holding {:.1} MiB, table file \
4638 {:.1} MiB = {:.3}x the payload and {:.1}x the pack; ack {:.0} ms, drain {:.0} ms",
4639 loadavg(),
4640 path,
4641 mib(pack.len() as u64),
4642 stats.rows,
4643 mib(held),
4644 mib(table),
4645 table as f64 / held.max(1) as f64,
4646 table as f64 / pack.len() as f64,
4647 ack.as_secs_f64() * 1e3,
4648 drain.as_secs_f64() * 1e3,
4649 );
4650 }
4651
4652 /// Where the child of the clean-shutdown guard finds its fixture and leaves
4653 /// its markers.
4654 const CLEAN_DIR: &str = "GUNNAR_CLEAN_SHUTDOWN_DIR";
4655
4656 /// **The other half of
4657 /// [`a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable`]: the process
4658 /// that exits normally.**
4659 ///
4660 /// A real process, a real push, a real drain, and then a real `drop` — every
4661 /// worker joined, every redb handle closed, no signal, exit status 0. That is
4662 /// what makes the parent's reopen a *clean* reopen rather than a recovery,
4663 /// and it is the case the crash guard cannot reach: after a clean shutdown
4664 /// every pack's bit is legitimately set and **nothing is re-queued**, so the
4665 /// derived tables have to have been on disk or they are gone.
4666 ///
4667 /// `#[ignore]`d because it is a fixture, not a guard — run on its own it has
4668 /// no directory to work in and returns.
4669 #[test]
4670 #[ignore = "spawned by a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable"]
4671 fn the_child_that_pushes_and_shuts_down_cleanly() {
4672 let Ok(dir) = std::env::var(CLEAN_DIR) else {
4673 return;
4674 };
4675 let dir = PathBuf::from(dir);
4676 let pack = std::fs::read(dir.join("fixture.pack")).expect("the parent's fixture");
4677
4678 let store = GitStore::open(&dir, "rickard").expect("open");
4679 store.put_pack(&pack).expect("the ack path");
4680 store.wait_indexed();
4681 assert_eq!(store.unindexed_packs(), 0, "the child shut down mid-index");
4682
4683 let tip = store
4684 .graph_snapshot()
4685 .into_iter()
4686 .max_by_key(|c| c.generation)
4687 .expect("the child built no graph at all");
4688 let tip_raw = hex::decode(&tip.oid).unwrap();
4689 let closure = store.reachable(&[&tip_raw], &[]).unwrap();
4690 assert!(!closure.is_empty(), "the child's own reachable() was empty");
4691
4692 std::fs::write(dir.join("tip"), &tip.oid).expect("marker");
4693 std::fs::write(dir.join("closure"), closure.len().to_string()).expect("marker");
4694 std::fs::write(dir.join("commits"), store.commit_count().to_string()).expect("marker");
4695 std::fs::write(dir.join("objects"), store.object_count().to_string()).expect("marker");
4696
4697 // **The clean shutdown.** Not a signal and not an abort: the store drops,
4698 // which closes the channel, joins the account indexer's worker and closes
4699 // redb. The process then returns normally with status 0.
4700 drop(store);
4701 std::fs::write(dir.join("clean"), b"closed").expect("marker");
4702 }
4703
4704 /// **`commit_count()` and `reachable()` are correct after a CLEAN shutdown
4705 /// and reopen.** This is the guard the exploded table exists to make pass.
4706 ///
4707 /// It was the live bug: the commit graph was folded from commit and tree
4708 /// *payloads* which were stored nowhere, accumulated in RAM as packs were
4709 /// absorbed. A store killed mid-absorb got them back only because the
4710 /// interrupted pack was re-queued and re-resolved
4711 /// ([`a_store_reopened_after_a_kill_mid_absorb_requeues_the_pack`]); a store
4712 /// that shut down **cleanly** re-queued nothing, and MEASURED on oden
4713 /// 2026-08-08 came back with 2687 rows and **0 of 551 commits** —
4714 /// `reachable()` returning a commit alone instead of its closure, and `gc`
4715 /// seeing an empty live set. Both are quiet wrong answers.
4716 ///
4717 /// The shape is the crash guard's: a **real** child process, which here exits
4718 /// normally, and the parent asserts against what is on disk. Two assertions
4719 /// carry it and they have to be in this order:
4720 ///
4721 /// 1. **`commit_count()` immediately after `open`**, before anything is
4722 /// waited on — so the graph can only have come from the fold over the
4723 /// exploded table that `open` performs, not from a re-absorb;
4724 /// 2. **`is_indexed(0)` is false after `wait_indexed()`** — the account
4725 /// indexer in *this* process publishes a pack only for a job somebody
4726 /// submitted, so a clear bit proves open re-queued nothing and the graph
4727 /// above is not a recovery artefact.
4728 ///
4729 /// `reachable()` is then asserted against the closure the child computed on
4730 /// the same tip, so it is the same answer and not merely a non-empty one.
4731 ///
4732 /// Seen RED by deleting `store.refold()?` from the end of
4733 /// `GitStore::open_with_arms`, so nothing folds the table on open: "the
4734 /// commit graph did not survive a clean shutdown: 0 of 551 commits — the
4735 /// exploded table is a warm cache, not a durable table", left 0 right 551.
4736 /// That is the recorded bug's own number, reproduced. Restored.
4737 ///
4738 /// Seen RED a second time by deleting `self.exploded.flush()?` from
4739 /// `Absorber::absorb_one`, which is the **durability** edit: the rows then
4740 /// live only in the un-flushed buffer, which answers perfectly inside the
4741 /// child and is gone when the process ends. It fires assertion (2) rather
4742 /// than (1) — "the reopen re-queued the pack, so the graph above is a
4743 /// recovery artefact rather than the table's" — and that is worth writing
4744 /// down, because assertion (1) *passed*: the 64 MiB flush threshold had
4745 /// tripped once mid-pack, leaving 1798 of 2687 rows on disk which happened
4746 /// to include all 551 commits. A guard with only assertion (1) would have
4747 /// called that edit green. Restored.
4748 #[test]
4749 fn a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable() {
4750 let dir = tmpdir("clean-shutdown-graph");
4751 let (pack, rows) = real_pack();
4752 std::fs::write(dir.join("fixture.pack"), &pack).unwrap();
4753
4754 let status = std::process::Command::new(std::env::current_exe().unwrap())
4755 .args([
4756 "--exact",
4757 "--ignored",
4758 "--nocapture",
4759 "git_ops::tests::the_child_that_pushes_and_shuts_down_cleanly",
4760 ])
4761 .env(CLEAN_DIR, &dir)
4762 .status()
4763 .expect("spawn the child that shuts down cleanly");
4764 assert!(
4765 status.success(),
4766 "the child did not exit cleanly — it {status:?}, so this is a crash guard and not a \
4767 clean-shutdown one"
4768 );
4769 assert!(
4770 dir.join("clean").exists(),
4771 "the child never reached its clean shutdown"
4772 );
4773
4774 let tip_hex = std::fs::read_to_string(dir.join("tip")).unwrap();
4775 let closure_before: usize = std::fs::read_to_string(dir.join("closure"))
4776 .unwrap()
4777 .parse()
4778 .unwrap();
4779 let commits_before: usize = std::fs::read_to_string(dir.join("commits"))
4780 .unwrap()
4781 .parse()
4782 .unwrap();
4783 assert_eq!(
4784 commits_before,
4785 commits_in(&rows),
4786 "the child's own graph was already wrong, so the reopen proves nothing"
4787 );
4788
4789 // ── the reopen, and (1): the graph before anything is waited on ──────
4790 let store = GitStore::open(&dir, "rickard").unwrap();
4791 assert_eq!(
4792 store.commit_count(),
4793 commits_in(&rows),
4794 "the commit graph did not survive a clean shutdown: {} of {} commits — the exploded \
4795 table is a warm cache, not a durable table",
4796 store.commit_count(),
4797 commits_in(&rows)
4798 );
4799 let tip_raw = hex::decode(&tip_hex).unwrap();
4800 let closure = store.reachable(&[&tip_raw], &[]).unwrap();
4801 assert_eq!(
4802 closure.len(),
4803 closure_before,
4804 "reachable() answers differently after a clean reopen: {} objects against the {} the \
4805 same tip closed over before the shutdown",
4806 closure.len(),
4807 closure_before
4808 );
4809 assert!(closure.contains(&tip_raw));
4810
4811 // ── (2): and none of it came from a re-absorb ────────────────────────
4812 store.wait_indexed();
4813 assert!(
4814 !store.indexer().is_indexed(0),
4815 "the reopen re-queued the pack, so the graph above is a recovery artefact rather \
4816 than the table's"
4817 );
4818 assert_eq!(store.unindexed_packs(), 0);
4819 assert_eq!(
4820 store.absorb_pending().unwrap(),
4821 0,
4822 "the reopen left index work"
4823 );
4824 assert_eq!(store.object_count(), rows.len());
4825 assert_eq!(
4826 store.exploded_stats().written,
4827 0,
4828 "this process wrote exploded rows, so the table was rebuilt rather than read"
4829 );
4830
4831 eprintln!(
4832 "load {}; clean shutdown → reopen: {} rows, {} commits, tip closure {} objects \
4833 (rebuilt nothing)",
4834 loadavg(),
4835 store.object_count(),
4836 store.commit_count(),
4837 closure.len(),
4838 );
4839 }
4840
4841 /// **A content read is served from the table, not re-derived** — and the only
4842 /// thing that can say so is a counter.
4843 ///
4844 /// Both paths return the **identical** bytes: the table hands back what the
4845 /// resolver produced, and the fallback re-resolves the same pack and produces
4846 /// it again. A byte comparison therefore proves nothing at all about which
4847 /// one ran — this is the identity-value trap in its exact form, and it fired
4848 /// on this codebase once already. So the assertion is on
4849 /// [`ExplodedStats::served`] against [`ExplodedStats::rederived`], both of
4850 /// which are bumped where the work happens.
4851 ///
4852 /// The fallback is not asserted away: it is exercised in the same test, on a
4853 /// store whose table has been dropped, and the counters move the other way.
4854 /// A guard that only saw the fast path could not tell a working counter from
4855 /// one that is never incremented.
4856 ///
4857 /// Seen RED by deleting the `if let Some(hit) = self.exploded.content(oid)?`
4858 /// early return in `Absorber::resolved`, so every read re-derives: "512 content
4859 /// reads re-resolved a whole pack instead of hitting the table — served 0,
4860 /// rederived 512". The bytes were still correct on every one of them, which
4861 /// is the point. Restored.
4862 ///
4863 /// Seen RED a second time by bumping `served` unconditionally at the top of
4864 /// `ExplodedTable::content`, before the `pending`/redb probe — the counter
4865 /// then reports a hit whether or not the table had the object. The first half
4866 /// fires on the double count, "512 content reads re-resolved a whole pack
4867 /// instead of hitting the table — served 1024, rederived 0", and
4868 /// `exploded::tests::a_row_round_trips_and_the_kind_index_finds_it` fires with
4869 /// it ("a content read was not counted", left 5 right 2). Restored — a
4870 /// counter that cannot be wrong about a miss is the only kind worth
4871 /// asserting on.
4872 #[test]
4873 fn a_content_read_is_served_from_the_table_and_not_re_derived() {
4874 const READS: usize = 512;
4875 let dir = tmpdir("exploded-served");
4876 let (pack, rows) = real_pack();
4877 let oids: Vec<Vec<u8>> = rows
4878 .iter()
4879 .map(|r| r.oid.clone())
4880 .collect::<HashSet<Vec<u8>>>()
4881 .into_iter()
4882 .take(READS)
4883 .collect();
4884
4885 {
4886 let store = GitStore::open(&dir, "rickard").unwrap();
4887 store.put_pack(&pack).unwrap();
4888 store.wait_indexed();
4889
4890 let before = store.exploded_stats();
4891 let t = Instant::now();
4892 for oid in &oids {
4893 assert!(
4894 store.content(oid).unwrap().is_some(),
4895 "the table lost an object"
4896 );
4897 }
4898 let served_in = t.elapsed();
4899 let after = store.exploded_stats();
4900 assert_eq!(
4901 (
4902 after.served - before.served,
4903 after.rederived - before.rederived
4904 ),
4905 (oids.len() as u64, 0),
4906 "{} content reads re-resolved a whole pack instead of hitting the table — \
4907 served {}, rederived {}",
4908 oids.len(),
4909 after.served - before.served,
4910 after.rederived - before.rederived,
4911 );
4912 eprintln!(
4913 "load {}; {} content reads from §14's table in {:.1} ms ({:.0} ns/read)",
4914 loadavg(),
4915 oids.len(),
4916 served_in.as_secs_f64() * 1e3,
4917 served_in.as_secs_f64() * 1e9 / oids.len() as f64,
4918 );
4919 }
4920
4921 // ── and the fallback, so the counter is not a constant ───────────────
4922 //
4923 // The table is dropped and the *drain* is what would rebuild it, so the
4924 // reads have to happen before it gets there. `absorb_pending` is called
4925 // by hand first so the fallback under test is the content path's and not
4926 // the index's.
4927 std::fs::remove_file(dir.join("objects.exploded")).unwrap();
4928 let store = GitStore::open(&dir, "rickard").unwrap();
4929 let gate = store.hold_absorb_gate();
4930 let before = store.exploded_stats();
4931 let mut rederived = 0u64;
4932 for oid in oids.iter().take(8) {
4933 // Straight at the absorber: the store's `content` would absorb the
4934 // re-queued pack first and rebuild the very table this is testing
4935 // the absence of.
4936 assert!(
4937 store.absorber.resolved(oid).unwrap().is_some(),
4938 "the verbatim truth could not re-derive {}",
4939 hex::encode(oid)
4940 );
4941 rederived += 1;
4942 }
4943 let after = store.exploded_stats();
4944 drop(gate);
4945 assert_eq!(
4946 after.served - before.served,
4947 0,
4948 "a dropped table still reported {} reads served",
4949 after.served - before.served
4950 );
4951 assert_eq!(
4952 after.rederived - before.rederived,
4953 rederived,
4954 "the fallback ran but was not counted"
4955 );
4956 }
4957
4958 /// **Dropping the table and reopening still answers correctly — fall back,
4959 /// then rebuild.**
4960 ///
4961 /// §14 calls the resolved table *droppable*: derived, verifiable against the
4962 /// verbatim truth, and deletable at any time without consulting a client.
4963 /// This is that sentence as an operation — `rm objects.exploded` between two
4964 /// opens — and §13.12's rule applied to it unchanged: **absent means fall
4965 /// back, never means wrong.**
4966 ///
4967 /// Three things are asserted, in this order:
4968 ///
4969 /// 1. the file really is gone and the reopened store really does start with
4970 /// an empty table — otherwise the rest is testing a warm one;
4971 /// 2. **`reachable()` is already correct**, asked before any drain has been
4972 /// waited on: the fall-back absorb runs inline and rebuilds what the
4973 /// answer needs;
4974 /// 3. the table is **rebuilt** — one row per object again, the graph back,
4975 /// and the rows written by *this* process, which is what says a rebuild
4976 /// happened rather than a survival.
4977 ///
4978 /// Seen RED by `let exploded_is_whole = self.exploded.rows()? >= self.objects.len()
4979 /// as u64;` → `= true;` in `Absorber::adopt_journal`, so a dropped table is
4980 /// never noticed and nothing is re-queued. Assertion (2) fires first:
4981 /// **"reachable() after dropping the table: 1 objects, against 2683
4982 /// before"** — one object, the tip alone, which is the recorded bug's exact
4983 /// symptom ("`reachable` on a commit returns that commit alone rather than
4984 /// its closure") arrived at from the other direction. Restored.
4985 ///
4986 /// Seen RED a second time by `>=` → `>` in the same expression, which makes a
4987 /// *whole* table look short. This guard stays green — the rebuild is
4988 /// correct — and two others fire instead:
4989 /// [`a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable`]'s "the
4990 /// reopen re-queued the pack, so the graph above is a recovery artefact
4991 /// rather than the table's", and the pre-existing
4992 /// [`a_clean_reopen_requeues_nothing_and_the_next_push_gets_a_fresh_ordinal`]'s
4993 /// "a fully absorbed pack was re-queued on open — the diff is not exact".
4994 /// That pair is the point: one guard catches the table not being rebuilt when
4995 /// it should be, the others catch it being rebuilt when it should not.
4996 /// Restored.
4997 #[test]
4998 fn dropping_the_exploded_table_and_reopening_still_answers() {
4999 let dir = tmpdir("exploded-dropped");
5000 let (pack, rows) = real_pack();
5001 let tip_hex;
5002 let closure_before;
5003 {
5004 let store = GitStore::open(&dir, "rickard").unwrap();
5005 store.put_pack(&pack).unwrap();
5006 store.wait_indexed();
5007 let tip = store
5008 .graph_snapshot()
5009 .into_iter()
5010 .max_by_key(|c| c.generation)
5011 .expect("a graph");
5012 let tip_raw = hex::decode(&tip.oid).unwrap();
5013 closure_before = store.reachable(&[&tip_raw], &[]).unwrap().len();
5014 tip_hex = tip.oid;
5015 }
5016
5017 // (1) — the drop.
5018 let table = dir.join("objects.exploded");
5019 assert!(table.exists(), "the store never built an exploded table");
5020 std::fs::remove_file(&table).unwrap();
5021 {
5022 let fresh = crate::exploded_arrow::ExplodedArchive::open(&table).unwrap();
5023 assert_eq!(fresh.rows().unwrap(), 0, "the drop did not drop anything");
5024 }
5025 // Opening an absent table does not create one — so this only has
5026 // anything to remove if the assertion above was inspecting a live file.
5027 let _ = std::fs::remove_file(&table);
5028
5029 // (2) — a correct answer with the table gone.
5030 let store = GitStore::open(&dir, "rickard").unwrap();
5031 let tip_raw = hex::decode(&tip_hex).unwrap();
5032 let closure = store.reachable(&[&tip_raw], &[]).unwrap();
5033 assert_eq!(
5034 closure.len(),
5035 closure_before,
5036 "reachable() after dropping the table: {} objects, against {} before",
5037 closure.len(),
5038 closure_before
5039 );
5040 assert!(store.has(&rows[0].oid).unwrap());
5041
5042 // (3) — and it is back.
5043 store.wait_indexed();
5044 let stats = store.exploded_stats();
5045 let distinct: HashSet<Vec<u8>> = rows.iter().map(|r| r.oid.clone()).collect();
5046 assert_eq!(
5047 stats.rows,
5048 distinct.len() as u64,
5049 "the dropped table was never rebuilt: {} of {} rows",
5050 stats.rows,
5051 distinct.len()
5052 );
5053 assert!(
5054 stats.written > 0,
5055 "the table came back without this process writing a row — it was never dropped"
5056 );
5057 assert_eq!(
5058 store.commit_count(),
5059 commits_in(&rows),
5060 "the graph did not come back with the table: {} of {}",
5061 store.commit_count(),
5062 commits_in(&rows)
5063 );
5064 assert_eq!(
5065 store.object_count(),
5066 rows.len(),
5067 "the rebuild lost index rows"
5068 );
5069 eprintln!(
5070 "load {}; dropped and rebuilt: {} rows written, {} commits, tip closure {} objects",
5071 loadavg(),
5072 stats.written,
5073 store.commit_count(),
5074 closure.len(),
5075 );
5076 }
5077}