nordisk/znippy
plugins/native/znippy-plugin-git/src/git_ops.rs
//! **`GitOps` — ours, typed, and base znippy never sees it.**
//!
//! §13.2: `ArchiveTypePlugin` is base znippy's contract and
//! [`crate::NativeGitPlugin`] implements it — base stores, the plugin annotates,
//! exactly as maven, python, rust-toolchain and media already work. `GitOps` is
//! the *other* trait: the one gunnar calls, with git's own vocabulary in it.
//!
//! §13.3: `ArchiveTypePlugin::run_command(cmd, args)` is **CLI-only string
//! dispatch** and delegates here. A server's hot path does not go through
//! stringly-typed dispatch, so nothing below takes a `&str` command or an
//! `args: &[String]`.
//!
//! ```text
//! gunnar ──typed──► GitOps (this file, twelve methods)
//! │
//! ┌────────────────┼──────────────────┬─────────────┐
//! ▼ ▼ ▼ ▼
//! ArchiveWrite ObjectIndex RefLog Gc
//! SafeWriter ObjectReadStack push log NewGeneration
//! (blob fsync, (stree → Arrow (Arrow IPC (link, compact,
//! journal row) → redb tail) frame = tx) verify, rename)
//! ```
//!
//! Every arm under that diagram already existed and was already measured. This
//! file adds no storage, no index and no durability mechanism of its own — if a
//! method here looks like it is inventing one, that is a bug.
//!
//! # The one thing that is NOT here
//!
//! `run_command` still lives on the plugin in [`crate::lib`], and it now
//! delegates. Nothing git-shaped went into base znippy for any of this.
use std::collections::{BTreeMap, HashMap};
use std::fs::File;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, RwLock};
use anyhow::{anyhow, bail, Context, Result};
use znippy_common::ReservedSection;
use crate::arms::StoreConfig;
use crate::exploded::ExplodedStats;
use crate::exploded_cache::ExplodedCache;
use crate::gc::{Gc, GcReport};
use crate::graph::{assign_generations, CommitNode};
use crate::index_layout::{ObjectIndex, OneTableFourColumns};
use crate::indexer::{IndexJob, ObjectAbsorb, PushPath};
use crate::object::{GitHashKind, GitObjectKind};
use crate::pack_walk::PackWalk;
use crate::read_stack::{ObjectReadStack, RebuildTriggers};
use crate::refs::{RefLog, RefUpdate};
use crate::resolve::BaseSource;
// ── the contract, now owned by `git-storage-trait` ────────────────────────────
//
// `GitOps` (the neutral ELEVEN) and its value types live in the
// `git-storage-trait` crate at this repo's root, so a gix backend
// (edda/storage-git-gix) can implement the same contract without linking Arrow
// or OpenZL. The twelfth method, `seal()`, returns `znippy_common::
// ReservedSection` (an Arrow type a gix backend has no analog for), so it is an
// **inherent** method on [`GitStore`] / [`SelectedStore`], not part of the
// trait. The re-exports below keep every existing path
// (`znippy_plugin_git::git_ops::{GitOps, Oid, …}`) alive.
pub use git_storage_trait::{Extent, GitOps, Oid, RefRow, Stored, TxId};
// ── the batch shape, which is measured and must not be papered over ───────────
/// Which path a lookup of `n` oids takes.
///
/// The number this dispatch was written for is `index_layout_bench`'s: on the bare
/// Arrow arms, `lookup_batch` with a **single** element costs 818 ns against
/// `lookup`'s 482 ns — 1.7x worse — because the batch path allocates a result
/// vector, a key vector and a second pass over them to serve one oid.
///
/// **RE-MEASURED against the shipping read stack, and the magnitude does not
/// carry over.** oden 2026-08-07, release, 1-min loadavg 4.94 (another tenant's
/// work — see LAW: never measure on a busy box, and read these as a ratio rather
/// than as absolutes), `ObjectReadStack<OneTableFourColumns>` over a real
/// repository's 2687-object pack, 20 000 iterations
/// ([`crate::store::tests::the_batch_of_one_really_is_slower_than_the_serial_path`]):
///
/// | call | per oid |
/// |---|---:|
/// | `lookup` | **628 ns** |
/// | `lookup_batch` with 1 oid | **639 ns** — 1.02x worse |
/// | `lookup_batch` with 100 oids | **161 ns** — 3.9x better |
///
/// So: the *direction* holds and the dispatch is right — a batch of one is never
/// faster — but on this stack at this size it costs 2%, not 70%. The real finding
/// is the third row: the batch path is worth **3.9x** at
/// [`BATCH_SATURATES_AT`], which is a much stronger reason to keep the two paths
/// apart than the batch-of-one penalty ever was. Both numbers are stated because
/// quoting only the 1.7x would be quoting a figure this stack does not reproduce.
///
/// The dispatch is a named function with its own guard precisely so it cannot
/// quietly become `lookup_batch` for every n.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LookupPath {
/// One oid: `ObjectIndex::lookup`.
Serial,
/// Two or more: `ObjectIndex::lookup_batch`, one pass.
Batch,
}
/// Oids at which the batch path stops getting faster. Reported, not enforced:
/// splitting a bigger batch would cost a pass and buy nothing.
pub const BATCH_SATURATES_AT: usize = 100;
/// The dispatch, in one place.
pub fn lookup_path(n: usize) -> LookupPath {
if n == 1 {
LookupPath::Serial
} else {
LookupPath::Batch
}
}
// ── the store ─────────────────────────────────────────────────────────────────
/// A pack whose bytes are durable but whose objects the index has not absorbed.
///
/// §13.12's `indexed` bit, in the form this store needs it: membership in
/// [`PackState::pending`] *is* the bit being clear. A read that arrives while it
/// is non-empty falls back to absorbing it — slower, never wrong.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct PendingPack {
pack_id: u64,
offset: u64,
len: u64,
/// The journal queued it on open (a restart), not a push in this process.
/// What makes the progress row say `resumed 0%` after a reboot.
resumed: bool,
}
impl PendingPack {
fn from_job(j: IndexJob) -> Self {
Self {
pack_id: j.pack_id,
offset: j.offset,
len: j.len,
resumed: false,
}
}
}
/// **§13.12's `indexed` bit — one bit per pack, indexed by pack ordinal.**
///
/// A plain bitset, and the two rejected alternatives say why:
///
/// * **not a hash set.** Pack ordinals are dense and small: they are assigned
/// `0..N` within **one repository** (§13.14 scopes a negotiation, and therefore
/// this store, to one repository — account-wide they would be sparse and the
/// array would grow with the account's total packs). Dense means the bit is one
/// shift and one AND with no hashing, and 10 000 packs is **1250 bytes** of
/// `words`, which lives in L1.
/// * **not a bloom filter.** A bloom's false positive would be *actively wrong*
/// here: it would report a pack absorbed when it is not, the read path would
/// stop falling back, and negotiation would be told `absent` about an object
/// whose bytes are durable — which makes a client withhold objects and lose
/// data. The bitset has no such mode. `1` means absorbed, `0` means fall back,
/// and both are exact.
///
/// **The bit is never stored.** It is derived on open by diffing two facts that
/// are already durable — the journal's extents against the index's rows — in
/// [`Absorber::adopt_journal`]. There is nothing for a derived bit to drift from.
///
/// **Why the extents sit beside it.** A `0` bit is only actionable with the
/// extent to absorb, so `extent[i]` carries it; `None` means "this store has not
/// been told about ordinal `i`", which is not the same as "not absorbed" and must
/// not be counted as work.
///
/// **Why `absorbed` is a separate fact from `extent`.** The ack path does two
/// things in this order: record the pack pending, then hand its extent to the
/// account's channel. The indexer's worker is a different thread and can reach
/// `absorb` *before* the first of those returns. With only a pending list, that
/// race ends in a bit that is set after the rows are already in and is never
/// cleared, so every later read pays a pointless fallback for the life of the
/// process. The bit is what the two writers agree on: [`PackState::note`] will
/// not queue a pack that is already done, and `absorb` will not do the work
/// twice.
///
/// Touched only under short critical sections — never across an absorb, which is
/// what the separate [`Absorber::gate`] is for. **The ack path must never block
/// behind index work.**
#[derive(Default)]
struct PackState {
/// One bit per pack ordinal, 64 to a word. `1` = its rows are in the index.
words: Vec<u64>,
/// The extent of each known ordinal, so a `0` bit can be acted on.
extent: Vec<Option<Extent>>,
/// Known ordinals whose bit is `0` — what [`GitStore::unindexed_packs`]
/// reports, kept as a count so the read path's check is a field load rather
/// than a scan.
unabsorbed: usize,
}
impl PackState {
/// The bit.
fn is_absorbed(&self, pack_id: u64) -> bool {
let i = pack_id as usize;
self.words
.get(i / 64)
.is_some_and(|w| w >> (i % 64) & 1 == 1)
}
/// Record a durable pack whose rows are not in yet. The bit **going clear**.
///
/// **Idempotent, and the extent slot is the single thing that makes it so.**
/// A second `note` for an ordinal that already has one — the ack path
/// arriving after the drain has already absorbed the pack, or a re-queue on
/// open racing the same pack's push — finds the slot filled and counts no new
/// work. [`PackState::mark_absorbed`] fills that same slot, so a pack that is
/// already done cannot be queued behind its own completion, and there is no
/// second condition here that could disagree with the first.
fn note(&mut self, pack_id: u64, extent: Extent) {
let i = pack_id as usize;
self.grow_to(i);
if self.extent[i].is_none() {
self.extent[i] = Some(extent);
self.unabsorbed += 1;
}
}
/// Room for ordinal `i` in both arrays. Dense ordinals are what makes this a
/// `resize` and not an insert.
fn grow_to(&mut self, i: usize) {
if self.words.len() <= i / 64 {
self.words.resize(i / 64 + 1, 0);
}
if self.extent.len() <= i {
self.extent.resize(i + 1, None);
}
}
/// The bit **going up**: this pack's rows are in the index.
fn mark_absorbed(&mut self, pack_id: u64, extent: Extent) {
let i = pack_id as usize;
self.grow_to(i);
if self.words[i / 64] >> (i % 64) & 1 == 0 {
self.words[i / 64] |= 1 << (i % 64);
if self.extent[i].is_some() {
self.unabsorbed -= 1;
}
}
self.extent[i] = Some(extent);
}
/// Every known pack whose bit is clear, in ordinal order.
fn pending(&self) -> Vec<PendingPack> {
self.extent
.iter()
.enumerate()
.filter(|(i, e)| e.is_some() && !self.is_absorbed(*i as u64))
.map(|(i, e)| {
let (offset, len) = e.expect("filtered on Some");
PendingPack {
pack_id: i as u64,
offset,
len,
resumed: true,
}
})
.collect()
}
}
/// The derived tables: the commit graph, the tree payloads the reachability
/// bitmaps are built from, and this store's own ordinal space.
///
/// **The ordinal space is this struct's, not the projection's.** An
/// [`crate::index_layout::IndexRow::ordinal`] is a row address inside one
/// projection generation and is re-derived by every rebuild, so a bitmap built
/// over projection ordinals would silently address different objects after any
/// rebuild. The bitmaps here are over `ordinal`, which is rebuilt in the same
/// fold that rebuilds them — the two cannot drift because one call produces both.
/// **Nothing in here is a source of truth.** Every field is folded from the
/// `objects` table and §14's exploded table by [`Absorber::refold`], and both of
/// those are on disk — which is what makes a clean reopen come back with the
/// same graph it shut down with. Before the exploded table existed, `graph` and
/// `trees` were *accumulated* here as packs were absorbed and stored nowhere, so
/// a clean shutdown lost them silently: MEASURED on oden 2026-08-08, a reopened
/// store over a fully absorbed 2687-object pack had all 2687 rows and **0 of 551
/// commits**, with `reachable()` returning a commit alone instead of its closure
/// and `gc` seeing an empty live set.
#[derive(Default)]
struct Derived {
/// Parents-before-children, generations assigned. Folded from the exploded
/// table's commit payloads at open, and **extended** after a push: the
/// previous rows plus the pushed commits, re-numbered by
/// [`crate::graph::assign_generations`]. The hex spelling is kept for the
/// sealed graph section and `graph_snapshot`; nothing that serves a request
/// reads it.
graph: Vec<CommitNode>,
/// [`Derived::graph`] in integers — rows, ordinals, parent rows — rebuilt
/// beside it at every fold with one hex decode per node, so no serving path
/// touches a hex string. Behind an `Arc` so a negotiation borrows it with a
/// refcount bump and a fold that lands mid-answer replaces the pointer, not
/// the rows under the reader.
ord_graph: Arc<crate::reach::OrdGraph>,
/// **The ordinal space, and it is append-only.**
///
/// A sorted base laid down by the first fold, then every object a later
/// push brings in appended behind it ([`crate::reach::OrdinalTable`]). The
/// sorted space this replaced shifted every ordinal above the new oids on
/// every push, so every stored bitmap had to be rewritten bit by bit
/// (`carry_reach`: 512 entries × every object of the repository, after a
/// one-commit push — that is the 4 s of CPU GUN-PERF's Z-2 measured behind
/// a 3-object fetch on git.git). Here a push appends and the table stands.
///
/// The space is re-sorted only when the appended part outgrows
/// [`ORDINAL_COMPACT_FRACTION`] of the base or the store seals, and then
/// through ONE remap table ([`crate::reach::OrdinalTable::remap_to`]) —
/// one lookup per object, one array index per bit.
ordinals: crate::reach::OrdinalTable,
/// The live reachability table over `ordinals`, or empty if nothing has
/// needed it yet. Behind an `Arc` (2026-08-14, a measurement, not a
/// tidy-up: the negotiation used to deep-copy every bitmap twice per
/// request). Survives every fold; see [`Absorber::refold`].
reach: Arc<crate::reach::ReachTable>,
/// row → entry index into [`Derived::reach`], rebuilt with `ord_graph` at
/// every fold so the pair can never be out of step.
reach_by_row: Arc<Vec<u32>>,
/// blake3 over the sorted oid list the last fold saw — the token the
/// persisted `objects.graph` is validated against.
token: [u8; 32],
/// Verbatim packs, so any absorbed object's content can be re-derived from
/// the truth on demand. `(archive offset, len)`.
packs: Vec<(u64, u64)>,
}
/// **The compaction rule**: a fold re-sorts the ordinal space when the appended
/// part is larger than this fraction of the base (and than
/// [`ORDINAL_COMPACT_MIN`] objects), renumbering the reachability table once
/// through one remap table. Until then every push appends and touches no
/// bitmap. A quarter keeps the appended index small (it is searched separately,
/// through a sorted index) and the renumbering rare.
pub(crate) const ORDINAL_COMPACT_FRACTION: usize = 4;
/// The floor under the rule: a tiny repository is never compacted for a
/// handful of objects.
pub(crate) const ORDINAL_COMPACT_MIN: usize = 4096;
/// **How often an extended table is written back.** The persisted table stays
/// valid across pushes it does not know about — a load `follow`s the store's
/// current oid list and extends for the commits it lacks — so writing it after
/// every push would buy nothing but the write. A full build and a compaction
/// always persist; an extension persists every this-many folds, and the drop
/// of the store persists whatever is left.
pub(crate) const REACH_PERSIST_EVERY: u64 = 16;
/// The delta-chain memo a full reachability build reads its trees through
/// ([`crate::resolve::ChainCache`]): bounded, FIFO by bytes, dropped with the
/// build. 128 MiB holds the last few thousand materialised trees of a packed
/// repository, which is where the next chain's base is.
pub(crate) const REACH_BUILD_CHAIN_CACHE_BYTES: usize = 128 << 20;
/// …and no more than this per object the store holds (floor 8 MiB), so a
/// small repository's build does not reserve a large repository's memo.
pub(crate) const REACH_BUILD_CHAIN_CACHE_PER_OBJECT: usize = 512;
/// **The object-level index, and the one thing that writes into it.**
///
/// Its own allocation rather than a field group inside [`GitStore`], and §13.9-12
/// is the only reason: the drain runs on the account indexer's worker thread, so
/// whatever it absorbs into cannot be reachable solely through a `&GitStore`.
/// Splitting it out is what lets the background worker and a falling-back read
/// call **the same** absorb instead of two copies of it (LAW 5) — the alternative
/// was a second implementation on the worker side, which is precisely the
/// twinning that law forbids.
///
/// It is a **leaf**: nothing in here points back at the store, so the worker's
/// `Arc` closes no cycle and the store still drops (and joins its worker)
/// normally.
/// Entries below which [`GitStore::resolve_emit_payloads`] does **not** fan out.
///
/// A `std::thread::scope` spawn is ~10–20 µs a worker and the per-entry work it
/// would divide is a bounds check plus a varint; a 100-object fetch would pay
/// four spawns to save four microseconds of parsing. The threshold is what keeps
/// `crate::serve`'s "one core's worth per transfer" true for every request that
/// is not a bulk clone — and a bulk clone is the only request the copy this
/// replaces was ever measured on.
const FAN_OUT_AT: usize = 4096;
/// The folded commit graph, cached beside the verbatim pack so a reopen *reads*
/// it instead of re-deriving it from every commit payload (S-065).
const GRAPH_CACHE_FILE: &str = "objects.graph";
/// `objects.graph` framing: 4-byte magic + 1-byte version + 32-byte token, then
/// the Arrow IPC commit graph.
const GRAPH_CACHE_MAGIC: &[u8; 4] = b"GRPH";
const GRAPH_CACHE_HEADER: usize = 4 + 1 + 32;
/// The live reachability table and its ordinal space, cached beside the
/// verbatim pack so a reopen *reads* them instead of walking every tree again
/// (GUN-PERF Z-1: 85 to 150 s on git.git after every start). See
/// [`Absorber::load_persisted_reach`] for the frame.
const REACH_CACHE_FILE: &str = "objects.reach";
const REACH_CACHE_MAGIC: &[u8; 4] = b"RECH";
/// magic (4) + version (1) + oid_len (u32) + base bytes (u64) + extra bytes (u64).
const REACH_CACHE_HEADER: usize = 4 + 1 + 4 + 8 + 8;
/// **`objects.ordinal` is no longer written, and an existing one is removed at
/// open.** It was S-082's memo of the ordinal space — every oid in sorted order,
/// with a `(epoch, count)` header validated against the redb tail — so that a
/// cold open could *read* the space instead of scanning the whole redb tail to
/// rebuild it. Since 2026-09-04 the read stack persists its projection
/// (`objects.index`) and the sorted oid list IS that index's oid column, mapped;
/// [`ObjectReadStack::oids_raw`] hands it back with a merge of the few redb rows
/// since the last fold. The scan the memo spared no longer exists, and a second
/// file carrying the same bytes under a bespoke header is exactly the kind of
/// file this store should not have. On the kernel mirror it was 278 964 902 B —
/// 20.0 B/object, the oid column to the byte. The name is kept only so the
/// stale file can be found and deleted.
const ORDINAL_CACHE_FILE: &str = "objects.ordinal";
/// The validity token for a persisted commit graph: blake3 over the sorted raw
/// oids of **every** object in the store.
///
/// It is both necessary and sufficient. The graph is a pure, deterministic
/// function of the commit payloads, which are content-addressed by their oids,
/// so an identical object set yields a byte-identical graph — and this hashes
/// the whole object set, in the one order `refold` already sorts it into, with
/// no read the fold did not already make. A mismatch (any push, any gc) forces a
/// recompute; a 256-bit blake3 makes an accidental match of two different object
/// sets not a case worth engineering against.
/// **`seen − live`: what a GC may drop.** `seen` is every oid the index held
/// when the GC started (`oids_in_order`), `live` the reachable set computed
/// after it. The difference is the dead set, and it is a POSITIVE set on
/// purpose: `retain(|o| live.contains(o))` would also drop every row that
/// arrived between the snapshot and the drop — a push in flight, whose refs
/// are not updated yet — because to `live` those are simply unknown. A row in
/// neither set is kept. `live` is hex, `seen` raw; the decode happens once
/// here rather than once per row in the retain predicate.
pub(crate) fn dead_oids(
seen: Vec<Vec<u8>>,
live: &std::collections::HashSet<String>,
) -> std::collections::HashSet<Vec<u8>> {
let live_raw: std::collections::HashSet<Vec<u8>> =
live.iter().filter_map(|h| hex::decode(h).ok()).collect();
seen.into_iter().filter(|o| !live_raw.contains(o)).collect()
}
fn graph_token(oids_raw: &[u8]) -> [u8; 32] {
let mut h = znippy_common::blake3::Hasher::new();
h.update(oids_raw);
*h.finalize().as_bytes()
}
/// Commits that get a reachability bitmap on the **live** selection path.
///
/// # It was `usize::MAX`, and that was the single most expensive line in a fetch
///
/// The comment that stood here said a sampled table *"is for the sealed archive
/// where the cost is disk"* and that every commit must be bitmapped on the live
/// path. That was not a preference, it was forced:
/// [`GitStore::reachable_oids`] had no walk, so a `want` with no bitmap could
/// only contribute itself — a silent under-send — and `crate::serve`'s `select`
/// therefore had to refuse unless *"in the graph"* and *"has a bitmap"* were the
/// same fact.
///
/// MEASURED on oden 2026-08-14, one 4-object negotiated fetch out of the
/// znippy-served `nornir` mirror (1 581 commits, 12 455 objects): the
/// `select.bitmap` stage was **75 % of the request's CPU**, and **530 218 of its
/// 575 349 allocations were this build** — which `refold` throws away again on
/// the next push. The gix arm answered the identical fetch in 6.8 ms because it
/// *reads* a `.bitmap` instead of building one.
///
/// [`crate::reach::accumulate`] is the walk that makes the cap legal, and 512 is
/// the number [`crate::reach::ReachPolicy::default`] already uses for the sealed
/// archive — the same figure git's own bitmap selection is of the order of, and
/// deliberately not a second, different constant to keep in step.
const LIVE_REACH_COMMITS: usize = 512;
/// [`LIVE_REACH_COMMITS`], or `ZNIPPY_GIT_REACH_COMMITS`.
///
/// **An operator knob, not an arm.** It selects no implementation and changes no
/// byte of any answer: the walk covers whatever the table does not, so the cap
/// trades build cost against walk cost and nothing else. That property is the
/// whole subject of
/// `the_bounded_walk_answers_exactly_what_a_full_bitmap_table_answers`, which
/// drives this very knob to its two extremes over one store.
///
/// Read **once per process** through [`crate::arms::read_env`], so it is
/// counted with every other environment read this crate makes, and so a
/// reachability request does not cost a `getenv` — until 2026-08-21 it did,
/// once per `reach_bitmaps`/`reachable_oids` call, uncounted.
pub(crate) fn live_reach_policy() -> crate::reach::ReachPolicy {
static MAX: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
let max_commits = *MAX.get_or_init(|| {
crate::arms::read_env(crate::arms::ENV_REACH_COMMITS)
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(LIVE_REACH_COMMITS)
});
crate::reach::ReachPolicy { max_commits }
}
/// Ceiling on phase-1 workers **per request**.
///
/// The serving tier admits ~32 concurrent transfers, and the fan-out primitive
/// has no reentrancy detection, so an uncapped `gatling_for_each` here is
/// 32 × ncores threads on a 128-core box. Four is a bound that leaves the
/// admission gate in charge of the machine.
const MAX_RESOLVE_WORKERS: usize = 4;
/// Workers for phase 1: [`MAX_RESOLVE_WORKERS`], or `ZNIPPY_GIT_EMIT_WORKERS`.
///
/// **An operator knob, not an arm.** It selects no implementation and changes no
/// byte of the emitted pack — a pack emitted at one worker and at sixteen is
/// byte-identical, because phase 1 only decides *where each entry's bytes are*
/// and phase 2 writes them in one serial order either way. `0` is read as "as
/// many as this box has cores", which is `gatling_for_each`'s own convention.
///
/// Read **once per process** through [`crate::arms::read_env`] — see
/// [`live_reach_policy`] for why; this one was a `getenv` per emitted request
/// until 2026-08-21.
fn emit_workers() -> usize {
static WORKERS: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
*WORKERS.get_or_init(|| {
crate::arms::read_env(crate::arms::ENV_EMIT_WORKERS)
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(MAX_RESOLVE_WORKERS)
})
}
pub(crate) struct Absorber<S: ObjectIndex = OneTableFourColumns> {
blobs: PathBuf,
/// Read handle on the verbatim packs, for `pread` of an extent.
///
/// The **fallback**, since 2026-08-14, not the ordinary path: `map` below
/// answers every extent the mapping covers, and this reads the ones past its
/// end (a push that landed after the snapshot was taken).
///
/// **It is the writer's own open file description** (2026-08-30, lane AH),
/// not a second `File::open`: this reads with `pread` at an explicit offset
/// and shares no cursor with `append`, so a private descriptor bought a
/// kernel object and a page-cache alias per open repository and no
/// isolation. See [`crate::archive_write::open_blobs`] and
/// `tests/fd_per_store.rs`.
reader: Arc<File>,
/// The same file, mapped read-only. See [`crate::archive_map`] for why an
/// append-only file `gc` never rewrites is safe to map, and for the ~20 % of
/// serve cycles the `pread`-per-object shape was spending on copies.
map: crate::archive_map::ArchiveMap,
hash: GitHashKind,
/// The `objects` table: stree → Arrow → redb tail.
///
/// `S` is the payload layout the projection is built in
/// ([`crate::arms::IndexArm`]) and it is a **type parameter, not a field**:
/// this is the hot path, every lookup goes through it, and the whole reason
/// the arms exist is to measure the differences between them. A `dyn` here
/// would put a virtual call on every single oid.
objects: ObjectReadStack<S>,
/// **§14's exploded objects table** — oid → resolved type and content, built
/// eagerly by this absorb and by nothing else.
///
/// Beside the `objects` table rather than inside it because the two have
/// different lifetimes: `objects` is the index and losing it loses the
/// repository's addressing, while this one is derived, verifiable against the
/// verbatim bytes and deletable at any time. They are separate files so that
/// "droppable" is a `rm`, not a schema migration.
///
/// SWAPPED 2026-08-13 from the redb `ExplodedTable` to **one Arrow IPC
/// table**, payload in the row. redb held 16.8 GB of kernel payload in a
/// 204 GB file — 12× — and serialised the gatling fan-out behind its one
/// writer at 96.8% of a single core. See [`crate::exploded_arrow`].
///
/// **BUDGETED AND EVICTABLE since lane Q (2026-08-30)**: the two-pool
/// [`ExplodedCache`] — the indexer's eager pool (commits and tags) beside a
/// lazy pool the FIRST reader of a blob or tree fills and a byte budget
/// empties. The unbounded eager tier it replaces MEASURED **141.7 GiB** for
/// the 8.12 GiB `torvalds/linux` pack on the rig, 17× the pack, which is
/// more than one appliance holds for all its repositories together.
exploded: Arc<ExplodedCache>,
derived: RwLock<Derived>,
/// How many folds since open read the persisted commit graph off disk vs
/// recomputed it from the exploded table. Counters, not clocks — the only
/// thing that can say which path a fold took (LAW 2). See
/// [`Absorber::graph_source_stats`].
graph_read: std::sync::atomic::AtomicU64,
graph_recomputed: std::sync::atomic::AtomicU64,
/// **Is `objects.graph` read and written at all** ([`crate::arms::ENV_GRAPH_CACHE`],
/// read once at open). `false` folds from the exploded table every time and
/// leaves no file: a store that serves no reads holds no derived memo.
graph_cache: std::sync::atomic::AtomicBool,
/// How many folds CARRIED the reachability table across (renumbered, no
/// tree opened) vs how many reads had to build it from scratch. The only
/// witness that the carry ran — the bytes a carried and a rebuilt table
/// answer with are identical by construction (LAW 2). See
/// [`Absorber::reach_source_stats`].
reach_carried: std::sync::atomic::AtomicU64,
reach_rebuilt: std::sync::atomic::AtomicU64,
/// How many folds READ the persisted table (`objects.reach`), how many
/// EXTENDED it for pushed commits, and how many extensions are not written
/// back yet. See [`Absorber::reach_stats`].
reach_loaded: std::sync::atomic::AtomicU64,
reach_extended: std::sync::atomic::AtomicU64,
reach_unpersisted: std::sync::atomic::AtomicU64,
/// Where the last fold's **ordinal space** came from: how many oids straight
/// off the persisted index's column, how many off a scan of the redb tail.
/// The same shape as [`Self::graph_read`]/[`Self::graph_recomputed`], and for
/// the same reason: which source produced the space is not visible in the
/// (identical) bytes, only in a counter (LAW 2). See
/// [`Absorber::ordinal_source_stats`].
ordinal_folded: std::sync::atomic::AtomicU64,
ordinal_tail: std::sync::atomic::AtomicU64,
packs: Mutex<PackState>,
/// Serialises **absorption**, which is long. Held across a whole pack.
///
/// Two absorbs of one pack are duplicated *work* and nothing worse: every
/// writer under it is keyed by oid — [`ObjectReadStack::append`] swallows an
/// identical second row, the exploded table overwrites its own key, and the
/// commit graph is folded from that table rather than pushed at, so it cannot
/// double. (It could before §14's table existed, when the graph was a `Vec`
/// accumulated per absorb; that is why this paragraph used to say the gate
/// was load-bearing for correctness and now does not.) The gate is still what
/// makes the ordering below hold: **the rows go in, and only then is the bit
/// cleared**, so a read either finds the bit still clear (and waits here) or
/// finds the rows. Neither order can produce "absent".
gate: Mutex<()>,
/// **The progress counter** ([`crate::progress`]) a dashboard or a stats
/// call reads. One per store, registered process-wide.
progress: Arc<crate::progress::IndexProgress>,
/// Set by [`GitStore::cancel_index`]: the repository is being deleted. The
/// resolver checks it every few thousand objects and aborts the pack; the
/// drain skips every pack still queued.
cancel: std::sync::atomic::AtomicBool,
/// Names the store in log lines: its root directory.
label: String,
/// The memory watchdog this store's worker pauses on. The process-wide one
/// unless a test hands in a fake ([`GitStore::set_mem_watchdog`]).
watchdog: RwLock<Arc<crate::memguard::MemWatchdog>>,
/// **Pending packs a reader has already scanned**, keyed by pack offset:
/// the walk and an oid-sorted row table, built by ONE read-only pass and
/// kept until the pack is absorbed. Every later lookup into that pack is
/// a binary search; a content read inflates one chain. Without this,
/// every `has` during the index window re-resolved the whole pack —
/// MEASURED on the rig 2026-08-30: a `push --mirror` of linux stalled
/// pre-ack, ~800 ref updates × one 8 GiB resolve each.
scanned: Mutex<HashMap<u64, Arc<ScannedPack>>>,
/// One scan build at a time: a second reader waits for the first reader's
/// pass rather than starting its own. That is waiting on a *reader*, not
/// on the indexer.
scan_build: Mutex<()>,
/// **Packs this store has stopped trying to index**, by ordinal, with how
/// many attempts it took to stop and the reason the last one gave.
///
/// 🔴 The state this replaces was *no state*. A pack whose absorb failed
/// stayed pending, every absorb loop hit it FIRST (they run in ordinal
/// order) and aborted on it with `?`, so the packs behind it were never
/// reached — and the next push, and the one after that, re-did the same
/// doomed resolve and was refused for it. MEASURED on oden 2026-08-30
/// (lane AI): one pack that could not be absorbed refused **every** push to
/// that repository from then on, 21 of 140, and lane AD's standby re-asked
/// for it every 300 s for 21 minutes. Work that cannot succeed has to stop
/// and say why; see [`MAX_ABSORB_ATTEMPTS`] and [`Absorber::absorb_each`].
unindexable: Mutex<HashMap<u64, Unindexable>>,
}
/// A pack this store will not try to index again, and why.
#[derive(Debug, Clone)]
pub struct Unindexable {
/// The pack's ordinal.
pub pack_id: u64,
/// Its extent in the verbatim archive. The bytes are still there.
pub extent: Extent,
/// How many absorbs were spent before it stopped.
pub attempts: u32,
/// The last attempt's root cause — the backend's own sentence, which is
/// what a client and an operator both get told.
pub reason: String,
}
/// How many times one pack's absorb is attempted before the store stops.
///
/// Three, and the reason it is not one: an absorb can fail for a reason outside
/// the pack — the memory watchdog gave up, the repository was being deleted, a
/// transient read error — and a single bad moment must not condemn durable
/// bytes. It is not thirty either, because the resolve is deterministic in the
/// store's own state: a pack that failed three times with nothing else changing
/// fails the fourth time for the same reason, and every one of those attempts
/// is paid for by a client waiting on a push.
pub const MAX_ABSORB_ATTEMPTS: u32 = 3;
// **The pack this thread is inside an absorb of**, if any.
//
// `absorb_gated` holds `Absorber::gate` for the whole pack, and the resolver it
// runs asks this very store for delta bases through `BaseSource::content`. That
// re-entry must not take the gate again — a `std::sync::Mutex` is not
// reentrant, so it would not be a wait, it would be a deadlock — and it must
// not treat the pack it is halfway through as something it can absorb. One
// `Cell` answers both questions, and it is the same guard shape
// `scan_pending`'s `SCANNING` uses for the same reason.
thread_local! {
static IN_ABSORB: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
}
/// Sets [`IN_ABSORB`] for the duration of one absorb and restores whatever was
/// there before — absorbs nest (a pack whose base is in another pending pack).
struct InAbsorb(Option<u64>);
impl InAbsorb {
fn enter(pack_id: u64) -> Self {
InAbsorb(IN_ABSORB.with(|c| c.replace(Some(pack_id))))
}
}
impl Drop for InAbsorb {
fn drop(&mut self) {
IN_ABSORB.with(|c| c.set(self.0));
}
}
/// One pending pack, scanned read-only: what a lookup needs and nothing that
/// the index will hold. O(objects) memory — for linux ~56 B of walk + 32 B of
/// row per object, ~1.2 GB — and gone when the pack's bit goes up.
///
/// **Nothing on the production read path builds one of these any more.** A
/// reader that needs an un-indexed pack indexes it instead
/// ([`Absorber::index_pending_now`]); this remains for the one caller that
/// cannot — a resolver already inside `absorb_gated`, which may consult a scan
/// somebody else already built but must never start one.
struct ScannedPack {
walk: crate::pack_walk::PackWalk,
/// `(oid, entry index, resolved size)`, sorted by oid.
rows: Vec<(Vec<u8>, u32, u64)>,
archive_offset: u64,
}
impl ScannedPack {
fn find(&self, oid: &[u8]) -> Option<(usize, u64)> {
self.rows
.binary_search_by(|(o, _, _)| o.as_slice().cmp(oid))
.ok()
.map(|i| (self.rows[i].1 as usize, self.rows[i].2))
}
fn row(&self, oid: &[u8]) -> Option<crate::index_layout::IndexRow> {
let (entry, size) = self.find(oid)?;
let e = &self.walk.entries[entry];
Some(crate::index_layout::IndexRow {
// Not ranked: the row is not in the index yet.
ordinal: u32::MAX,
offset: e.offset + self.archive_offset,
len: e.len,
obj_type: e.obj_type,
uncompressed_size: size,
delta_base: match &e.delta_base {
crate::pack_walk::DeltaBase::Offset(o) => o + self.archive_offset,
_ => 0,
},
})
}
}
/// One repository's git store.
///
/// | on disk | what it is | who owns the format |
/// |---|---|---|
/// | `<root>/objects.pack` | the verbatim pack bytes, appended | [`SafeWriter`] |
/// | `<root>/objects.pack.journal` | the durable extent rows | [`SafeWriter`] |
/// | `<root>/objects.tail` | the un-sealed index tail | [`ObjectReadStack`] (redb) |
/// | `<root>/objects.exploded` | §14's resolved content, **droppable** | [`ExplodedArchive`] (one Arrow IPC table) |
/// | `<root>/refs.log` | the ref transaction log | [`RefLog`] (Arrow IPC frames) |
/// | `<root>/<name>.znippy` | the sealed archive GC compacts | base znippy |
///
/// Nothing in that table is new. The store is the composition, not a fifth
/// format.
/// # Which implementation of each trait, and how a caller picks
///
/// Three of the four boxes in the diagram at the top of this file have more than
/// one implementation. [`StoreConfig`] is the choice, [`crate::arms`] documents
/// what each arm promises, and the dispatch is split deliberately:
///
/// | trait | selected by | dispatch | cost |
/// |---|---|---|---|
/// | [`ObjectIndex`] | the type parameter `S` | **static** | none — the hot path is monomorphised |
/// | [`ArchiveWrite`](crate::archive_write::ArchiveWrite) | [`StoreConfig::writer`] | `dyn`, inside [`PushPath`] as it already was | one virtual call **per pack append**, not per object |
/// | [`Gc`] | [`StoreConfig::gc`] | `dyn` | one virtual call per `gc()` run |
///
/// `S` defaults to [`OneTableFourColumns`], so `GitStore` still names exactly
/// what it named before and every existing caller compiles unchanged.
pub struct GitStore<S: ObjectIndex = OneTableFourColumns> {
root: PathBuf,
/// The verbatim pack bytes. The **truth** (§14).
blobs: PathBuf,
/// The sealed znippy archive `gc` compacts. Only GC and `seal` touch it.
archive: PathBuf,
/// The selected [`ArchiveWrite`](crate::archive_write::ArchiveWrite) arm
/// plus the per-account index channel. The ack path.
push: PushPath,
account: String,
hash: GitHashKind,
/// The object-level index and the absorb behind it — **shared with the
/// account indexer's worker**, which is how a push ends in object rows
/// without a read having to ask for them.
absorber: Arc<Absorber<S>>,
refs: RefLog,
/// Serialises the ref log's read-compare-append. A CAS that read the
/// namespace, then appended, without holding this could be overtaken between
/// the two and would overwrite the update it did not see.
ref_gate: Mutex<()>,
/// The selected [`Gc`] arm. Built once, here, and run by
/// [`GitOps::gc`] — a `Box<dyn>` because a GC is a whole-archive
/// compaction that happens on a maintenance timer, so one virtual call
/// against it is unmeasurable and a third type parameter on every signature
/// in the crate is not free to read.
gc: Box<dyn Gc + Send + Sync>,
/// What this store was built with, for a bench row or a config dump. A
/// description of the arms already constructed — **nothing consults it on a
/// serving path**.
arms: StoreConfig,
}
impl GitStore<OneTableFourColumns> {
/// Open (creating) the store for one repository under `root`.
///
/// **The default arms, and it reads no environment to get them.** An
/// operator who exported `ZNIPPY_GIT_WRITER=fast` does not change what this
/// builds — a caller that never asked for a selector must not have its
/// durability contract moved out from under it. Use
/// [`open_from_env`] when the selection is wanted.
///
/// ⚠ One variable *is* read here, and it is not an arm:
/// `ZNIPPY_GIT_REDB_CACHE_BYTES` (see
/// [`crate::arms::redb_cache_bytes`]). It selects no implementation,
/// changes no durability contract and leaves the bytes on disk identical —
/// it caps redb's page cache, whose own default is 1 GiB *per database*, of
/// which this store opens two. The sentence above is about arms, and a
/// memory ceiling is not one.
pub fn open(root: &Path, account: &str) -> Result<Self> {
Self::open_with(root, account, GitHashKind::Sha1)
}
/// Same, with an explicit oid width. sha256 repositories exist and the store
/// does not get to assume otherwise; every oid it stores is
/// [`GitHashKind::oid_len`] bytes wide and the tail refuses a mixed batch.
///
/// # What a reopen recovers, and why nothing had to be stored for it
///
/// A store opened over an `objects.pack` some earlier process was still
/// indexing has to end up in the same state that process would have reached.
/// §13.12's `indexed` bit is what says so, and it is **derived here rather
/// than persisted**: a pack is unabsorbed *iff* its extent is in the journal
/// and its rows are not in the index, and both of those are already durable
/// (see [`Absorber::adopt_journal`]). The packs that come back clear are put
/// straight onto the account's channel, so the drain finishes the interrupted
/// work with no read having to ask — and until it does, every read falls back
/// exactly as it would have before the crash. **Slower, never wrong**, across
/// a restart as well as within one.
///
/// The bits are set **before** the channel is fed, which is the same ordering
/// the ack path uses for the same reason: a read that arrives in between finds
/// the bit clear and waits, never an index that does not mention the pack.
///
/// # What a reopen recovers of the DERIVED tables, and how that was fixed
///
/// [`Derived::graph`] and [`Derived::trees`] come back **in full**, for every
/// pack, whether or not anything is re-queued. They are folded from §14's
/// exploded table ([`Absorber::refold`]), which is on disk, so a clean
/// shutdown carries them across.
///
/// This was a live correctness hole until 2026-08-08 and it is recorded
/// rather than quietly repaired. The graph was *accumulated in RAM* as packs
/// were absorbed and stored nowhere; a store killed mid-absorb got it back
/// because the interrupted pack was re-queued and re-resolved, but a store
/// that shut down **cleanly** re-queued nothing and came back with an empty
/// graph. MEASURED on oden 2026-08-08: 2687 rows and **0 of 551 commits**,
/// `reachable` on a commit returning that commit alone instead of its
/// closure — a *quiet* wrong answer — and `gc` seeing an empty live set. The
/// fix is the table, because the payloads the graph is folded from had no
/// home; [`tests::a_clean_shutdown_and_reopen_keeps_the_graph_and_reachable`]
/// is what proves it, over a real process that exits normally.
///
/// The exploded table is still **droppable**: if its row count is short of
/// the `objects` table's, [`Absorber::adopt_journal`] declares no pack
/// absorbed and every one of them is re-queued and re-exploded. Absent means
/// rebuild, never wrong (§13.12, applied unchanged to a derived table).
///
/// **Also the default arms, and also no environment read** — see
/// [`open`](GitStore::open).
pub fn open_with(root: &Path, account: &str, hash: GitHashKind) -> Result<Self> {
// The default ARMS, and the environment's CEILING: the one variable
// this constructor reads, because an operator capping a server's
// footprint must not have to pick a different constructor to do it.
// See [`crate::arms::redb_cache_bytes`].
Self::open_with_arms(
root,
account,
hash,
StoreConfig::DEFAULT.with_redb_cache_bytes(crate::arms::redb_cache_bytes()?),
)
}
}
/// Where a stored entry's delta base sits, relative to the request being served.
///
/// The `Outside` arm carries the base in **both** namings because the two delta
/// forms know only one each — an `OFS_DELTA` its archive offset, a `REF_DELTA`
/// its oid — and [`GitStore::emit_set`]'s thin arm has to ask "does the client
/// hold this" in whichever one the entry speaks.
enum BaseOf {
/// Not a delta at all. There is nothing to place it after and nothing to
/// re-head; the stored entry is copied as it stands.
Whole,
/// A delta whose base **is** in the request, at this archive offset.
///
/// The offset is carried rather than recomputed because a `REF_DELTA` does
/// not know it — it names its base by oid, and its own `delta_base` column
/// is `0` by construction — and the emitter needs it twice: as the
/// `EmitEntry::delta_base` that [`crate::pack_walk::topological_order`]
/// orders on, and as the base an in-pack ref-delta is re-headed to point at.
Inside { at: u64 },
/// A delta whose base the request does not carry.
Outside {
/// The base's archive offset, `0` when the entry named an oid instead.
offset: u64,
/// The base's oid, `None` when the entry named an offset instead — or
/// when the entry is too short to read one out of.
oid: Option<Vec<u8>>,
},
}
/// The bases a **thin** pack may name without carrying them: what the receiver
/// already held before this transfer, keyed both ways.
///
/// Built by [`GitStore::client_bases`], which is where the argument for trusting
/// it is written down.
struct ExternalBases {
/// The held object's oid, keyed by its archive offset — the coordinate an
/// `OFS_DELTA` names its base in, and the only reason this map exists.
by_offset: HashMap<u64, OidKey>,
/// The same objects keyed the way a `REF_DELTA` names them.
held: std::collections::HashSet<OidKey>,
}
/// **A raw oid as a fixed-size, `Copy` hash key — 20 bytes for SHA-1, 32 for
/// SHA-256, and no heap allocation either way.**
///
/// The two maps in [`ExternalBases`] are repository-sized: the client's `have`
/// closure on an incremental fetch is very nearly every object in the store.
/// Keying them by `Vec<u8>` is one heap allocation per entry per request —
/// 12 112 of them for a 69-object fetch, measured on oden 2026-08-15 — for a
/// question (*does the client hold this base?*) that never looks at more than
/// `oid_len` bytes.
///
/// Padded to the widest hash rather than parameterised by width: a store holds
/// exactly one hash kind, so every key in one set has the same real length and
/// two different oids can never collide through the padding. The length is
/// carried anyway, because a key that silently compared equal across widths is
/// the kind of wrong that only shows up in a mixed-hash mirror.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct OidKey {
bytes: [u8; 32],
len: u8,
}
impl OidKey {
/// The key for `oid`, or a refusal for one wider than any git hash.
fn new(oid: &[u8]) -> Result<Self> {
let len = oid.len();
if len == 0 || len > 32 {
bail!("{len} is not the width of any git object id (1..=32 bytes)");
}
let mut bytes = [0u8; 32];
bytes[..len].copy_from_slice(oid);
Ok(OidKey {
bytes,
len: len as u8,
})
}
/// The oid back, at its real width.
fn as_slice(&self) -> &[u8] {
&self.bytes[..self.len as usize]
}
}
impl<S: ObjectIndex + 'static> GitStore<S> {
/// **The one constructor**, with every arm named.
///
/// `S` picks the [`ObjectIndex`] layout at compile time; `arms` picks the
/// [`ArchiveWrite`](crate::archive_write::ArchiveWrite) and [`Gc`]
/// implementations at run time. [`open`](GitStore::open) and
/// [`open_with`](GitStore::open_with) are this function with
/// [`StoreConfig::DEFAULT`], which is why nothing about them moved.
///
/// `arms.index` is **not** consulted here — `S` is the index arm, and a
/// value that disagreed with the type would be a second source of truth.
/// [`open_from_env`] is the one place the two are tied together, and it does
/// it by choosing `S` from the value.
///
/// ```no_run
/// use znippy_plugin_git::arms::{GcArm, StoreConfig, WriterArm};
/// use znippy_plugin_git::index_layout::PackedPayload;
/// use znippy_plugin_git::{GitHashKind, GitStore};
///
/// // A rebuildable import mirror: the ceiling writer, the packed payload
/// // layout, and an in-place compaction.
/// let arms = StoreConfig::DEFAULT
/// .with_writer(WriterArm::Fast)
/// .with_gc(GcArm::CompactInPlace);
/// let store = GitStore::<PackedPayload>::open_with_arms(
/// std::path::Path::new("/srv/repo"),
/// "alice",
/// GitHashKind::Sha1,
/// arms,
/// )?;
/// # Ok::<(), anyhow::Error>(())
/// ```
pub fn open_with_arms(
root: &Path,
account: &str,
hash: GitHashKind,
arms: StoreConfig,
) -> Result<Self> {
std::fs::create_dir_all(root)
.with_context(|| format!("creating the store root {}", root.display()))?;
let blobs = root.join("objects.pack");
let writer = arms.writer.create(&blobs)?;
// `None` for an arm that keeps no durable ack log — see
// [`crate::arms::WriterArm::journal`]. Everything below that touches the
// journal is written against that `Option` rather than assuming one.
let journal = arms.writer.journal(&blobs);
// **The redb ceiling**, carried in on the config rather
// than read here: `StoreConfig::from_env` reads it with the arms, and
// `open`/`open_with` put the environment's value on `DEFAULT` before
// getting here, so every constructor still honours the variable and
// `arms.to_string()` prints what was actually applied. Not an arm — see
// [`crate::arms::redb_cache_bytes`] — redb's own default is 1 GiB *per
// database*, and this store opens exactly ONE of them, the
// `ObjectReadStack` below. The `let _ = cache_bytes` two lines down is
// what is left of the second.
let cache_bytes = arms.redb_cache_bytes;
let objects = ObjectReadStack::<S>::open(
&root.join("objects.tail"),
RebuildTriggers::default(),
cache_bytes,
)?;
let _ = cache_bytes; // no page cache to size: the blob is pread, the index is a slice
// A store written before 2026-09-04 carries `objects.ordinal`, the memo
// of the sorted oid list the read stack's persisted index now IS. Its
// bytes are never read again, so it goes on the first open by this
// build — 278 964 902 B on the kernel mirror. Absent is the normal case
// and not an event; anything else that stops the delete is logged, not
// fatal, because the store does not depend on the file being gone.
match std::fs::remove_file(root.join(ORDINAL_CACHE_FILE)) {
Ok(()) => tracing::info!(
target: "znippy.git",
path = %root.join(ORDINAL_CACHE_FILE).display(),
"removed a stale objects.ordinal — the persisted index's oid column is the \
ordinal space now"
),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => tracing::warn!(
target: "znippy.git",
path = %root.join(ORDINAL_CACHE_FILE).display(),
error = %e,
"could not remove a stale objects.ordinal; it is dead weight, not a fault"
),
}
let exploded = Arc::new(ExplodedCache::open(
&root.join("objects.exploded"),
crate::exploded_arrow::ExplodePolicy::from_env()?,
crate::exploded_cache::budget_from_env(root)?,
)?)
.registered();
// **One descriptor on the blob file, not three.** The writer opened it
// read/write and `flock`ed it; the absorber's `pread` fallback and the
// index workers' `read_exact_at` both address it positionally, so they
// take the same open file description rather than two more of their
// own. 5 descriptors per open store → 3, which is the divisor of
// gunnar's `open_repo_budget` (`tests/fd_per_store.rs`).
let absorber = Arc::new(Absorber::<S>::open(
&blobs,
writer.blobs_shared(),
hash,
objects,
exploded,
)?);
// **The `indexed` bit, derived.** Every extent this archive has ever
// acked, diffed against the rows the index holds. `SafeWriter::create`
// above appends to that journal and never truncates it, which is what
// makes the older extents still be here to diff against.
//
// An arm with no journal has nothing to derive the bit from, and that is
// the honest consequence of choosing it rather than a gap: a
// `FastWriter` store has no durable record that a pack was acked, so a
// reopen re-queues nothing and there is nothing for it to re-queue.
let acked = match journal.as_deref() {
Some(p) if p.exists() => crate::archive_write::read_journal(p).with_context(|| {
format!(
"reading {} — it is the durable half of the indexed bit and a store \
cannot be opened without knowing which packs it owes index work for",
p.display()
)
})?,
_ => Vec::new(),
};
let requeue = absorber.adopt_journal(&acked)?;
// **The wiring.** The absorber goes into the push path, which hands it to
// every account indexer it starts, so the drain behind `push_pack` ends
// in object rows instead of one pack row. Nothing about the ack changes:
// `append` still returns after the two fsyncs and this is all downstream
// of that (§13.9).
let push = PushPath::with_absorber(
writer,
&blobs,
journal,
absorber.clone() as Arc<dyn ObjectAbsorb>,
)?;
let refs = RefLog::new(root.join("refs.log"));
let archive = root.join("repository.znippy");
let store = Self {
root: root.to_path_buf(),
blobs,
archive,
push,
account: account.to_string(),
hash,
absorber,
refs,
ref_gate: Mutex::new(()),
gc: arms.gc.create(),
arms,
};
// The interrupted work, back on the channel it fell off. Nothing reads
// here and nothing blocks: this is the same 24-byte handoff a push makes,
// and the drain picks it up on its own thread (§13.9-11).
let indexer = store.push.indexer(&store.account);
for p in &requeue {
indexer.submit(IndexJob {
pack_id: p.pack_id,
offset: p.offset,
len: p.len,
})?;
}
// A store reopened over an existing tail already knows its objects; the
// derived tables are rebuilt from them rather than from a side-car,
// because a side-car is a second truth.
store.refold()?;
Ok(store)
}
/// The arms this store was built with. A description of what was
/// constructed, not a switch — nothing on a serving path reads it.
pub fn arms(&self) -> StoreConfig {
self.arms
}
/// The selected writer's name, as it goes on a bench row.
pub fn writer_name(&self) -> &'static str {
self.push.name()
}
/// What the selected writer's `append` actually promises. Printed next to a
/// throughput, because a throughput without it is not a comparison.
pub fn writer_durability(&self) -> &'static str {
self.push.durability()
}
/// The selected [`Gc`] arm.
pub(crate) fn gc_arm(&self) -> &dyn Gc {
self.gc.as_ref()
}
/// Where the verbatim packs are.
pub fn blobs_path(&self) -> &Path {
&self.blobs
}
/// The sealed archive GC operates on.
pub fn archive_path(&self) -> &Path {
&self.archive
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn hash_kind(&self) -> GitHashKind {
self.hash
}
/// The `objects` table, for a caller that wants the index directly (the
/// bench does).
pub fn index(&self) -> &ObjectReadStack<S> {
&self.absorber.objects
}
/// Objects in the `objects` table.
pub fn object_count(&self) -> usize {
self.absorber.objects.len()
}
/// Commits in the graph.
pub fn commit_count(&self) -> usize {
self.absorber
.derived
.read()
.expect("derived lock")
.graph
.len()
}
/// `(graph_read_from_disk, graph_recomputed)` since this store was opened —
/// how many folds read the persisted `objects.graph` vs re-derived it.
pub fn graph_source_stats(&self) -> (u64, u64) {
self.absorber.graph_source_stats()
}
/// `(reach_carried, reach_rebuilt)` since this store was opened — how many
/// folds carried the reachability table across (renumbered, no tree
/// opened) vs how many reads built it from scratch. The witness for the
/// clone-after-push lever; see `REACH_CARRY_MAX_FRACTION`.
pub fn reach_source_stats(&self) -> (u64, u64) {
self.absorber.reach_source_stats()
}
/// `(folded_rows, tail_rows)` of the ordinal space the last fold built — how
/// many oids came off the persisted index's oid column and how many off a
/// scan of the redb tail. Replaces S-082's `(read, recomputed)`: there is no
/// memo to read any more, the index IS the space.
pub fn ordinal_source_stats(&self) -> (u64, u64) {
self.absorber.ordinal_source_stats()
}
/// The folded ordinal space — every oid in ordinal order, hex. For the tests
/// that assert a memo read reproduces the scanned space exactly (same oids,
/// same order, same ordinals — the ordinal is the index into this vector).
///
/// Formatted on demand from `oids_raw`: there is no resident hex `Vec` any
/// more (S-082), and a test that wants hex encodes the flat buffer here
/// rather than the store carrying a second copy for it.
#[cfg(test)]
pub(crate) fn ordinal_oids(&self) -> Vec<String> {
let oid_len = self.hash.oid_len();
self.absorber
.derived
.read()
.expect("derived lock")
.ordinals
.view()
.concat()
.chunks_exact(oid_len)
.map(hex::encode)
.collect()
}
/// **One object's resolved content** — §14's exploded table, in one lookup.
///
/// Not one of the twelve and deliberately not on [`GitOps`]: the twelve are
/// the wire vocabulary and they hand back **stored** bytes ([`GitOps::get`]),
/// which for a delta entry is a delta. This is the other question — *what
/// does this object contain* — and it is the one §14's table exists to make
/// a point lookup instead of a chain walk.
///
/// It absorbs any pending pack first, for the same reason
/// [`GitOps::has`] does: a store with durable bytes it has not indexed cannot
/// answer "absent" without lying.
pub fn content(&self, oid: Oid<'_>) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
// No absorb here any more: a pack the worker has not reached is read
// for this one oid by `Absorber::resolved`'s third arm — on this thread,
// from the mmapped pack, without waiting for the worker.
self.absorber.resolved(oid)
}
/// **One object, with the bill for producing it.**
///
/// [`Self::content`] plus what the read cost: `None` for a cost when the
/// exploded pools answered (nothing was rebuilt), `Some(cost)` when the
/// object was inflated out of the pack. What a measurement needs and a
/// server does not — the cache admission decision is made inside, not by
/// the caller.
pub fn content_costed(
&self,
oid: Oid<'_>,
) -> Result<Option<(GitObjectKind, Vec<u8>, Option<crate::resolve::ReadCost>)>> {
self.absorber.resolved_costed(oid)
}
/// §14's table, counted: rows, rows written, reads it served, reads that fell
/// through to re-deriving from the verbatim truth.
///
/// Applied output. Both content paths return identical bytes, so `served`
/// against `rederived` is the only thing that can say which one ran.
pub fn exploded_stats(&self) -> ExplodedStats {
self.absorber.exploded.engine_stats()
}
/// **What the exploded tier is holding and what it is allowed to hold** —
/// the row a dashboard prints (`explode 3.1G/8G`) and `Admin.StorageInfo`
/// carries, per repository.
pub fn explode_cache_use(&self) -> crate::exploded_cache::CacheUse {
self.absorber.exploded.use_now()
}
/// Rows in the EAGER pool only — the graph the push wrote, with nothing a
/// reader has since cached. What `adopt_journal` compares against the index
/// and what a test asking "did the push explode the graph" must ask,
/// because [`Self::exploded_stats`]'s `rows` counts both pools and a read
/// moves it.
pub fn exploded_eager_rows(&self) -> Result<u64> {
self.absorber.exploded.eager_rows()
}
/// **Set this repository's explode budget** — what the control plane calls
/// when an account's quota changes. Takes effect at the next rotation; it
/// never refuses a read, never refuses a write and never deletes on the
/// caller's thread. A bigger budget buys a bigger cache and therefore faster
/// first reads; a smaller one costs re-inflation, never an error.
pub fn set_explode_budget(&self, bytes: u64) {
self.absorber.exploded.set_budget(bytes);
}
/// Flush the lazy pool, rotate it and evict it back under budget — on
/// **this** thread. What a test and a `gc` call; a reader never does, it
/// starts a write-back thread and walks away.
pub fn explode_cache_maintain(&self) -> Result<u64> {
self.absorber.exploded.maintain()
}
/// **`repo gc` reclaims the cache**: every lazy segment unlinked, the graph
/// pool kept. Returns the bytes that left the filesystem.
pub fn drop_explode_cache(&self) -> Result<u64> {
self.absorber.exploded.drop_cache()
}
/// Where §14's table lives. Deleting this file while no store holds it is the
/// sanctioned way to drop it; the next open re-explodes every pack.
pub fn exploded_path(&self) -> &Path {
self.absorber.exploded.path()
}
/// Every live row of one kind, with its payload — what the graph fold reads,
/// exposed so a measurement can ask **the table** how many bytes it holds
/// rather than dividing a file size by what the pack claimed.
pub fn exploded_of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
self.absorber.exploded.of_kind(kind)
}
/// The commit graph as it stands, generations included. A snapshot: the
/// caller holds no lock and the next fold replaces it wholesale.
pub fn graph_snapshot(&self) -> Vec<CommitNode> {
self.absorber
.derived
.read()
.expect("derived lock")
.graph
.clone()
}
/// Packs whose objects are not in the index yet — §13.12's bit, counted.
pub fn unindexed_packs(&self) -> usize {
self.absorber.unindexed_packs()
}
/// This account's indexer: the thread that drains the channel a push queues
/// on. `is_indexed(pack_id)` on it is §13.12's bit at the pack level, and it
/// is set only once that pack's **objects** are in the index.
pub fn indexer(&self) -> std::sync::Arc<crate::indexer::AccountIndexer> {
self.push.indexer(&self.account)
}
/// Block until the background drain has absorbed everything this store has
/// pushed. What a maintenance tick, a seal or a test uses instead of racing
/// it.
pub fn wait_indexed(&self) {
self.push.pool().wait_caught_up();
}
/// **Stop this store's index work — the repository is being deleted.**
///
/// Sets the cancel flag the resolver polls every few thousand objects and
/// the drain checks before every pack. The pack in flight is **aborted at
/// its next checkpoint** (milliseconds), not finished: its rows are keyed
/// by oid and the exploded tier is droppable, and the whole directory is
/// about to be removed anyway. Every pack still queued is skipped. After
/// this the store can be dropped without the drop joining a worker that is
/// hours into a kernel pack — which is what `Drop for AccountIndexer` would
/// otherwise do.
pub fn cancel_index(&self) {
self.absorber.cancel.store(true, std::sync::atomic::Ordering::SeqCst);
self.push.pool().cancel_all();
}
/// Put a different memory watchdog under this store's worker — a test's
/// fake `/proc/meminfo`. Takes effect from the next pack.
pub fn set_mem_watchdog(&self, dog: Arc<crate::memguard::MemWatchdog>) {
*self.absorber.watchdog.write().unwrap_or_else(|e| e.into_inner()) = dog;
}
/// Has [`cancel_index`](Self::cancel_index) been called?
pub fn index_cancelled(&self) -> bool {
self.absorber.cancel.load(std::sync::atomic::Ordering::SeqCst)
}
/// This store's index progress counter, for a dashboard or a stats call.
pub fn index_progress(&self) -> crate::progress::Snapshot {
self.absorber.progress.snapshot()
}
/// **The indexer's half of the split**, and not one of the twelve.
///
/// Runs on the account indexer's worker (§13.9), and on a read that arrives
/// before that worker got there — the same function either way, which is why
/// there is no such thing as a background index that behaves differently from
/// the fallback one.
pub fn absorb_pending(&self) -> Result<usize> {
self.absorber.absorb_pending()
}
/// **Make every durable object addressable, on this thread, now** —
/// [`git_storage_trait::GitServe::index_now`]'s implementation and the ack
/// path's one call. See [`Absorber::index_pending_now`] for why the
/// connectivity check is the only caller allowed to wait here.
pub fn index_now(&self) -> Result<usize> {
self.absorber.index_pending_now()
}
/// **The packs this store has stopped trying to index**, with the reason
/// each one stopped for.
///
/// Empty is the ordinary answer and the only good one. A non-empty list is
/// a repository that is missing objects whose bytes are durable, and it is
/// the fact an operator needs before wondering why a push was refused or a
/// standby is behind — see [`MAX_ABSORB_ATTEMPTS`].
pub fn unindexable_packs(&self) -> Vec<Unindexable> {
self.absorber.unindexable_packs()
}
/// **The object ids a stored pack introduced**, and not one of the twelve.
///
/// [`GitOps::put_pack`] cannot answer this and never will: the walk it does
/// reads pack *entries*, and a pack entry does not carry its oid — that is
/// why [`crate::pack_walk::Closure`] can only name the `REF_DELTA` bases and
/// says so. The oids exist once the indexer has absorbed the pack, so this
/// absorbs first rather than racing the drain, and is therefore the one
/// caller that deliberately pays §13.9's deferral back.
///
/// **Only call it when something needs the set.** A push whose policy has no
/// gate over objects must not pay this, which is the whole reason it is a
/// separate call and not a field on `TxId`.
///
/// The extent *is* the identity: §14 stores a pack verbatim as one
/// contiguous append, so "introduced by this pack" and "indexed at an offset
/// inside this pack's extent" are the same set, with no pack column to keep
/// in step. Bases the push deltas against live in *other* extents and are
/// excluded by construction — which is what a thin push needs, and what
/// reading a written `.idx` back could never do.
///
/// Cost is one pass over the index, so it is O(objects in the repository),
/// not O(objects pushed). Recorded rather than hidden: a per-pack ordinal
/// range would make it O(pushed), and needs the indexer to keep one.
pub fn oids_in_extent(&self, extent: Extent) -> Result<Vec<Vec<u8>>> {
let (start, len) = extent;
// **Never waits on the indexer.** If the gate is free this thread
// absorbs the pending packs (the same work, done now, by the caller
// that needs it); if the worker holds it, the extent is resolved
// read-only on this thread and its oids are the answer — the same
// resolver, no sink, no rows written.
if self.try_absorb_pending()?.is_none() {
let snap = self.archive_snapshot()?;
let bytes = self.extent(&snap, start, len)?;
let walked = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
let mut oids: Vec<Vec<u8>> = Vec::with_capacity(walked.entries.len());
let limits = crate::resolve::ResolveLimits {
base_budget: (crate::resolve::base_budget_from_env() / 4).max(1),
..crate::resolve::ResolveLimits::none()
};
crate::resolve::resolve_streaming(
&bytes,
&walked,
self.hash,
start,
&*self.absorber,
&crate::exploded::NoSink,
limits,
&mut |rows| {
oids.extend(rows.drain(..).map(|r| r.oid));
Ok(())
},
)
.context("resolving the pushed pack read-only to enumerate its objects")?;
return Ok(oids);
}
let end = start.saturating_add(len);
let all = self.index().oids_in_order()?;
let refs: Vec<&[u8]> = all.iter().map(Vec::as_slice).collect();
let extents = self.index().extents_batch(&refs);
Ok(all
.into_iter()
.zip(extents)
.filter_map(|(oid, at)| match at {
Some((offset, _)) if offset >= start && offset < end => Some(oid),
_ => None,
})
.collect())
}
/// **Everything needed to emit a pack for EXACTLY `oids`** — nothing added,
/// nothing dropped.
///
/// Not one of the twelve. It sits beside [`absorb_pending`](Self::absorb_pending)
/// for the same reason those do: it is about how storage is *shaped*, not
/// about what a caller stores or reads.
///
/// # 🔴 It used to add the delta bases, and that shipped broken clones
///
/// Until 2026-08-11 this closed the request over its delta bases: an
/// `OFS_DELTA` names its base by position, so a base outside the set cannot
/// be encoded, and *adding the base* looked like the answer that is always
/// safe. It is not, and the failure is silent on the server:
///
/// ```text
/// git clone --bare --single-branch --branch base <url>
/// fatal: did not receive expected object 8601ec33920b7d701c9887fa04916501136d6e90
/// fatal: fetch-pack: invalid index-pack output
/// event="git.upload_pack.served" objects=33773 ← the server said SUCCESS
/// ```
///
/// Measured on the `h2h-linear-sha1-2048c-1024f-16k` fixture: `base` reaches
/// **31 805** objects, the delta-base closure added **1 968** more — 33 773,
/// exactly the count the server logged — and 507 of those additions were
/// **trees**. A tree that is not reachable from the wants drags its own
/// children in as a *requirement*, because `git index-pack
/// --check-self-contained-and-connected` (which is what a clone runs, and
/// which sets `strict`) walks every received object's links and then demands
/// each one exist. Those 507 trees named **204** objects the pack did not
/// contain. `8601ec33…` is one of them.
///
/// So a delta base pulled into a narrowed request is not free: it satisfies
/// the *pack format* and violates *connectivity*. Closing over the added
/// bases' children in turn does terminate — the same fixture reaches a fixed
/// point at 34 534 objects — but it is an over-send by construction, and on
/// a `--filter=blob:none` fetch it adds back exactly the blobs the filter
/// excluded, which is the defect
/// `emit_pack_emits_exactly_the_set_it_is_given_and_never_its_closure`
/// exists to forbid.
///
/// # What it does instead: the base decides how the entry is encoded
///
/// The set is never widened. For each requested object:
///
/// * **base inside the request** → the stored entry is copied byte for byte,
/// which is the whole point of this engine and is what a full clone does
/// for every single object;
/// * **base outside the request** → the entry is **rebuilt** from §14's
/// resolved content and marked
/// [`EmitEntry::recompressed`](crate::pack_walk::EmitEntry::recompressed) —
/// as a computed delta against a base the request *does* carry where one
/// can be found (see [`Self::in_request_chain_ancestor`] and
/// [`crate::delta`]), and whole where it cannot.
///
/// That is what stock `pack-objects` does — it reuses a stored delta only
/// when the base is also being packed — and it keeps the receipt a fact:
/// `copied` and `recompressed` add up to the objects written, and
/// `recompressed` is **0 for a whole-repository clone** because such a
/// selection contains every base. The cost lands on the boundary a narrowed
/// request cuts through and nowhere else: 1 968 of 33 773 entries, 5.9 %, on
/// the fixture above.
///
/// `ofs_delta_ok` is the client's `ofs-delta` capability. An old client that
/// cannot parse one is not refused any more: the entry is re-headed as a
/// `REF_DELTA` naming its base by oid, **carrying the same compressed delta
/// payload**, so that arm stays a copy too. Only a base outside the request
/// costs a rebuild.
///
/// # `thin_haves` — the boundary entry that does NOT have to be rebuilt
///
/// `Some(tips)` says the client consented to a thin pack *and* the
/// negotiation found common ground; [`crate::serve::GitStore::emit_oids`] is
/// where the two halves are required together, because either alone is a
/// failed fetch rather than a smaller one. Given it, a base outside the
/// request is checked against [`the client's own objects`](GitStore::client_bases),
/// and one the client already holds is named by oid in a `REF_DELTA` that
/// carries the **stored delta stream unchanged**. That entry counts as
/// `copied`, because it is: nothing was inflated and nothing was deflated.
///
/// `None` — a clone, or a client that did not ask for a thin pack — behaves
/// exactly as before, and a **full clone is byte-identical either way**,
/// because a whole-repository request has no base outside itself for any of
/// this to apply to.
///
/// It does nothing for the *clone* half of the boundary cost, and cannot: a
/// clone's receiver holds nothing, so there is no external base to name.
/// That half is the arm below it — [`crate::delta`] computes a delta against
/// something the pack **does** carry, which is the only move left when there
/// is no receiver to lean on. The two do not overlap: `thin_haves` reaches a
/// boundary entry first and costs nothing when it fires, so a fetch stays a
/// pure copy and only a clone pays for a computed delta.
///
/// # An oid this repository does not hold is REFUSED
///
/// It used to be skipped, which is the same silent-under-send shape as the
/// closure defect above wearing a different hat: N asked, N-1 emitted,
/// `PackStats` reporting success, and the client the first to know. gunnar's
/// in-memory arm has always refused it, so the skip also meant the two
/// engines behind one contract disagreed about what a missing object means.
/// A partial-clone filter does not depend on the skip and cannot: a filter
/// *removes* oids from the selection, it never adds one the store lacks.
///
/// # Cost
///
/// O(objects requested). The previous shape read `oids_in_order()` and
/// `lookup_batch` over the **whole repository** on every call, to build an
/// offset→oid map it needed only for the closure; with no closure the
/// selection's own offsets answer "is this base in the request", so that
/// scan is gone and no reverse index has to exist.
pub fn emit_set(
&self,
oids: &[Oid<'_>],
ofs_delta_ok: bool,
thin_haves: Option<&[Oid<'_>]>,
) -> Result<Vec<crate::pack_walk::EmitEntry>> {
use crate::index_layout::ObjType;
use std::collections::hash_map::Entry;
// If the gate is free, absorb now; if the worker holds it, the rows
// come from the index plus a read-only pass over the pending packs
// (`lookup_batch_never_wait` below). Never a wait.
let _ = self
.try_absorb_pending()
.context("absorbing pending index work before emitting a pack")?;
// Deduplicate, preserving the caller's order. A repeated oid in the
// request must not become a repeated entry in the pack.
//
// **A map and not a set**, keyed oid → position in `asked` (which is the
// position in `rows`). The membership question is the one it always was;
// what the position buys is the base's *archive offset* for a
// `REF_DELTA`, which names its base by oid and can learn where that base
// sits no other way. It is a `usize` per requested object on a vector
// this function already builds — not a second pass and not a second
// index lookup.
let mut asked: Vec<&[u8]> = Vec::with_capacity(oids.len());
let mut seen: HashMap<&[u8], usize> = HashMap::with_capacity(oids.len());
for oid in oids {
if let Entry::Vacant(slot) = seen.entry(*oid) {
slot.insert(asked.len());
asked.push(*oid);
}
}
let rows = self.lookup_batch_never_wait(&asked)?;
// **One mapping for the whole emit set.** Only the header probes below
// read through it — the entries themselves are recorded as extents and
// resolved by [`Self::resolve_emit_payloads`], which takes its own
// snapshot of the same [`ArchiveMap`](crate::archive_map::ArchiveMap) and
// therefore the same `Arc` unless a push landed in between.
let snap = self.archive_snapshot()?;
// The two questions "is this base in the request" are asked in the two
// coordinate systems the two delta forms use, and both are answered off
// the request itself rather than off the index.
//
// **Offset → position in the request**, not a bare set of offsets. The
// membership question is the same one it always was; what the position
// buys is the base's *oid*, which the re-delta arm below needs in order
// to read the base's content. It is the same single pass over `rows`
// that built the set, so it costs a `usize` per requested object and no
// extra work — and it is emphatically not the whole-repository
// offset→oid scan the old delta-base closure used to do.
let in_set_offsets: HashMap<u64, usize> = rows
.iter()
.enumerate()
.filter_map(|(i, r)| r.as_ref().map(|r| (r.offset, i)))
.collect();
let oid_len = self.hash_kind().oid_len();
// The base of the last boundary entry that was re-deltified, kept so a
// run of entries hanging off one ancestor inflates it once. One slot and
// not an LRU: measured on `h2h-linear-sha1-2048c-1024f-16k`, 961
// boundary entries name 855 distinct stored bases, so there is no reuse
// worth a cache that could hold a repository's worth of objects — and
// this crate is under a live RSS budget.
let mut base_content: Option<(u64, Vec<u8>)> = None;
// What the client already holds, built **lazily and at most once**. A
// clone never asks for it (`thin_haves` is `None`), and a fetch every one
// of whose bases is inside the request never reaches the line that builds
// it either — only a request that actually cuts a delta chain pays, which
// is the only request it can help.
let mut client_bases: Option<ExternalBases> = None;
let mut out = Vec::with_capacity(asked.len());
for (oid, row) in asked.iter().zip(rows.iter()) {
// **An oid this repository does not hold is REFUSED, not dropped.**
//
// This used to `continue`, and that is the same silent-under-send
// shape as the closure defect above: the request asks for N objects,
// the pack carries N-1, the receipt says success and the client is
// the first to find out. It also made the two engines disagree about
// what a missing object *means* — gunnar's in-memory arm has always
// said *"emit_pack was asked for X, which this store does not hold;
// a pack emitter that skipped it would send a pack no client can
// close"* — so a benchmark across the two was comparing two
// contracts.
//
// No legitimate caller relies on the skip. Every set that reaches
// here is built out of this store's own index: `select` answers from
// the reachability bitmaps, and gunnar's own walk resolves each id
// against the store before it selects it. A partial-clone filter is
// no exception and is the case worth naming, because it is the one
// that *looks* like it might be: `--filter=blob:none` **removes**
// oids from the selection, it never adds one the store lacks, so a
// filtered request is a smaller set of objects that are all present
// and this refusal never fires on it.
let Some(row) = row.as_ref() else {
bail!(
"this pack was asked for {}, which this repository does not hold; emitting \
the rest would be a short pack reported as a success, so this refuses \
instead",
hex::encode(oid)
);
};
// Which base this entry names, if any, and whether the request holds
// it. A `REF_DELTA`'s `delta_base` column is 0 by construction —
// `DeltaBase::as_offset` reports 0 for a ref base and 0 is also the
// no-base sentinel — so its base is read out of the entry's own
// bytes, directly after the type/size varint.
//
// A base that is *outside* the request is carried in **both**
// coordinate systems rather than as a bare `false`, because that is
// what the thin arm below needs: an `OFS_DELTA` knows only where its
// base sits, a `REF_DELTA` knows only what it is called, and
// "does the client hold it" has to be asked in whichever one the
// entry speaks.
let base = match row.obj_type {
ObjType::OfsDelta => match in_set_offsets.get(&row.delta_base) {
Some(_) => BaseOf::Inside {
at: row.delta_base,
},
None => BaseOf::Outside {
offset: row.delta_base,
oid: None,
},
},
ObjType::RefDelta => {
let head = self.extent(&snap, row.offset, row.len.min(64))?;
let (_, _, n) = crate::pack_walk::type_and_size_of(&head)?;
match head.get(n..n + oid_len) {
// In the request — **and where**. `rows[i]` is `None`
// only for an oid this store does not hold, which the
// refusal above turns into an error the moment the loop
// reaches it; until then the base is treated as outside,
// which is the answer that cannot be wrong.
Some(b) => match seen
.get(b)
.and_then(|i| rows[*i].as_ref())
.map(|r| r.offset)
{
Some(at) => BaseOf::Inside { at },
None => BaseOf::Outside {
offset: 0,
oid: Some(b.to_vec()),
},
},
// A truncated entry names no base this can read. It is
// not "inside", and it is not offered to the client
// either — it falls through to the whole rebuild, which
// is the answer that cannot be wrong.
None => BaseOf::Outside {
offset: 0,
oid: None,
},
}
}
_ => BaseOf::Whole,
};
if let BaseOf::Whole | BaseOf::Inside { .. } = base {
let inside_at = match base {
BaseOf::Inside { at } => Some(at),
_ => None,
};
// **The whole-clone arm, and it reads not one byte here.** The
// entry is recorded as its address in the archive; the bytes are
// resolved against the mapping at emit time and go straight from
// the page cache to the wire. This line used to be a `pread`
// into a fresh `Vec` — 13.8 M of each on a `linux.git` clone, all
// of them held at once, which is the 2314 MB peak RSS
// [`crate::pack_walk::EntryBytes`] documents.
let extent = crate::pack_walk::EntryBytes::Extent {
offset: row.offset,
len: row.len,
};
let (stored, obj_type, delta_base) = match (row.obj_type, inside_at) {
// An `OFS_DELTA` for a client that cannot read one: re-head it
// as a `REF_DELTA` naming the same base by oid. The compressed
// payload is the delta stream either way, so this is still a
// copy in the `recompressed` sense — nothing is inflated — but
// the concatenation of a new header and an old payload exists
// nowhere on disk, so it cannot be an extent and is `Owned`.
(ObjType::OfsDelta, Some(_)) if !ofs_delta_ok => {
let base_oid = self.oid_at_offset(row.delta_base, &asked, &rows)?;
let bytes = self.extent(&snap, row.offset, row.len)?;
(
crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(
&bytes, &base_oid,
)?),
ObjType::RefDelta,
0,
)
}
// 🚨 **A stored `REF_DELTA` whose base is IN THIS PACK is
// re-headed as an `OFS_DELTA`, and that is not an
// optimisation.**
//
// gitoxide — which is `gunnar_client`, `gunnar`'s own
// `receive-pack`, and every other gix-based receiver — reads
// every `OBJ_REF_DELTA` in an incoming pack as naming an
// object *the receiver already has*.
// `gix_pack::data::input::LookupRefDeltaObjectsIter` consults
// the local object database and the bases it has already
// spliced in, and **nowhere else**: never inside the pack it
// is reading. A base that is in the pack and not in the
// receiver's store is `input::Error::NotFound`, and the fetch
// dies with `The object <base> could not be decoded or wasn't
// found` — with the whole pack rejected, on a clone whose
// target is empty by definition, so *every* such entry is
// fatal. That is `S-003` / gitoxide#2882, and `gunnar-wire`'s
// `Selection::client_has` carries the same rule for gunnar's
// own emitter.
//
// Stock git resolves it happily, which is why this survived
// five green `git clone` runs against the store that could
// not serve one `gunnar_client` fetch (`gunnar.rewrite`
// round 0 and `gunnar.year` week 0, all four znippy columns,
// 2026-08-14 — the server logged the same request `served`).
//
// The entry keeps its compressed delta stream byte for byte;
// only the header changes, from "base named by oid" to "base
// named by distance", which is the naming every receiver in
// existence resolves inside the pack. This is where such an
// entry comes from in the first place: `git index-pack
// --fix-thin` appends the base into the pack and leaves the
// ref-delta naming it by oid, so any repository pushed to
// more than once holds them.
//
// # What it costs, stated rather than glossed
//
// `Owned`, so this is the one arm that gives up the zero-copy
// property for an entry it does **not** rebuild: the
// compressed delta stream is memcpy'd once, because a new
// header concatenated with an old payload exists nowhere on
// disk and no extent can address it. Bounded by the
// compressed size of the ref-deltas whose base is in the
// request — not by the repository, and not by the pack.
//
// It could be avoided: `emit_pack` re-encodes every header
// from `EmitEntry::obj_type` while `header_len` finds the
// payload from the STORED bytes' own type, so an `Extent`
// over the untouched ref-delta declared as an `OfsDelta`
// would emit the identical pack with no copy at all. That is
// deliberately not done: it makes an `EmitEntry` disagree
// with the bytes it points at, and the entry model being
// trustworthy is worth more than the memcpy. If the RSS
// budget ever says otherwise, the honest form is a third
// `EntryBytes` variant that carries the new header beside
// the extent, not a silent disagreement.
(ObjType::RefDelta, Some(at)) if ofs_delta_ok => {
let bytes = self.extent(&snap, row.offset, row.len)?;
(
crate::pack_walk::EntryBytes::Owned(Self::as_ofs_delta(
&bytes, oid_len, row.offset, at,
)?),
ObjType::OfsDelta,
at,
)
}
// Everything else copies. A `REF_DELTA` left as one here is
// the `!ofs_delta_ok` client, which cannot be handed a
// distance at all — and which is stock git older than 1.4.4,
// never a gix receiver, since gix has always advertised
// `ofs-delta`.
_ => (extent, row.obj_type, row.delta_base),
};
out.push(crate::pack_walk::EmitEntry {
oid: oid.to_vec(),
stored,
obj_type,
uncompressed_size: row.uncompressed_size,
delta_base,
offset: row.offset,
recompressed: false,
deltified: false,
});
continue;
}
// ── the base is outside the request ──────────────────────────────
//
// **Thin first.** If the client consented to a thin pack and the
// negotiation vouches that it already holds this base, the entry
// goes out as a `REF_DELTA` naming that base by oid, carrying the
// **same compressed delta stream**. Nothing is inflated, nothing is
// deflated, and the pack is smaller than the whole object by exactly
// the margin the delta was worth. That is what `caps.thin` has always
// been the allowance for.
//
// An `OFS_DELTA` cannot survive this: it names its base by a
// backwards distance *within the pack*, and the base is not in the
// pack. Re-heading it as a `REF_DELTA` is the same header swap the
// no-`ofs-delta` client path does, over an unchanged payload.
if let Some(haves) = thin_haves {
if client_bases.is_none() {
client_bases = Some(self.client_bases(haves)?);
}
let ext = client_bases
.as_ref()
.expect("the client's bases were just built");
let base_oid: Option<Vec<u8>> = match &base {
// Unreachable: the arm above `continue`s on both. Matched
// rather than `unreachable!()` so that a future arm falling
// through here declines the thin path instead of panicking
// inside a serving thread.
BaseOf::Whole | BaseOf::Inside { .. } => None,
BaseOf::Outside {
oid: Some(named), ..
} => ext
.held
.contains(&OidKey::new(named)?)
.then(|| named.clone()),
BaseOf::Outside {
offset,
oid: None,
} => ext.by_offset.get(offset).map(|k| k.as_slice().to_vec()),
};
if let Some(base_oid) = base_oid {
// A stored `REF_DELTA` already names its base by oid, so its
// bytes go out untouched and stay an extent. An `OFS_DELTA`
// has to be re-headed, and the result is new bytes.
let stored = if row.obj_type == ObjType::OfsDelta {
let bytes = self.extent(&snap, row.offset, row.len)?;
crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(&bytes, &base_oid)?)
} else {
crate::pack_walk::EntryBytes::Extent {
offset: row.offset,
len: row.len,
}
};
out.push(crate::pack_walk::EmitEntry {
oid: oid.to_vec(),
stored,
obj_type: ObjType::RefDelta,
uncompressed_size: row.uncompressed_size,
// The base is NOT in this pack, so nothing may be ordered
// after it: `topological_order` reads 0 as "names no base
// in here", which for an external base is the truth.
delta_base: 0,
offset: row.offset,
// Copied. The payload never left the archive's bytes.
recompressed: false,
deltified: false,
});
continue;
}
}
// The base is not being sent and the client cannot supply it, so the
// stored bytes cannot go out as they are. Two answers remain, and
// the first is worth ~4× the second on the wire.
let (kind, body) = self.content(oid)?.ok_or_else(|| {
anyhow!(
"{} is a delta whose base is outside this request, so it must be sent whole, \
and this repository cannot produce its content — neither §14's exploded \
table nor a re-derivation from the verbatim packs answered",
hex::encode(oid)
)
})?;
// ── re-delta against a base the request DOES carry ────────────────
//
// The candidate comes from the entry's own stored delta chain: the
// nearest ancestor that is inside the request. That is the packer's
// original similarity judgement, already made and already written
// down — no window, no sort, no scan of the repository.
let candidate = if crate::delta::enabled() {
self.in_request_chain_ancestor(row.delta_base, row.offset, &in_set_offsets)?
} else {
None
};
if let Some(base_at) = candidate {
let at = *in_set_offsets
.get(&base_at)
.expect("the ancestor was found by lookup in this very map");
// Reuse the last base if this entry hangs off the same one.
if base_content.as_ref().is_none_or(|(o, _)| *o != base_at) {
let base_oid = asked[at];
let (_, base_body) = self.content(base_oid)?.ok_or_else(|| {
anyhow!(
"{} is in this request and is the delta base chosen for {}, and this \
repository cannot produce its content",
hex::encode(base_oid),
hex::encode(oid)
)
})?;
base_content = Some((base_at, base_body));
}
let (_, base_body) = base_content.as_ref().expect("just filled");
// `None` is the encoder declining — an unrelated base, or a
// delta that came out no smaller than the object. Falling
// through to the whole rebuild is then strictly right.
if let Some(d) = crate::delta::delta(base_body, &body)? {
let stored = Self::ofs_delta_entry(&d, row.offset, base_at)?;
// A client with no `ofs-delta` is served the same computed
// delta under a `REF_DELTA` head naming the base by oid —
// the identical header swap the base-inside arm above does,
// over a payload that is already built.
// **The `Owned` exception, in its primary shape.** These
// bytes were computed here and exist on no disk, so no
// extent addresses them; carrying them on the entry is
// correct and is bounded by the boundary a narrowed request
// cuts through (961 of 33 773 entries on the fixture below),
// never by the repository.
let (stored, obj_type, delta_base) = if ofs_delta_ok {
(
crate::pack_walk::EntryBytes::Owned(stored),
ObjType::OfsDelta,
base_at,
)
} else {
(
crate::pack_walk::EntryBytes::Owned(Self::as_ref_delta(
&stored, asked[at],
)?),
ObjType::RefDelta,
0,
)
};
out.push(crate::pack_walk::EmitEntry {
oid: oid.to_vec(),
stored,
obj_type,
uncompressed_size: row.uncompressed_size,
delta_base,
offset: row.offset,
// Inflated and re-deflated, so NOT a copy — and the
// receipt says which of the two rebuilds ran.
recompressed: true,
deltified: true,
});
continue;
}
}
// Ship the object whole rather than dragging an unreachable base —
// and its children — into a clone.
out.push(crate::pack_walk::EmitEntry {
oid: oid.to_vec(),
// Inflated and re-deflated from §14's resolved content: the
// `Owned` exception again, and again bounded by the boundary.
stored: crate::pack_walk::EntryBytes::Owned(Self::whole_entry(kind, &body)?),
obj_type: crate::serve::resolved_type(kind),
uncompressed_size: body.len() as u64,
// Whole: it names no base, and nothing may order it after one.
delta_base: 0,
offset: row.offset,
recompressed: true,
deltified: false,
});
}
Ok(out)
}
/// **PHASE 1 of emitting: resolve every entry's stored bytes, in parallel.**
///
/// Emission splits in two, and the split is not a preference — it is where
/// the ordering dependency actually is:
///
/// | phase | what it does | why |
/// |---|---|---|
/// | 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** |
/// | 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 |
///
/// Phase 2 is not parallelisable and no attempt is made: the distance chain
/// is a genuine serial dependency, and a running sha1 is another.
///
/// # No lock, and why that is a fact rather than an intention
///
/// [`Self::adopt_journal`] states the storage property this rests on — *"the
/// blob file is append-only and `gc` truncates nothing, so a retired pack's
/// bytes are still exactly where the journal says they are"*. Every byte a
/// snapshot covers is therefore immutable for the life of the file, so the
/// workers below share one `&Mapped` and read it concurrently with no
/// synchronisation of any kind. There is no mutex here; if one were needed
/// the data model would be wrong.
///
/// # 🚨 This is a fan-out inside the serving tier, and `crate::serve`'s
/// header says that tier is serial
///
/// It did, and the reason it gave is still true and still respected: the
/// constellation's fan-out primitive has no reentrancy detection, so an
/// unbounded fan-out per request inside an admission gate that already runs
/// 32 transfers spawns W² threads. That is why this is **bounded twice**:
///
/// * it does not fan out at all below [`FAN_OUT_AT`] entries, so every fetch
/// small enough for thread-spawn to dominate runs on the calling thread
/// exactly as before, and
/// * it never exceeds [`MAX_RESOLVE_WORKERS`] (override with
/// `ZNIPPY_GIT_EMIT_WORKERS`), so the worst case is 32 × 4 and not
/// 32 × ncores.
///
/// # What the fan-out is worth — stated honestly
///
/// Resolving a mapped extent is a bounds check and a pointer, so the work
/// this parallelises is **not** the resolve itself. It is the two things
/// beside it: the first-touch **page fault** on each entry's header (the
/// grammar is parsed here, which is what forces that fault, and it validates
/// the entry before phase 2 commits a byte to the wire), and the `pread`
/// fallback for any extent the mapping does not cover. On a warm page cache
/// with the whole archive mapped, phase 1 is nearly free and the fan-out
/// buys correspondingly little — the 20 % of cycles this change is aimed at
/// is recovered by *not copying*, not by threads. I could not measure the
/// fan-out's own contribution separately: oden was at loadavg 25.9 while
/// this was written (2026-08-14), and a shared box cannot answer a question
/// that fine.
pub(crate) fn resolve_emit_payloads<'a>(
&self,
entries: &'a [crate::pack_walk::EmitEntry],
snap: &'a crate::archive_map::Mapped,
) -> Result<Vec<std::borrow::Cow<'a, [u8]>>>
where
Self: Sync,
{
let workers = if entries.len() < FAN_OUT_AT {
1
} else {
emit_workers()
};
self.resolve_with(entries, snap, workers)
}
/// [`Self::resolve_emit_payloads`] with the worker count named rather than
/// decided — `1` runs inline on the calling thread.
///
/// Split out so the fan-out is **testable at a size a test fixture can
/// reach**. The production threshold is 4096 entries and the corpus these
/// guards run on holds 2687, so a test that went through the front door
/// would exercise the serial arm every time and the gatling arm never;
/// `the_fan_out_resolves_the_identical_bytes_the_serial_pass_does` calls
/// this with 1 and with 4 over the same set and requires the two to agree
/// byte for byte.
pub(crate) fn resolve_with<'a>(
&self,
entries: &'a [crate::pack_walk::EmitEntry],
snap: &'a crate::archive_map::Mapped,
workers: usize,
) -> Result<Vec<std::borrow::Cow<'a, [u8]>>>
where
Self: Sync,
{
use znippy_zoomies::gatling_forkjoin::gatling_for_each;
let one = |i: usize| -> Result<std::borrow::Cow<'a, [u8]>> {
let e = &entries[i];
let bytes = match &e.stored {
// The bytes live on the entry — the rebuilt-delta exception.
// Borrowed, never cloned: `EntryBytes::Owned` is already the
// only allocation this path makes and copying it would double it.
crate::pack_walk::EntryBytes::Owned(v) => std::borrow::Cow::Borrowed(v.as_slice()),
crate::pack_walk::EntryBytes::Extent { offset, len } => {
self.extent(snap, *offset, *len)?
}
};
// Parse the entry's own header now. Two things fall out: a malformed
// entry is refused **before** phase 2 has written any of the pack,
// and the first cache line of every entry is touched here — on
// whichever worker got it — rather than one at a time on the serial
// writer.
crate::pack_walk::type_and_size_of(&bytes).with_context(|| {
format!(
"resolving the stored bytes of {} for emission",
hex::encode(&e.oid)
)
})?;
Ok(bytes)
};
if workers <= 1 {
return (0..entries.len()).map(one).collect();
}
gatling_for_each(entries.len(), workers, one)
.into_iter()
.collect()
}
/// **Both phases, one call** — the only way a caller in this crate turns an
/// ordered emit set into pack bytes.
///
/// It is three lines, and it is a function rather than three lines because
/// LAW 5's *fix by construction* applies exactly here: the snapshot has to
/// outlive the slices taken from it, and a caller that took its own snapshot
/// per entry, or dropped it between the phases, would be writing a pack out
/// of a mapping that no longer exists. Routing `emit_oids` and every test
/// through one writer makes that impossible to get wrong twice.
pub(crate) fn emit_ordered(
&self,
ordered: &[crate::pack_walk::EmitEntry],
out: &mut dyn std::io::Write,
) -> Result<crate::pack_walk::EmitReport>
where
Self: Sync,
{
let snap = self
.archive_snapshot()
.context("mapping the archive to emit")?;
let payloads = self
.resolve_emit_payloads(ordered, &snap)
.context("resolving the emit set's stored bytes")?;
crate::pack_walk::emit_pack(
ordered,
self.hash_kind(),
out,
&|i| Ok(payloads[i].as_ref()),
)
}
/// **The nearest ancestor of `from`'s stored delta chain that is inside the
/// request**, or `None` when the chain leaves the request and never comes
/// back.
///
/// `from` is the archive offset of a base already known to be *outside* the
/// request; `entry_at` is the offset of the entry that named it.
///
/// # Why this cannot make a cycle, and why that is not a comment but an
/// invariant
///
/// The answer is required to satisfy `answer < entry_at`. In a well-formed
/// pack a base always precedes the delta that names it, so every edge in the
/// stored graph already points from a higher offset to a lower one; a new
/// edge with the same property keeps the emitted graph a DAG **by
/// construction**, whatever else the request contains. That matters because
/// the obvious better candidate — the entry's own delta *children*, one
/// revision away instead of two to nine — is exactly the set that would
/// close a cycle: a child's stored base is this entry, so pointing this
/// entry at the child makes `topological_order` unable to order either, and
/// `emit_pack` refuses the set. Measured cost of the restriction on
/// `h2h-linear-sha1-2048c-1024f-16k`: ~1 470 bytes per boundary entry
/// against ~314 for the cyclic candidate, and 5 800 for shipping it whole.
///
/// An `OFS_DELTA` decreases the offset every hop, so the bound is free
/// there; a `REF_DELTA` names its base by oid and could point anywhere, so
/// the comparison is made rather than assumed.
///
/// # Cost
///
/// One 64-byte `pread` per hop, no inflate, no allocation sized by an
/// object — the same walk and the same ceiling as
/// [`crate::serve::GitStore::resolved_type_at`], for the same reason: a
/// pushed pack's back-references are not under this server's control, so a
/// cycle must terminate rather than spin. Measured hop counts on the fixture
/// above: 2 for 392 of the 859 that resolve, 3 for 260, and 9 at the worst.
fn in_request_chain_ancestor(
&self,
from: u64,
entry_at: u64,
in_request: &HashMap<u64, usize>,
) -> Result<Option<u64>> {
use crate::index_layout::{ObjType, ObjectIndex as _};
/// git's own `pack.depth` ceiling, doubled — the same bound
/// `resolved_type_at` carries and for the same reason.
const MAX_LINKS: usize = 100;
/// Enough of an entry to read its type/size varint plus either an
/// `OFS_DELTA` distance or a `REF_DELTA` oid.
const HEADER_PROBE: u64 = 64;
let oid_len = self.hash_kind().oid_len();
let snap = self.archive_snapshot()?;
let mut at = from;
for _ in 0..MAX_LINKS {
if at != 0 && at < entry_at && in_request.contains_key(&at) {
return Ok(Some(at));
}
if at == 0 {
return Ok(None);
}
// Clamped, not `read_extent`: a short entry at the archive's end
// has fewer than HEADER_PROBE bytes after it (FORGE-SHALLOW).
let head = self.header_probe(&snap, at, HEADER_PROBE)?;
let (t, _, n) = crate::pack_walk::type_and_size_of(&head)?;
at = match t {
// A whole object with no base: the chain ends here, and it was
// not in the request.
ObjType::Commit | ObjType::Tree | ObjType::Blob | ObjType::Tag => {
return Ok(None)
}
ObjType::OfsDelta => {
let (distance, _) = crate::pack_walk::ofs_distance_of(&head[n..])?;
match at.checked_sub(distance) {
Some(next) if next != 0 => next,
// A base before the start of the archive is a corrupt
// back-reference. The whole rebuild is always a correct
// answer, so this declines rather than failing a clone
// over an entry it was only ever going to optimise.
_ => return Ok(None),
}
}
ObjType::RefDelta => {
let Some(base_oid) = head.get(n..n + oid_len) else {
return Ok(None);
};
match self.index().lookup(base_oid) {
Some(row) => row.offset,
None => return Ok(None),
}
}
};
}
Ok(None)
}
/// An `OFS_DELTA` pack entry carrying `d`, deflated.
///
/// The distance written here is the **archive** distance, which is the same
/// coordinate every stored entry's distance is in;
/// [`crate::pack_walk::emit_pack`] rewrites it into output coordinates for
/// every entry it emits, copied or not, so this only has to be well-formed
/// and self-consistent — [`crate::pack_walk::header_len`] parses it back to
/// find where the payload starts.
///
/// `Compression::fast` for the same reason [`Self::whole_entry`] uses it,
/// and the trade is far better here: the buffer being deflated is the delta,
/// which on the measured fixture is ~1.4 KB against the object's 16 KiB, so
/// this path deflates roughly a **tenth** of the bytes the whole rebuild it
/// replaces does.
fn ofs_delta_entry(d: &[u8], entry_at: u64, base_at: u64) -> Result<Vec<u8>> {
use crate::index_layout::ObjType;
use std::io::Write as _;
let distance = entry_at.checked_sub(base_at).ok_or_else(|| {
anyhow!(
"a computed delta at archive offset {entry_at} names a base at {base_at}, which \
is after it — an ofs-delta distance is backwards and this would not encode"
)
})?;
let mut out = Vec::with_capacity(d.len() / 2 + 32);
crate::pack_walk::encode_type_and_size(&mut out, ObjType::OfsDelta, d.len() as u64);
crate::pack_walk::encode_ofs_distance(&mut out, distance);
let mut z = flate2::write::ZlibEncoder::new(out, flate2::Compression::fast());
z.write_all(d)?;
Ok(z.finish()?)
}
/// **The objects the receiver already held before this transfer**, in both
/// coordinate systems a delta can name a base in.
///
/// `haves` is the negotiated common **tips**, exactly as
/// [`git_storage_trait::GitServe::emit_pack`] defines them, and the closure
/// over them is this store's own — the contract deliberately passes tips
/// rather than the walk's exclusion set, because the closure is
/// repository-sized and the engine can rebuild it from its own bitmaps.
///
/// # This is a VOUCHER, and the caller has to be able to stand behind it
///
/// Every object named here may be pointed at by a delta the pack does not
/// carry. If the receiver turns out **not** to hold one, its
/// `index-pack --fix-thin` dies with `pack has N unresolved deltas`, so a
/// caller that cannot vouch must not pass `caps.thin`. The case that matters
/// is a **partial clone**: a `blob:none` client's `have` tips reach blobs it
/// was deliberately never sent, so the closure over-states what it holds and
/// a filtered fetch must not be marked thin.
///
/// # Cost
///
/// One bitmap union plus one point lookup per held object, and it happens at
/// most once per request — and only on a request that has a base outside
/// itself to place. The `by_offset` map exists because an `OFS_DELTA` knows
/// its base only as an archive offset; building it over what the *client*
/// holds rather than over the repository is what keeps this off the
/// whole-repository scan the closure used to do.
///
/// **And it is spent in raw bytes now, end to end.** This used to ask
/// [`GitOps::reachable`] — hex out of the ordinal space, `hex::decode`d
/// straight back to bytes, then a third `to_vec` per object to key the
/// offset map, three heap allocations per held object for a set the size of
/// the client's history. It asks [`GitStore::reachable_raw`] instead: one
/// buffer for the closure, one hash table of [`OidKey`]s over it, and
/// nothing per object.
fn client_bases(&self, haves: &[Oid<'_>]) -> Result<ExternalBases> {
use crate::index_layout::ObjectIndex as _;
// From the last COMMITTED fold if the indexer holds the gate: a closure
// is a fact about immutable commits, and a `have` the fold does not
// know yet contributes nothing — a smaller voucher, the safe direction
// (fewer thin bases, never a base the client lacks). A thin fetch that
// met an index build used to be refused here with "index in progress",
// and gunnar re-emitted without the voucher (GF-4) to get past it.
let held = self.reachable_raw_committed(haves).context(
"closing over the negotiated common tips, to learn which bases the client can supply \
for itself",
)?;
let refs: Vec<&[u8]> = held.iter().collect();
let rows = self.index().lookup_batch(&refs);
let mut by_offset = HashMap::with_capacity(refs.len());
let mut keys = std::collections::HashSet::with_capacity(refs.len());
for (oid, row) in refs.iter().zip(rows.iter()) {
let key = OidKey::new(oid)?;
if let Some(row) = row {
by_offset.insert(row.offset, key);
}
keys.insert(key);
}
Ok(ExternalBases {
by_offset,
held: keys,
})
}
/// The oid of the entry at `offset`, looked up **in the request** — which is
/// the only place it can be, because this is called for a base the request
/// was just shown to contain.
///
/// Linear over the request rather than a map, because it runs only on the
/// `ofs-delta`-less client path: a capability no git since 1.4.4 omits, and
/// one that has to be *advertised absent* to reach here at all.
fn oid_at_offset(
&self,
offset: u64,
asked: &[&[u8]],
rows: &[Option<crate::index_layout::IndexRow>],
) -> Result<Vec<u8>> {
asked
.iter()
.zip(rows)
.find(|(_, r)| r.as_ref().is_some_and(|r| r.offset == offset))
.map(|(oid, _)| oid.to_vec())
.ok_or_else(|| {
anyhow!(
"the entry at archive offset {offset} was checked to be in this request and \
then could not be found in it"
)
})
}
/// Re-head a stored `OFS_DELTA` as a `REF_DELTA` naming `base_oid`.
///
/// The compressed delta stream is copied unchanged; only the header differs,
/// because the two forms differ in **how the base is named** and in nothing
/// else. This is why a client without `ofs-delta` costs bytes and not CPU.
fn as_ref_delta(stored: &[u8], base_oid: &[u8]) -> Result<Vec<u8>> {
use crate::index_layout::ObjType;
let (t, stated_size, n) = crate::pack_walk::type_and_size_of(stored)?;
if t != ObjType::OfsDelta {
bail!("only an ofs-delta can be re-headed as a ref-delta, this entry is {t:?}");
}
let (_, d) = crate::pack_walk::ofs_distance_of(stored.get(n..).unwrap_or(&[]))?;
let mut out = Vec::with_capacity(stored.len() + base_oid.len());
crate::pack_walk::encode_type_and_size(&mut out, ObjType::RefDelta, stated_size);
out.extend_from_slice(base_oid);
out.extend_from_slice(stored.get(n + d..).unwrap_or(&[]));
Ok(out)
}
/// Re-head a stored `REF_DELTA` as an `OFS_DELTA` naming the base that sits
/// at `base_at` — **the inverse of [`Self::as_ref_delta`]**, and the fix for
/// the one pack shape gitoxide cannot read.
///
/// The compressed delta stream is copied unchanged; only the header differs,
/// because the two forms differ in how the base is *named* and in nothing
/// else. The reason this has to exist is in the caller: a `REF_DELTA` whose
/// base is inside the same pack is resolved by stock git and refused by
/// every gix-based receiver.
///
/// # The distance written here is a placeholder, and that is by design
///
/// Exactly as [`Self::ofs_delta_entry`] documents: the distance an
/// `OFS_DELTA` carries is relative to a position in the pack it is *in*, so
/// [`crate::pack_walk::emit_pack`] re-encodes it for every entry from the
/// output offsets it is assigning, and [`crate::pack_walk::header_len`]
/// parses whatever is written here only to find where the payload starts. It
/// therefore has to be well-formed and self-consistent, and nothing more.
///
/// It cannot simply be `entry_at - base_at`: a `REF_DELTA` is allowed to
/// name a base that comes **after** it in the pack, and `git index-pack
/// --fix-thin` produces exactly that — it appends the base at the end. So a
/// forward reference encodes as `1`, and `emit_pack` writes the real
/// backwards distance once `topological_order` has put the base first.
fn as_ofs_delta(
stored: &[u8],
oid_len: usize,
entry_at: u64,
base_at: u64,
) -> Result<Vec<u8>> {
use crate::index_layout::ObjType;
let (t, stated_size, n) = crate::pack_walk::type_and_size_of(stored)?;
if t != ObjType::RefDelta {
bail!("only a ref-delta can be re-headed as an ofs-delta, this entry is {t:?}");
}
let payload = stored.get(n + oid_len..).ok_or_else(|| {
anyhow!(
"a ref-delta entry at archive offset {entry_at} is {} bytes, which is not even its \
header plus a {oid_len}-byte base oid",
stored.len()
)
})?;
let mut out = Vec::with_capacity(payload.len() + 32);
crate::pack_walk::encode_type_and_size(&mut out, ObjType::OfsDelta, stated_size);
crate::pack_walk::encode_ofs_distance(
&mut out,
entry_at.checked_sub(base_at).filter(|d| *d > 0).unwrap_or(1),
);
out.extend_from_slice(payload);
Ok(out)
}
/// A whole (non-delta) pack entry for `body`: type/size header, then the
/// body deflated.
///
/// The **only** place in this crate that deflates on a serving path, and the
/// only reason it exists is a delta whose base the request does not contain.
/// `Compression::fast` deliberately: these bytes are re-compressed because
/// they *cannot* be copied, so the trade is wire size against a clone's
/// latency, and the entries this runs on are a small boundary fraction of any
/// real request.
fn whole_entry(kind: crate::object::GitObjectKind, body: &[u8]) -> Result<Vec<u8>> {
use std::io::Write as _;
let mut out = Vec::with_capacity(body.len() / 2 + 32);
crate::pack_walk::encode_type_and_size(
&mut out,
crate::serve::resolved_type(kind),
body.len() as u64,
);
let mut z = flate2::write::ZlibEncoder::new(out, flate2::Compression::fast());
z.write_all(body)?;
Ok(z.finish()?)
}
/// Take the absorb gate and hold it.
///
/// The only way to observe "a read that arrived before the drain" as a
/// **state** rather than as a race: with this held, the worker is parked at
/// the top of its absorb, so the pack is provably durable, provably queued
/// and provably not indexed. Its guard is in this module's tests, and nothing
/// on a serving path takes it.
#[cfg(test)]
pub(crate) fn hold_absorb_gate(&self) -> std::sync::MutexGuard<'_, ()> {
self.absorber.gate.lock().expect("absorb gate poisoned")
}
/// `pread` an extent out of the verbatim archive, into a fresh `Vec`.
///
/// **The fallback and the owning path.** A caller that can hold a borrow
/// wants [`Self::archive_snapshot`] plus [`Self::extent`] instead: this one
/// costs a syscall, an allocation and a kernel→user copy, which is the shape
/// [`crate::archive_map`] exists to get off the emit path.
pub(crate) fn read_extent(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
self.absorber.read_extent(offset, len)
}
/// One mapping of the verbatim archive, to be taken **once per operation**
/// and sliced N times. See [`crate::archive_map::ArchiveMap::snapshot`].
pub(crate) fn archive_snapshot(&self) -> Result<Arc<crate::archive_map::Mapped>> {
self.absorber.map.snapshot()
}
/// **The one resolver**: an extent's bytes, borrowed from `snap` when it
/// covers them and `pread` into a fresh buffer when it does not.
///
/// `Cow::Borrowed` is the ordinary answer and the whole point — no syscall,
/// no allocation, no copy, just a bounds-checked slice of the page cache.
/// `Cow::Owned` happens for an extent past the mapped end, which means the
/// blob file grew after the snapshot was taken (a push landed mid-clone).
/// `Mapped::get` returning `None` is *"go and pread it"* and never *"there
/// are no bytes"*, so the two arms return the **same bytes** and differ only
/// in what they cost — which is the property
/// `an_extent_past_the_mapping_falls_back_and_returns_the_same_bytes`
/// asserts.
pub(crate) fn extent<'m>(
&self,
snap: &'m crate::archive_map::Mapped,
offset: u64,
len: u64,
) -> Result<std::borrow::Cow<'m, [u8]>> {
match snap.get(offset, len) {
Some(b) => Ok(std::borrow::Cow::Borrowed(b)),
None => Ok(std::borrow::Cow::Owned(self.read_extent(offset, len)?)),
}
}
/// **A header probe: AT MOST `len` bytes at `offset`, fewer at the archive's end.**
///
/// [`Self::extent`] is for an extent the index vouches for and refuses a
/// short read, which is right there: a short extent is a truncated archive.
/// A header probe is not such an extent. It asks for a fixed ceiling
/// ([`crate::serve`]'s `HEADER_PROBE`) to cover the longest header, and an
/// entry shorter than that at the END of the archive has fewer bytes after
/// it. That is not truncation and refusing it cost a live repository every
/// shallow clone (FORGE-SHALLOW, git.gunnar.rs 2026-10-28: edda's 41-byte
/// ref-delta, `short read at (16843881, 64)`). The parsers read a grammar,
/// not a fixed width, so they take what is there. The same clamp
/// [`ArchiveWindow`] gives [`crate::resolve::materialize_at`].
pub(crate) fn header_probe<'m>(
&self,
snap: &'m crate::archive_map::Mapped,
offset: u64,
len: u64,
) -> Result<std::borrow::Cow<'m, [u8]>> {
clamped_window(&self.absorber, snap, offset, len)
}
/// Record a durable pack as pending index work.
///
/// The **bit going clear**, and the ack path's last act before it returns.
pub(crate) fn queue(&self, pack_id: u64, extent: Extent) -> Result<()> {
self.absorber.queue(pack_id, extent)
}
fn reach_bitmaps(&self) -> Result<Arc<crate::reach::ReachTable>> {
self.absorber.reach_bitmaps()
}
/// [`Absorber::reach_bitmaps_with`], for the differential guard.
#[cfg(test)]
pub(crate) fn reach_bitmaps_with(
&self,
policy: crate::reach::ReachPolicy,
cache: bool,
) -> Result<Arc<crate::reach::ReachTable>> {
self.absorber.reach_bitmaps_with(policy, cache)
}
/// `(loaded, extended, unpersisted)` — how many folds read the persisted
/// reachability table, how many extended it for pushed commits, and how
/// many extensions have not been written back yet. The witnesses for the
/// persisted and incremental paths (LAW 2): the bytes a loaded, an extended
/// and a rebuilt table answer with are identical by construction.
pub fn reach_stats(&self) -> (u64, u64, u64) {
self.absorber.reach_stats()
}
pub(crate) fn refold(&self) -> Result<()> {
self.absorber.refold()
}
}
impl<S: ObjectIndex> Absorber<S> {
/// `shared` is the writer's descriptor on the same file
/// ([`crate::archive_write::ArchiveWrite::blobs_shared`]); `None` opens one,
/// which is what a caller with no writer does.
fn open(
blobs: &Path,
shared: Option<Arc<File>>,
hash: GitHashKind,
objects: ObjectReadStack<S>,
exploded: Arc<ExplodedCache>,
) -> Result<Self> {
let reader = match shared {
Some(f) => f,
None => Arc::new(
File::open(blobs)
.with_context(|| format!("opening {} for reads", blobs.display()))?,
),
};
let label = blobs
.parent()
.map(|p| p.display().to_string())
.unwrap_or_else(|| blobs.display().to_string());
let graph_cache = crate::arms::graph_cache_from_env()?;
if !graph_cache {
// A memo left by an earlier open with the cache on: this store keeps none.
let stale = blobs.with_file_name(GRAPH_CACHE_FILE);
match std::fs::remove_file(&stale) {
Ok(()) => {}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => tracing::warn!(
target: "znippy.git",
path = %stale.display(),
error = %e,
"the graph cache is off but a stale objects.graph could not be removed"
),
}
}
Ok(Self {
blobs: blobs.to_path_buf(),
reader,
map: crate::archive_map::ArchiveMap::new(blobs),
hash,
objects,
exploded,
derived: RwLock::new(Derived::default()),
graph_read: std::sync::atomic::AtomicU64::new(0),
graph_recomputed: std::sync::atomic::AtomicU64::new(0),
graph_cache: std::sync::atomic::AtomicBool::new(graph_cache),
reach_carried: std::sync::atomic::AtomicU64::new(0),
reach_rebuilt: std::sync::atomic::AtomicU64::new(0),
reach_loaded: std::sync::atomic::AtomicU64::new(0),
reach_extended: std::sync::atomic::AtomicU64::new(0),
reach_unpersisted: std::sync::atomic::AtomicU64::new(0),
ordinal_folded: std::sync::atomic::AtomicU64::new(0),
ordinal_tail: std::sync::atomic::AtomicU64::new(0),
packs: Mutex::new(PackState::default()),
gate: Mutex::new(()),
progress: crate::progress::IndexProgress::new(&label),
cancel: std::sync::atomic::AtomicBool::new(false),
label,
watchdog: RwLock::new(crate::memguard::global()),
scanned: Mutex::new(HashMap::new()),
scan_build: Mutex::new(()),
unindexable: Mutex::new(HashMap::new()),
})
}
/// **Tree payloads, on demand** — [`crate::reach::TreeSource`] over this
/// store.
///
/// One lookup per tree the walk actually visits, instead of a resident map
/// of every tree in the repository. Where the bytes come from is
/// [`Absorber::resolved`]'s business and nobody else's: the exploded cache
/// if the tree is in it, its own delta chain out of the mmapped pack if it
/// is not — and the read then writes it back, so a second walk finds it
/// warm.
///
/// It must not take the `derived` lock, and it does not: `resolved` reads
/// the exploded pools, the `objects` index and the pending packs, none of
/// which is `derived`. Both callers build one of these **inside** their
/// `derived.write()` section, and that is only sound because of this
/// sentence.
fn tree_source(&self) -> TreesFromStore<'_, S> {
TreesFromStore { abs: self, chains: None }
}
/// [`Self::tree_source`] with a [`crate::resolve::ChainCache`] of
/// [`REACH_BUILD_CHAIN_CACHE_BYTES`] behind it — for the one reader that
/// walks every tree: the full reachability build. Dropped with the build.
fn tree_source_for_build(&self, objects: usize) -> TreesFromStore<'_, S> {
// Sized to the repository: a small one must not pay the full memo
// (MEASURED: a 28 251-object repository peaked at 274 MB with the full
// 128 MiB memo against 121 MB without — the memo was most of it).
let cap = objects
.saturating_mul(REACH_BUILD_CHAIN_CACHE_PER_OBJECT)
.clamp(8 << 20, REACH_BUILD_CHAIN_CACHE_BYTES);
TreesFromStore {
abs: self,
chains: Some(Mutex::new(crate::resolve::ChainCache::new(cap))),
}
}
/// Packs whose objects are not in the index yet — §13.12's bit, counted.
fn unindexed_packs(&self) -> usize {
self.packs.lock().expect("pack state lock").unabsorbed
}
/// **Index everything durable, here, now — the ack path's half of §13.9.**
///
/// # The two hours, in one function
///
/// MEASURED by lane P2 on the oden rig 2026-08-30. `git.receive_pack.checking`
/// fired at 07:43:49; its very first `peel` of a pushed tip missed the
/// exploded table and missed the `objects` index (the worker was minutes
/// into a 13 948 244-object pack), and fell into [`Self::scan_pending`],
/// which ran **a second, read-only, throw-away resolve of the whole 8.12 GiB
/// pack on the connection thread**:
///
/// ```text
/// 09:20:21 index.scanned "a reader scanned a pending pack for itself:
/// 13948244 objects in 5792442 ms"
/// ```
///
/// 1h36m32s on the 16 GB box and 1h44m14s on the 8 GB one, for rows it then
/// threw away with the comment *"keeping them would be the index, and the
/// worker is building that"*. That sentence is the defect: two resolvers
/// over the same bytes, each holding its own O(pack) `PackWalk::entries` and
/// oid table, on four vCPUs — which is also **why the watchdog paused**. The
/// reader's second copy is what ate the `MemAvailable` the indexer tripped
/// over, and neither pass ever finished the check.
///
/// # Who may call this
///
/// **The ack path, and nothing else.** It takes [`Absorber::gate`], so it
/// waits for whoever is absorbing — and *"nobody waits on the indexer"* is a
/// law with its own red test
/// (`a_read_before_the_drain_is_served_from_the_pack_and_never_waits`): a
/// clone, an `ls-remote`, a `control status` must be answered while the
/// drain is parked, and they still are, through [`Self::scan_pending`].
///
/// The connectivity check is the one caller for which that is the wrong
/// trade. It has to prove that every object reachable from 2874 new tips is
/// present, which means reading them; an entry's oid exists only once its
/// delta chain has been applied; so the pack must be resolved before the
/// check can answer, and the only question is whether it is resolved **once
/// or twice**. This makes it once. `git receive-pack` makes the same choice
/// and calls it `index-pack`.
///
/// If the account worker already holds the gate, this blocks on it and then
/// finds `is_absorbed` set and returns: whoever got there first did the work.
///
/// Returns how many packs this call absorbed; `0` means it could not (it is
/// nested inside an absorb of the only pending pack, or there was nothing
/// to do).
fn index_pending_now(&self) -> Result<usize> {
if self.unindexed_packs() == 0 {
return Ok(0);
}
// **Nested**: this thread is already inside `absorb_gated` and holds the
// gate, so taking it again is a deadlock, not a wait. The other pending
// packs are still ours to absorb inline — that is lane U's case, a thin
// push whose base sits in a pack pushed a moment earlier — and the one
// we are in the middle of is not.
if let Some(current) = IN_ABSORB.with(|c| c.get()) {
let jobs: Vec<PendingPack> = {
let s = self.packs.lock().map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
return self.absorb_each(&jobs, Some(current));
}
let _gate = self.gate.lock().map_err(|_| anyhow!("absorb gate poisoned"))?;
let jobs: Vec<PendingPack> = {
let s = self.packs.lock().map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
// `absorb_gated` is a no-op for a pack somebody absorbed while we were
// queued on the gate, so the loser of the race pays nothing.
self.absorb_each(&jobs, None)
}
/// **Absorb the pending packs only if nobody else is absorbing.**
///
/// `Some(n)` = the gate was free and this caller did the work; `None` = the
/// worker (or another reader) holds the gate — possibly paused by the
/// memory watchdog for as long as the box is short — and this caller did
/// **not** wait for it. The owner's rule, 2026-08-30: *nobody waits on the
/// indexer.* A read that needs the index now answers from the pack for
/// itself ([`Self::scan_pending`]) and a read that needs the whole graph
/// says so by name rather than blocking behind a kernel pack.
fn try_absorb_pending(&self) -> Result<Option<usize>> {
if self.unindexed_packs() == 0 {
return Ok(Some(0));
}
let _gate = match self.gate.try_lock() {
Ok(g) => g,
Err(std::sync::TryLockError::WouldBlock) => return Ok(None),
Err(std::sync::TryLockError::Poisoned(_)) => bail!("absorb gate poisoned"),
};
let jobs: Vec<PendingPack> = {
let s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
self.absorb_each(&jobs, None).map(Some)
}
/// **A read served from the pack on the reader's own request, never by
/// waiting for the worker.**
///
/// Every oid in `wants` is looked up in the `objects` table first — the
/// rows the worker has already landed, batch by batch, are visible there
/// with no lock in common with it. The ones still missing, if any pack is
/// still un-indexed, are found by **resolving the pending packs read-only
/// on this thread**: the same resolver the worker runs
/// ([`crate::resolve::resolve_streaming`]), over the same mmapped bytes,
/// with a bounded base set, but with **no sink, no gate and no rows
/// written** — it produces rows for the caller and nothing else. The pass
/// stops the moment every wanted oid has been seen.
///
/// `capture` names one oid whose **content** the caller wants as well (a
/// [`GitStore::content`] read); it rides through a
/// [`crate::exploded::CaptureOne`] sink and comes back in the second slot.
///
/// # Cost, stated
///
/// An oid that is in a pending pack costs one read-only pass up to where it
/// sits; an oid that is **absent** costs a full pass over every pending
/// pack — for a kernel pack, minutes of one core. That is the price of
/// never answering "absent" for an object whose bytes are durable, paid by
/// the reader that asked and by nobody else. The rows it computes are not
/// kept: keeping them would be the index, and the worker is building that.
fn scan_pending(
&self,
wants: &[&[u8]],
capture: Option<&[u8]>,
) -> Result<(Vec<Option<crate::index_layout::IndexRow>>, Option<(GitObjectKind, Vec<u8>)>)> {
use crate::index_layout::IndexRow;
let mut out: Vec<Option<IndexRow>> = vec![None; wants.len()];
let mut missing: HashMap<&[u8], Vec<usize>> = HashMap::with_capacity(wants.len());
for (i, w) in wants.iter().enumerate() {
missing.entry(*w).or_default().push(i);
}
if missing.is_empty() && capture.is_none() {
return Ok((out, None));
}
// **No nested BUILD.** The resolver asks this absorber (as its
// `BaseSource`) for a thin pack's external base, and that base may
// itself be un-indexed — a second *scan build* started from inside the
// first, and a third from inside that, without end.
//
// The guard used to answer `None` outright when re-entered, and that
// was wider than the recursion it was written for: a base sitting in
// **another pack that is already scanned** is one binary search away,
// and refusing it made a thin push on top of a just-acked one fail
// `[remote rejected] … missing necessary objects` — lane U's rig
// finding, 2026-08-30. So a nested call now consults the scans that
// exist and builds none: a lookup terminates, a build is what does not.
thread_local! { static SCANNING: std::cell::Cell<bool> = const { std::cell::Cell::new(false) }; }
let nested = SCANNING.with(|s| s.get());
struct Unset(bool);
impl Drop for Unset {
fn drop(&mut self) {
if !self.0 {
SCANNING.with(|s| s.set(false));
}
}
}
SCANNING.with(|s| s.set(true));
let _unset = Unset(nested);
let pending: Vec<PendingPack> = {
let s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
if pending.is_empty() {
return Ok((out, None));
}
let snap = self.map.snapshot()?;
let mut content: Option<(GitObjectKind, Vec<u8>)> = None;
for job in &pending {
if missing.is_empty() && (capture.is_none() || content.is_some()) {
break;
}
let scanned = match self.cached_scan(job.offset) {
Some(s) => s,
// Inside another scan: no build, so the recursion the guard
// exists for cannot happen, and a pack nobody has scanned yet
// is simply not consulted for this base.
None if nested => continue,
None => self.scanned_pack(&snap, job)?,
};
missing.retain(|oid, slots| match scanned.row(oid) {
Some(r) => {
for &i in slots.iter() {
out[i] = Some(r.clone());
}
false
}
None => true,
});
if let Some(want) = capture
&& content.is_none()
&& let Some((entry, _)) = scanned.find(want)
{
let bytes: std::borrow::Cow<'_, [u8]> = match snap.get(job.offset, job.len) {
Some(b) => std::borrow::Cow::Borrowed(b),
None => std::borrow::Cow::Owned(self.read_extent(job.offset, job.len)?),
};
let sp = scanned.clone();
let offset_of = move |oid: &[u8]| sp.find(oid).map(|(e, _)| sp.walk.entries[e].offset);
content = Some(crate::resolve::materialize_entry(
&bytes,
&scanned.walk,
entry,
self.hash.oid_len(),
&offset_of,
self,
)?);
}
}
Ok((out, content))
}
/// The scan of one pending pack **if it is already built** — a lookup, and
/// nothing that could recurse. What a nested `scan_pending` is allowed.
fn cached_scan(&self, offset: u64) -> Option<Arc<ScannedPack>> {
self.scanned
.lock()
.unwrap_or_else(|e| e.into_inner())
.get(&offset)
.cloned()
}
/// The scan of one pending pack: cached, or built now by one read-only
/// pass (the same resolver, bounded bases, no sink, no rows written).
fn scanned_pack(&self, snap: &crate::archive_map::Mapped, job: &PendingPack) -> Result<Arc<ScannedPack>> {
if let Some(s) = self.scanned.lock().unwrap_or_else(|e| e.into_inner()).get(&job.offset) {
return Ok(s.clone());
}
let _one_at_a_time = self.scan_build.lock().unwrap_or_else(|e| e.into_inner());
if let Some(s) = self.scanned.lock().unwrap_or_else(|e| e.into_inner()).get(&job.offset) {
return Ok(s.clone());
}
let t0 = std::time::Instant::now();
let bytes: std::borrow::Cow<'_, [u8]> = match snap.get(job.offset, job.len) {
Some(b) => std::borrow::Cow::Borrowed(b),
None => std::borrow::Cow::Owned(self.read_extent(job.offset, job.len)?),
};
let walk = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
let mut rows: Vec<(Vec<u8>, u32, u64)> = Vec::with_capacity(walk.entries.len());
// A reader's own budget: a quarter of the worker's, no watchdog —
// the pause is the worker's and only the worker's.
let limits = crate::resolve::ResolveLimits {
base_budget: (crate::resolve::base_budget_from_env() / 4).max(1),
cancel: Some(&self.cancel),
label: &self.label,
..crate::resolve::ResolveLimits::none()
};
// Entry index = position in offset order; the rows come out in
// resolution order, so map offset → index by binary search.
let report = crate::resolve::resolve_streaming(
&bytes,
&walk,
self.hash,
job.offset,
self,
&crate::exploded::NoSink,
limits,
&mut |batch| {
for r in batch.drain(..) {
let pack_off = r.offset - job.offset;
let idx = walk
.entries
.binary_search_by_key(&pack_off, |e| e.offset)
.map_err(|_| anyhow!("resolved row at {pack_off} has no walk entry"))?;
rows.push((r.oid, idx as u32, r.uncompressed_size));
}
Ok(())
},
)
.context("resolving a pending pack read-only for lookups")?;
rows.sort_unstable_by(|a, b| a.0.cmp(&b.0));
let scanned = Arc::new(ScannedPack {
walk,
rows,
archive_offset: job.offset,
});
tracing::info!(
target: "gunnar::index",
event = "index.scanned",
repo = %self.label,
pack_id = job.pack_id,
objects = scanned.rows.len(),
reinflations = report.reinflations,
elapsed_ms = t0.elapsed().as_millis() as u64,
"a reader scanned a pending pack for itself: {} objects in {} ms",
scanned.rows.len(),
t0.elapsed().as_millis()
);
self.scanned
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(job.offset, scanned.clone());
Ok(scanned)
}
/// **Drop everything this store holds that can be rebuilt** — what the
/// watchdog's flush reaches: the readers' scanned packs (`scanned`, ~1.2
/// GB on linux, rebuilt by the next lookup's pass) and the exploded
/// table's oid tree and batch cache. The rows on disk are untouched.
///
/// **Returns the rows it actually dropped**, and that number is the point.
/// Lane P2's rig logged this event 190 and 627 times with
/// `scanned_rows_dropped=0` on every single one and kept waiting anyway,
/// because the caller could not tell a spill that freed a gigabyte from one
/// that freed nothing. Three zeros in a row now end the pause
/// ([`crate::memguard::PauseEnd::NothingToGive`]).
///
/// The exploded table's own caches are an mmap index and a small batch
/// cache; they are dropped here too and deliberately not counted, because
/// they are not where a kernel-sized pack's memory is.
fn spill(&self) -> u64 {
let dropped = {
let mut s = self.scanned.lock().unwrap_or_else(|e| e.into_inner());
let n: usize = s.values().map(|p| p.rows.len()).sum();
s.clear();
n
};
self.exploded.drop_caches();
tracing::info!(
target: "gunnar::index",
event = "index.spilled",
repo = %self.label,
scanned_rows_dropped = dropped,
"the store dropped its rebuildable caches for the memory watchdog"
);
dropped as u64
}
/// This store's watchdog (see the field).
fn watchdog(&self) -> Arc<crate::memguard::MemWatchdog> {
self.watchdog.read().unwrap_or_else(|e| e.into_inner()).clone()
}
/// One oid, never waiting: the index, then the pending packs.
fn lookup_never_wait(&self, oid: &[u8]) -> Result<Option<crate::index_layout::IndexRow>> {
if let Some(r) = self.objects.lookup(oid) {
return Ok(Some(r));
}
// **Check-then-act, and the answer is to look again, not to wait.**
// The index miss above and the bit read below are two reads with the
// drain running between them: the pack's rows can land AND its bit
// clear after the miss, and "no pending pack, so absent" was then a
// lie about a durable object — the one answer §13.12 forbids
// (`no_read_is_told_absent_while_the_drain_is_running` caught it,
// 1 absent answer in 1352, 2 of 10 full-suite runs under load,
// 2026-09-28). The bit clears only AFTER the rows go in, so once it
// reads clear a second index read is authoritative.
if self.unindexed_packs() == 0 {
return Ok(self.objects.lookup(oid));
}
let (mut rows, _) = self.scan_pending(&[oid], None)?;
if let Some(r) = rows.pop().flatten() {
return Ok(Some(r));
}
// The same window one step later: the pack was absorbed and left the
// pending list between the bit read and the scan.
Ok(self.objects.lookup(oid))
}
/// Many oids, never waiting: one index pass, then one read-only pass over
/// the pending packs for whatever the index did not have.
fn lookup_batch_never_wait(&self, oids: &[&[u8]]) -> Result<Vec<Option<crate::index_layout::IndexRow>>> {
let mut rows = self.objects.lookup_batch(oids);
if rows.iter().all(Option::is_some) {
return Ok(rows);
}
// See `lookup_never_wait`: a miss followed by a clear bit is re-read,
// because the drain may have landed the rows between the two.
if self.unindexed_packs() == 0 {
return Ok(self.objects.lookup_batch(oids));
}
let missing: Vec<&[u8]> = oids
.iter()
.zip(&rows)
.filter(|(_, r)| r.is_none())
.map(|(o, _)| *o)
.collect();
let (found, _) = self.scan_pending(&missing, None)?;
let mut by_oid: HashMap<&[u8], crate::index_layout::IndexRow> = HashMap::with_capacity(found.len());
for (o, r) in missing.iter().zip(found) {
if let Some(r) = r {
by_oid.insert(*o, r);
}
}
for (o, slot) in oids.iter().zip(rows.iter_mut()) {
if slot.is_none()
&& let Some(r) = by_oid.get(*o)
{
*slot = Some(r.clone());
}
}
// Absorbed between the bit read and the scan: whatever is still
// missing gets one more index read.
if rows.iter().any(Option::is_none) {
for (o, slot) in oids.iter().zip(rows.iter_mut()) {
if slot.is_none() {
*slot = self.objects.lookup(o);
}
}
}
Ok(rows)
}
/// The bit going clear: this pack is durable and its objects are not in.
///
/// Skips a pack the drain already absorbed — the ack path queues *after* it
/// has handed the extent to the channel, so the worker can legitimately be
/// finished before this runs, and clearing a bit for a pack that is already
/// indexed would leave it clear for ever.
fn queue(&self, pack_id: u64, extent: Extent) -> Result<()> {
let mut s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.note(pack_id, extent);
Ok(())
}
/// **The bit, derived from the two durable facts, on open.**
///
/// `journal` is every row this archive's journal holds, in append order.
/// [`acked_packs`](crate::archive_write::acked_packs) is what turns that into
/// the packs: its index *is* the pack ordinal — dense by construction, and
/// re-derived the same way by every later open. For each one, the question is
/// only *does this pack have any rows in the index*, which
/// [`ObjectReadStack::extents_with_rows`] answers in one early-terminating
/// scan rather than per pack.
///
/// Returns the packs whose bit came up **clear**: durable bytes, no rows. The
/// caller re-queues them on the account's channel, which is what makes a
/// pack that was mid-absorb when the process died get absorbed by the next
/// one instead of being durable-but-invisible for ever.
///
/// The verbatim extents are also handed to [`Derived::packs`] — *all* of
/// them, not only the unabsorbed ones. That is what lets [`BaseSource`]
/// re-derive a delta base's content from the truth after a restart; without
/// it a reopened store knows the rows of a pack it can no longer find the
/// bytes of.
///
/// # The one case where "no rows" does not mean "never absorbed" — and the
/// third durable fact that now tells them apart
///
/// [`GitOps::gc`] is the only thing that removes rows. A pack **all** of
/// whose objects it finds dead therefore ends with a journal extent and no
/// rows, which is byte for byte the state a pack that was in flight when the
/// machine died leaves behind. This diff cannot tell them apart, and it must
/// not guess: re-queueing the first resurrects every dead object, and *not*
/// re-queueing the second loses a pack whose bytes were acked.
///
/// So `gc` says which it is, durably, before it drops a row:
/// [`retire_packs`](crate::archive_write::retire_packs) appends a tombstone
/// naming the retired pack's offset, and a pack named by one is never
/// re-queued here. The bit is still derived from durable facts only — there
/// are now three of them (the journal's pack rows, the journal's tombstones,
/// the index's rows) and none of them is a cache of another.
///
/// A **partly** dead pack is not tombstoned and still has rows, so it takes
/// the `absorbed` branch exactly as it always did.
///
/// A tombstoned pack that *does* still have rows is a `gc` that was killed
/// between the tombstone and the drop. It is treated as absorbed, which is
/// what its rows say: nothing is lost, and the next `gc` finishes the job.
///
/// Its extent still goes to [`Derived::packs`] either way. The blob file is
/// append-only and `gc` truncates nothing, so a retired pack's bytes are
/// still exactly where the journal says they are — and a live object may
/// legitimately be a delta against a base that is itself unreachable, so
/// dropping the extent from that list would break the resolve rather than
/// tidy it.
///
/// # The second durable fact: §14's exploded table
///
/// A pack counts as absorbed only if the exploded table is **whole**, which
/// is `exploded.rows() >= objects.len()` — one row per object, because §14's
/// table is eager and skips nothing. If it is short, the table has been
/// dropped (or a process died between the two commits) and **no** pack counts
/// as absorbed: every one is re-queued and re-exploded from the verbatim
/// truth. That is the whole of "droppable": deleting the file is a supported
/// operation whose only cost is one re-ingest, and §13.12's rule holds
/// unchanged over it — absent means fall back, never means wrong.
///
/// `>=` and not `==` on purpose. A resolve that failed part way leaves rows
/// for the objects it got to, and a `gc` retires both tables together, so the
/// table can legitimately hold *more* than the index names; only being
/// **short** is evidence of a gap.
fn adopt_journal(&self, journal: &[Extent]) -> Result<Vec<PendingPack>> {
if journal.is_empty() {
return Ok(Vec::new());
}
// The packs, in ordinal order, and the offsets a `gc` retired. A
// tombstone is a row and not a pack, so it holds no ordinal.
let packs = crate::archive_write::acked_packs(journal);
let retired = crate::archive_write::retired_offsets(journal);
if packs.is_empty() {
return Ok(Vec::new());
}
// Under `Graph` the table is short **by design** — commits and trees
// only — so the row-count comparison would call every store dropped and
// re-queue every pack on every open. It is weakened to "not empty",
// which still catches the case this rule exists for (the file deleted),
// and is stated rather than hidden: making it exact needs a durable
// commit-and-tree count, which the index does not publish today.
// Only `Full` explodes one row per object, so only `Full` can compare
// counts. Under `Commits` (the shipped default) and `Graph` the eager
// pool is short **by design**, and the comparison would call every store
// dropped and re-queue every pack on every open — a Linux-sized re-index
// per restart. The weakened test still catches what the rule exists for
// (the file deleted) and is stated rather than hidden: making it exact
// needs a durable per-kind count, which the index does not publish.
//
// The LAZY pool is deliberately not consulted. Its size is a function of
// what has been READ, so it says nothing about whether a pack was
// absorbed, and an eviction must never look like a lost index.
let exploded_is_whole = match self.exploded.policy() {
crate::exploded_arrow::ExplodePolicy::Full => {
self.exploded.eager_rows()? >= self.objects.len() as u64
}
crate::exploded_arrow::ExplodePolicy::Off => true,
_ => self.objects.len() == 0 || self.exploded.eager_rows()? > 0,
};
let has_rows = if exploded_is_whole {
self.objects.extents_with_rows(&packs)?
} else {
vec![false; packs.len()]
};
{
let mut d = self
.derived
.write()
.map_err(|_| anyhow!("derived poisoned"))?;
d.packs = packs.clone();
}
let mut s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
let mut requeue = Vec::new();
for (i, (&extent, &absorbed)) in packs.iter().zip(&has_rows).enumerate() {
let pack_id = i as u64;
if absorbed {
s.mark_absorbed(pack_id, extent);
} else if retired.contains(&extent.0) {
// A `gc` dropped every row this pack had and said so durably
// *before* dropping them. There is no index work owed for it,
// ever: re-queueing it is precisely how its dead objects would
// come back. The bit goes **up** rather than being left clear,
// because a clear bit is what makes a read fall back to
// absorbing — which is the same resurrection by another door.
s.mark_absorbed(pack_id, extent);
} else {
s.note(pack_id, extent);
requeue.push(PendingPack {
pack_id,
offset: extent.0,
len: extent.1,
resumed: true,
});
}
}
Ok(requeue)
}
/// **The indexer's half of the split**, and not one of the twelve.
///
/// Reads each pending pack's verbatim bytes back out of the archive,
/// resolves them to oids ([`crate::resolve`]), appends the rows to the
/// `objects` table and re-folds the derived tables. This is what §13.9-12
/// puts behind the channel; it runs after the ack, never before one.
///
/// Two callers, one function: the account indexer's worker (through
/// [`ObjectAbsorb`]) and a read that arrived before that worker did. The
/// second is §13.12's fallback in its literal form — *slower, never wrong*.
fn absorb_pending(&self) -> Result<usize> {
// Nothing to do, and — this is the point — **no gate taken**: the common
// case is a store whose drain has kept up, and it must not queue behind
// one that is running.
if self.unindexed_packs() == 0 {
return Ok(0);
}
let _gate = self
.gate
.lock()
.map_err(|_| anyhow!("absorb gate poisoned"))?;
let jobs: Vec<PendingPack> = {
let s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.pending()
};
self.absorb_each(&jobs, None)
}
/// **Every pack it can, skipping the ones it has already given up on.**
///
/// The one loop all four absorb entry points run (LAW 5): they were four
/// copies of `for job in &jobs { self.absorb_gated(job)?; }`, and that `?`
/// is the bug this function exists to remove. Two rules, and both are
/// consequences of the same sentence — *work that can never succeed must
/// stop and say why*:
///
/// 1. **A failure does not abandon the packs behind it.** The loop runs to
/// the end and reports the FIRST failure afterwards, so a pack that
/// cannot be indexed no longer stops the push that arrived after it from
/// being indexed and acked.
/// 2. **A pack that has been given up on is not attempted again.** After
/// [`MAX_ABSORB_ATTEMPTS`] it is recorded in
/// [`Absorber::unindexable`] with the reason, logged once as
/// `index.unindexable`, and skipped from then on. Its objects are then
/// genuinely absent from the index, which every read path already knows
/// how to say — a push that needs one is refused by name at
/// `external_bases_exist`, and a clone that needs one fails the
/// connectivity check. What no longer happens is the store re-resolving
/// it on every request for the rest of the process's life.
///
/// `skip` is the pack this thread is standing inside an absorb of
/// ([`IN_ABSORB`]).
fn absorb_each(&self, jobs: &[PendingPack], skip: Option<u64>) -> Result<usize> {
let mut absorbed = 0usize;
let mut first: Option<anyhow::Error> = None;
for job in jobs {
if skip == Some(job.pack_id) || self.given_up_on(job.pack_id) {
continue;
}
match self.absorb_gated(job) {
Ok(()) => absorbed += 1,
Err(e) => {
// Cancellation is the repository being deleted: there is no
// queue left worth draining and nothing to record.
if self.cancel.load(std::sync::atomic::Ordering::Relaxed) {
return Err(e);
}
self.note_absorb_failure(job, &e);
if first.is_none() {
first = Some(e);
}
}
}
}
match first {
None => Ok(absorbed),
Some(e) => Err(e),
}
}
/// Every pack this store has given up on, ordinal order.
fn unindexable_packs(&self) -> Vec<Unindexable> {
let mut v: Vec<Unindexable> = self
.unindexable
.lock()
.unwrap_or_else(|e| e.into_inner())
.values()
.cloned()
.collect();
v.sort_by_key(|u| u.pack_id);
v
}
/// Has this store stopped trying to index that pack?
fn given_up_on(&self, pack_id: u64) -> bool {
self.unindexable
.lock()
.unwrap_or_else(|e| e.into_inner())
.contains_key(&pack_id)
}
/// Count one failed absorb, and stop at [`MAX_ABSORB_ATTEMPTS`].
///
/// The reason recorded is `root_cause`, not the outermost wrapper: that is
/// the sentence the backend actually has to offer, and lane Z's rule is that
/// the client gets it verbatim rather than a layer's paraphrase of it.
fn note_absorb_failure(&self, job: &PendingPack, err: &anyhow::Error) {
let giving_up;
{
let mut map = self.unindexable.lock().unwrap_or_else(|e| e.into_inner());
// A pack already given up on is not attempted, so an entry here is
// always a fresh count.
let e = map.entry(job.pack_id).or_insert_with(|| Unindexable {
pack_id: job.pack_id,
extent: (job.offset, job.len),
attempts: 0,
reason: String::new(),
});
e.attempts += 1;
e.reason = format!("{}", err.root_cause());
if e.attempts < MAX_ABSORB_ATTEMPTS {
let attempts = e.attempts;
map.remove(&job.pack_id);
tracing::warn!(
target: "znippy.git",
store = %self.label,
pack = job.pack_id,
attempts,
max = MAX_ABSORB_ATTEMPTS,
error = %err.root_cause(),
"a pack could not be indexed; it stays pending and will be tried again"
);
return;
}
giving_up = Some(e.clone());
}
if let Some(u) = giving_up {
tracing::error!(
target: "znippy.git",
store = %self.label,
event = "index.unindexable",
pack = u.pack_id,
offset = u.extent.0,
len = u.extent.1,
attempts = u.attempts,
error = %u.reason,
"this store has STOPPED trying to index this pack after {} attempts. Its bytes \
are durable and untouched, and every object it introduced is absent from the \
index — a push that deltas against one is refused by name and a clone that \
needs one fails its connectivity check, both immediately. Nothing will retry \
it: re-push the pack, or delete and re-create the repository.",
u.attempts
);
}
}
/// One pack, **with [`Absorber::gate`] already held**.
///
/// The order at the end is the correctness argument: the rows are in the
/// index *before* the bit is cleared. A read that finds the bit still clear
/// blocks on the gate and then finds the rows; a read that finds it set finds
/// the rows too. There is no interleaving in which a durable object answers
/// "absent" — which during negotiation would make a client withhold objects
/// and lose data.
fn absorb_gated(&self, job: &PendingPack) -> Result<()> {
if self.cancel.load(std::sync::atomic::Ordering::Relaxed) {
// The repository is being deleted: no pack still queued is worth a
// byte of work, and the drain must get through its queue fast so
// the store can be dropped and the directory removed.
return Err(crate::resolve::ResolveError::cancelled());
}
{
let s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
if s.is_absorbed(job.pack_id) {
// Already in, by whoever reached it first. Re-resolving is now
// *correct* — the exploded table is keyed by oid and the graph is
// folded from it rather than pushed at, so a second absorb of one
// pack changes nothing (that idempotence is what makes dropping
// the table and re-queueing every pack a safe operation). It is
// still a whole pack re-read, re-inflated and re-hashed for
// nothing, so it is skipped.
return Ok(());
}
}
// From here to the end of `absorb_one` this thread holds the gate, and
// the resolver is about to ask this same store for delta bases. See
// [`IN_ABSORB`]: the re-entry must not take the gate again and must not
// try to absorb the pack it is standing in.
let _in = InAbsorb::enter(job.pack_id);
self.absorb_one(job).map_err(|e| {
// The entry stays pending by construction — it was never removed —
// so the bytes are durable, the pack is still un-indexed, and reads
// keep falling back rather than answering absent.
e.context(format!(
"absorbing pack {} at ({}, {}) — its bytes are durable and it stays \
un-indexed; reads will keep falling back rather than answer absent",
job.pack_id, job.offset, job.len
))
})?;
let mut s = self
.packs
.lock()
.map_err(|_| anyhow!("pack state poisoned"))?;
s.mark_absorbed(job.pack_id, (job.offset, job.len));
drop(s);
// The index answers for it now; the reader's scan is dead weight.
self.scanned.lock().unwrap_or_else(|e| e.into_inner()).remove(&job.offset);
Ok(())
}
/// One pack: verbatim bytes in, `objects` rows **and** §14's exploded rows
/// out, from **one** walk and **one** resolve.
///
/// The exploded rows are not a second pass and not a second resolver (LAW 5):
/// [`crate::resolve::resolve_walked`] hands every object it produces to the
/// sink as it produces it, including the blob payloads it is about to drop,
/// so the side table falls out of the pass that was already being made.
///
/// Ordering is the correctness argument, and it is the same one the drain
/// makes one level up: **the exploded rows are committed before the fold that
/// reads them**, and the fold runs before this returns, so there is no
/// interleaving in which the graph is folded over a table that does not yet
/// hold this pack's commits.
fn absorb_one(&self, job: &PendingPack) -> Result<()> {
let t0 = std::time::Instant::now();
self.progress.begin(job.pack_id, job.len, job.resumed);
// **A live row every 10 s for as long as this pack takes**, carrying
// `peak_rss` and the re-inflation count. Those two rode only on
// `index.finished`, so lane P2's two rig boxes — which ran for 2h14m
// and 2h12m and never finished a pack — produced neither, in exactly
// the failure they were built for. The ticker is a thread and not a
// hook inside the resolve loop on purpose: it prints while the resolver
// is PAUSED too, which is when an operator most needs it.
let _tick = crate::progress::ProgressTicker::start(self.progress.clone());
let result = self.absorb_one_inner(job, t0);
self.progress.sample_memory();
self.progress.finish();
result
}
/// The body of [`absorb_one`](Self::absorb_one), so `finish` runs on every
/// exit and the progress row never sticks at 99 %.
fn absorb_one_inner(&self, job: &PendingPack, t0: std::time::Instant) -> Result<()> {
use crate::progress::Phase;
let progress = &self.progress;
// **The pack is read where it already is.** `read_extent` was a
// `pread` of the whole extent into a fresh `Vec` — 8.12 GiB of
// anonymous heap for torvalds/linux, before one byte was inflated, on a
// 16 GB box. The mapping costs nothing the kernel cannot reclaim
// ([`crate::archive_map`]), and `MemAvailable` — what the watchdog
// reads — knows that.
let snap = self.map.snapshot()?;
let bytes: std::borrow::Cow<'_, [u8]> = match snap.get(job.offset, job.len) {
Some(b) => std::borrow::Cow::Borrowed(b),
// Past the mapping: the file grew after the snapshot. Same bytes,
// one `pread` — the exception, not the shape.
None => std::borrow::Cow::Owned(self.read_extent(job.offset, job.len)?),
};
progress.set_phase(Phase::Walk);
let walked = crate::pack_walk::walk(&bytes, self.hash.oid_len())?;
let t_walk = t0.elapsed();
let dog = self.watchdog();
let spill = || self.spill();
let limits = crate::resolve::ResolveLimits::for_indexer(
&self.label,
&self.cancel,
progress,
dog,
&spill,
);
let mut batches = 0u64;
let mut appended = 0u64;
let mut entries: Vec<crate::index_layout::IndexEntry> = Vec::with_capacity(limits.row_batch);
let report = crate::resolve::resolve_streaming(
&bytes,
&walked,
self.hash,
job.offset,
self,
&*self.exploded,
limits,
&mut |rows| {
// **One batch of rows lands, then the next.** The exploded
// tier already flushed itself every 64 MiB; the index rows
// used to wait for the whole pack. Now the batch is the unit:
// rows go to the `objects` table as they are produced, and the
// pack's own "indexed" bit still goes up only after the LAST
// batch — `mark_absorbed` is the caller's, after this returns.
entries.clear();
entries.extend(rows.iter().map(|r| r.index_entry()));
rows.clear();
self.objects.append(&entries)?;
appended += entries.len() as u64;
batches += 1;
Ok(())
},
)?;
drop(entries);
let t_resolve = t0.elapsed() - t_walk;
progress.set_phase(Phase::Flush);
// Down before anything folds over it.
self.exploded.flush()?;
// A push is what fills the disk, so the push is what asks the cache to
// give space back (the disk line, `exploded_cache::DISK_LINE_PERCENT`).
// Until 2026-09-28 only a reader's write-back started the maintenance
// thread, so a store nobody read never looked at the disk. `kick`
// spawns and returns: the indexer does not wait for an unlink.
self.exploded.kick();
{
let mut d = self
.derived
.write()
.map_err(|_| anyhow!("derived poisoned"))?;
// A reopen already listed every acked extent (`adopt_journal`), so
// this is the first absorb of a pack pushed in *this* process — or a
// re-queued one that is already listed. `find` is what reads this
// list, so a duplicate would not be wrong, only unbounded.
if !d.packs.contains(&(job.offset, job.len)) {
d.packs.push((job.offset, job.len));
}
}
progress.set_phase(Phase::Fold);
let t_fold0 = std::time::Instant::now();
// The graph and the tree payloads are folded from the exploded table, not
// pushed at from here. That is what makes a re-absorb idempotent — which
// it has to be, because dropping the table re-queues every pack.
self.refold()?;
let _ = self.objects.maybe_rebuild()?;
let t_fold = t_fold0.elapsed();
// **The audit line**: every number is applied output of this absorb.
// `peak_rss` is the process's VmHWM off /proc/self/status — the high
// water mark of the whole server, which on the appliance is the box.
tracing::info!(
target: "gunnar::index",
event = "index.finished",
repo = %self.label,
pack_id = job.pack_id,
pack_bytes = job.len,
peak_rss = crate::memguard::peak_rss_bytes(),
mem_available = self.watchdog().available().unwrap_or(0),
objects = report.rows,
batches = batches,
rows_appended = appended,
reinflations = report.reinflations,
evictions = report.evictions,
watchdog_drops = report.watchdog_drops,
peak_base_bytes = report.peak_base_bytes,
pauses = report.pauses,
paused_ms = report.paused_ms,
walk_ms = t_walk.as_millis() as u64,
resolve_ms = t_resolve.as_millis() as u64,
fold_ms = t_fold.as_millis() as u64,
total_ms = t0.elapsed().as_millis() as u64,
resumed = job.resumed,
"indexed {} objects in {} batches: walk {} ms, resolve {} ms, fold {} ms; peak bases {}, {} evictions, {} re-inflations, {} pauses ({} ms); peak rss {}",
report.rows,
batches,
t_walk.as_millis(),
t_resolve.as_millis(),
t_fold.as_millis(),
crate::memguard::fmt_bytes(report.peak_base_bytes),
report.evictions,
report.reinflations,
report.pauses,
report.paused_ms,
crate::memguard::fmt_bytes(crate::memguard::peak_rss_bytes())
);
Ok(())
}
/// Recompute everything derived from the object set: the commit graph, the
/// tree payloads, the ordinal space, the generation numbers, and (lazily) the
/// bitmaps.
///
/// **The graph is folded from §14's exploded table, not accumulated.** That
/// is the whole fix for the clean-reopen hole: commit and tree payloads are
/// on disk, so this call produces the same graph after a restart as before
/// one, and it is called at the end of [`GitStore::open_with_arms`] for
/// exactly that reason. It also makes a re-absorb idempotent — the old
/// accumulate-into-a-`Vec` shape doubled every commit if a pack was absorbed
/// twice, which is why dropping the table and re-queueing everything was not
/// a safe operation before this.
///
/// **Generations are stored and must be recomputed in any fold** (§13,
/// decided). A generation number is `1 + max(parents)`, so it is invalidated
/// the moment history's shape changes — which is exactly what absorbing a
/// pack does. [`assign_generations`] is the one writer of that number and it
/// is called here, over the whole graph, rather than incrementally: an
/// incremental update would have to know which descendants moved, and being
/// wrong about that produces an ancestry cutoff that skips real ancestors.
///
/// # Cost
///
/// One range scan of the exploded table's commits and one of its trees, per
/// absorbed pack. It is the same order as the `tail_oids` scan below, which
/// this function already made on every absorb, and it reads **no blob
/// payload** — that is what the kind index in
/// [`crate::exploded`] is for.
///
/// ## And that order is the whole repository, on every push
///
/// Said plainly, because "the same order as the scan below" reads like a
/// reassurance and is not one: **nothing here is proportional to the push.**
/// Every call rebuilds every tree payload, every commit node, every
/// generation number, a `HashMap<String, u32>` over every oid in the
/// repository and a sorted `Vec<String>` of the same — for a push of one
/// commit exactly as for a push of ten thousand.
///
/// MEASURED on oden 2026-08-11, release, quiet box (`some avg10` 0.00). The
/// **same** 100-commit increment, pushed onto the same history seeded to
/// four different depths, server CPU for the push:
///
/// | objects already in the repository | CPU |
/// |---:|---:|
/// | 9 614 | 0.12 s |
/// | 35 350 | 0.23 s |
/// | 103 988 | 0.45 s |
/// | 206 925 | 0.77 s |
///
/// That is a straight line: **≈88 ms fixed, plus ≈3.3 µs per object already
/// stored**, and the fit predicts the four points to within 20 ms. It is
/// also why a push is slower than the ack path suggests — §13.9 puts the
/// index behind a channel, but `put_refs` checks every ref target with
/// `has`, `has` calls `lookup_one`, and `lookup_one` absorbs pending packs
/// first. The drain is off the *ack* and squarely on the *push*.
///
/// This is not fixed here and the numbers are recorded so the next attempt
/// starts from them rather than from a guess. The obvious lever — fold
/// incrementally — is the one the paragraph above rules out for
/// generations, so it needs its own argument and its own gate.
fn refold(&self) -> Result<()> {
// Everything derived is computed **before** the write lock, which is then
// held only for the swap: the fold is the long part and nothing that
// reads `derived` should queue behind it.
//
// Ordinals first — **read, not re-scanned, when the tail has not moved.**
// S-082's fix, mirroring the commit-graph memo one paragraph down:
// `objects.ordinal` beside the pack holds `oids_raw` (every oid in the
// store, in ordinal = sorted-raw order), and a cold open reads its couple
// of megabytes back sequentially in milliseconds instead of walking the
// whole redb B-tree — REFOLD measured that scan at ~65% (0.45 s) of a
// 136k-object reopen, and proved a flat buffer measured the same, so the
// cost is the SCAN, not the allocation. It is a memo, not a second truth:
// the redb tail stays authoritative and the file is discarded unless its
// (epoch, count) both match the live tail, so a push, a gc or a torn
// write all fall back to the scan and re-persist.
//
// The raw concatenation is BOTH the ordinal space and the validity token
// for the commit graph below, so the graph is read whenever the object
// set is unchanged; the hex form and the ordinal map fold out of the same
// bytes, so ordinal `o` names the same object in every representation by
// construction rather than by agreeing computations (LAW 5). The tail
// hands oids in oid order, which is byte order over fixed-width oids and
// therefore already the sorted concatenation the memo stores.
//
// **2026-09-04: the memo is gone, because the scan it spared is gone.**
// The read stack persists its projection now (`objects.index`), and the
// sorted oid list IS that index's oid column — one contiguous copy out
// of a mapped buffer, plus a merge of the few rows redb still holds
// since the last fold. No file of its own, no header, no epoch to
// validate: the projection is current by construction. What is kept is
// the provenance — how many oids came off the index and how many off
// the tail — because that is the only thing that can prove the space
// was not a scan (both paths yield identical bytes).
let space = self.objects.oids_raw()?;
self.ordinal_folded
.store(space.folded_rows, std::sync::atomic::Ordering::Relaxed);
self.ordinal_tail
.store(space.tail_rows, std::sync::atomic::Ordering::Relaxed);
let oids_raw = space.raw;
let token = graph_token(&oids_raw);
// The commit graph — **read, not recomputed, when the object set has not
// moved.** S-065's fix: `objects.graph` beside the pack holds the folded
// graph with generations, written by this same call at derive time and
// read here. It is not a second truth — it is a memoized projection
// validated on every read against the authoritative object set (the
// token is a blake3 over every oid in the store), and an absent file, a
// stale token or a decode failure all fall back to the fold and
// re-persist. Same shape as lane REFCAS persisting a tag's peeled target.
//
// On a fold, commit payloads are read out of §14's exploded table (no
// tree payload — that was lane Q's 28.24 GB and most of the 50.9 GiB RSS
// on `torvalds/linux`), parsed, and generation-numbered. That is the
// whole of the O(commits) open cost this persistence removes.
let oid_len = self.hash.oid_len();
let cached = if self.graph_cache.load(std::sync::atomic::Ordering::Relaxed) {
self.load_persisted_graph(&token)?
} else {
None
};
// The whole fold runs under the write lock. It is the short part now:
// after a push the graph is extended by the pushed commits, the ordinal
// space appends, and the reachability table gets entries for the new
// tips — all O(pushed), whatever the repository's size. The one long
// fold is the first one of a store with no memo, and nothing reads a
// store before its open returns.
let mut d = self
.derived
.write()
.map_err(|_| anyhow!("derived poisoned"))?;
// ── the ordinal space: follow the store, never renumber ───────────
//
// `appended` is the range of ordinals this fold added. `None` means the
// space was (re)built from the sorted list — the first fold, or an
// object removed under the table (gc, retain) — and every entry of the
// reachability table was dropped with it.
let mut space_loaded = false;
if d.ordinals.is_empty() && d.reach.is_empty() {
// A cold open: the persisted table, if it is still exact.
if let Some((ordinals, table)) = self.load_persisted_reach(oid_len)? {
d.ordinals = ordinals;
d.reach = Arc::new(table);
space_loaded = true;
}
}
let appended: Option<std::ops::Range<u32>> = if d.ordinals.is_empty() {
None
} else {
d.ordinals.follow(&oids_raw)
};
if appended.is_none() {
d.ordinals = crate::reach::OrdinalTable::from_sorted(oids_raw.clone(), oid_len);
d.reach = Arc::default();
space_loaded = false;
}
if space_loaded {
self.reach_loaded
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
// ── the commit graph ──────────────────────────────────────────────
let graph: Vec<CommitNode> = match cached {
Some(g) => {
self.graph_read.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
g
}
None => {
self.graph_recomputed
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let g = match &appended {
// **Extend, do not re-derive.** The previous fold's rows are
// exact (a commit is immutable); only the objects this fold
// appended can be new commits, and the exploded table says
// which of them are, without resolving a single delta.
Some(r) if !d.graph.is_empty() => {
let mut nodes = std::mem::take(&mut d.graph);
let space = d.ordinals.view();
for o in r.clone() {
let Some(oid) = space.oid(o) else { continue };
if let Some((GitObjectKind::Commit, payload)) = self.exploded.content(oid)? {
let h = crate::object::parse_commit(&payload);
nodes.push(CommitNode {
oid: hex::encode(oid),
parents: h.parents,
tree: h.tree,
committer_time: h.committer_time,
generation: 0,
});
}
}
assign_generations(nodes)
}
_ => {
let commits = self.exploded.of_kind(GitObjectKind::Commit)?;
let g: Vec<CommitNode> = commits
.iter()
.map(|(oid, payload)| {
let h = crate::object::parse_commit(payload);
CommitNode {
oid: hex::encode(oid),
parents: h.parents,
tree: h.tree,
committer_time: h.committer_time,
generation: 0,
}
})
.collect();
assign_generations(g)
}
};
// Persist for the next open. A write failure is not fatal to the
// fold — the graph is correct in memory and the next open simply
// recomputes — so it is logged, not propagated.
if !self.graph_cache.load(std::sync::atomic::Ordering::Relaxed) {
// Switched off: no memo on this disk.
} else if let Err(e) = self.persist_graph(&token, &g) {
tracing::warn!(store = %self.label, "could not persist objects.graph: {e:#}");
}
g
}
};
let mut ord_graph = crate::reach::OrdGraph::from_nodes(&graph, &d.ordinals.view());
// ── the reachability table: extend for the pushed commits ─────────
//
// Entries describe immutable commits over a space that just appended,
// so every one of them stands. What is new is the pushed commits; the
// new tips (and a sample of a long pushed branch) get entries by the
// bounded walk, inside this fold, so the first fetch after the push
// pays nothing. See `crate::reach::extend_reach_table`.
let mut reach_changed = false;
if !d.reach.is_empty() {
if let Some(r) = &appended {
let new_rows: Vec<u32> = (0..ord_graph.len() as u32)
.filter(|&row| {
let o = ord_graph.ords[row as usize];
o != crate::reach::NONE && o >= r.start
})
.collect();
if !new_rows.is_empty() {
let trees = self.tree_source();
let facts = crate::reach::ObjectFacts {
ordinal: d.ordinals.view(),
trees: &trees,
oid_len,
};
let started = std::time::Instant::now();
let extended = crate::reach::extend_reach_table(
&d.reach,
&ord_graph,
&facts,
&new_rows,
live_reach_policy(),
);
tracing::info!(
store = %self.label,
new_commits = new_rows.len(),
entries = extended.len(),
elapsed_ms = started.elapsed().as_millis() as u64,
"fold: extended the reachability table for the pushed commits"
);
d.reach = Arc::new(extended);
self.reach_extended
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
reach_changed = true;
}
}
self.reach_carried
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
// ── compaction: re-sort the space once it has grown enough ────────
let mut compacted = false;
if d.ordinals.extra_len() > d.ordinals.base_len() / ORDINAL_COMPACT_FRACTION
&& d.ordinals.extra_len() > ORDINAL_COMPACT_MIN
{
let fresh = crate::reach::OrdinalTable::from_sorted(oids_raw, oid_len);
let remap = d.ordinals.remap_to(&fresh.view());
let kept = d.reach.remapped(&remap).unwrap_or_default();
tracing::info!(
store = %self.label,
appended = d.ordinals.extra_len(),
base = d.ordinals.base_len(),
entries = kept.len(),
"fold: compacted the ordinal space (re-sorted, table renumbered once)"
);
d.ordinals = fresh;
d.reach = Arc::new(kept);
ord_graph = crate::reach::OrdGraph::from_nodes(&graph, &d.ordinals.view());
compacted = true;
reach_changed = true;
}
d.graph = graph;
d.reach_by_row = Arc::new(d.reach.by_row(&ord_graph));
d.ord_graph = Arc::new(ord_graph);
d.token = token;
tracing::info!(
store = %self.label,
graph_len = d.graph.len(),
space_len = d.ordinals.len(),
appended = appended.as_ref().map(|r| r.len()).unwrap_or(0),
space_rebuilt = appended.is_none(),
reach_entries = d.reach.len(),
reach_loaded = space_loaded,
The file goes on; the console shows all of it.