Skip to main content

znippy_plugin_git/
exploded_arrow.rs

1//! **The exploded objects table: ONE Arrow IPC file, one row per object.**
2//!
3//! # What this replaces, and what it measured
4//!
5//! [`crate::exploded`] holds `oid → [kind][payload…]` in **redb**. It works, and
6//! it is the wrong medium — MEASURED on `linux.git`, oden 2026-08-13,
7//! 11 697 976 objects:
8//!
9//! | | |
10//! |---|---:|
11//! | pack as pushed (verbatim, deflated + delta-encoded) | 6.4 GB |
12//! | its objects fully resolved (`git verify-pack -v`, summed) | **16.8 GB** |
13//! | `objects.exploded` on disk | **204 GB** |
14//!
15//! So the 2026-08-08 eager decision was RIGHT — resolved content is **2.6×** the
16//! pack, comfortably inside the 3× it budgeted — and the medium was wrong by
17//! **12×**. redb is a copy-on-write B-tree: 11.7 million small keys committed in
18//! [`FLUSH_BYTES`] batches rewrite interior pages over and over, and 187 of
19//! those 204 GB are page churn rather than anybody's data.
20//!
21//! It cost throughput for the same reason. [`crate::indexer`] fans a drained
22//! batch across N workers with `gatling_for_each` (LAW 3, never rayon) and every
23//! one of them then queued behind redb's single write transaction — MEASURED:
24//! **96.8% of ONE core** with 31 idle while the exploder ran.
25//!
26//! # The shape
27//!
28//! One table, one file, the payload IN the row — which is what this engine
29//! already does everywhere else ([`crate::reach`] stores a bitmap as
30//! `DataType::Binary`, [`crate::secrets`] a ciphertext):
31//!
32//! ```text
33//!   oid | object_type | mode | path | payload
34//! ```
35//!
36//! An Arrow IPC stream is `[schema message][batch message]…` — append-only, no
37//! pages to rewrite, so the file is the payload total plus framing. Each flush
38//! encodes one batch with [`IpcDataGenerator`] and appends it with
39//! [`write_message`]; nothing earlier in the file is ever touched again.
40//!
41//! `payload` is `LargeBinary` and not `Binary` deliberately: `Binary` offsets are
42//! `i32`, so one batch would cap at 2 GiB of payload and overflow *silently* on a
43//! corpus nobody tested. The wider offset costs four bytes a row.
44//!
45//! ## `mode` and `path` — the row IS a file
46//!
47//! A git object on its own is content without a name. The two nullable columns
48//! carry the name when the caller knows it (a tree walk does; a bare pack
49//! resolve does not), which is what lets this table be read as a filesystem
50//! rather than as a content-addressed bag. They are **nullable and best-effort**
51//! for a reason that is inherent and not a shortcut: one blob is reachable at
52//! many paths in many commits, so a path column can only ever record *a* path —
53//! the one it was first exploded at. Null means "not known here", never "at the
54//! root".
55//!
56//! # Reopening is cheap, and the index is lazy
57//!
58//! A batch's *metadata* carries its row count and its body length, so
59//! [`ExplodedArchive::open`] walks the message framing — seeking over the bodies,
60//! reading kilobytes — and knows how many rows the table holds and where every
61//! batch begins without touching a gigabyte. That is what `adopt_journal` needs
62//! on startup.
63//!
64//! The `oid → row` index costs one sequential pass and is therefore built on the
65//! **first lookup**, not on open. A push-heavy process never pays for it; a
66//! process that reads pays once. After that a point read is two `pread`s — the
67//! schema message and the one batch — decoded through [`StreamDecoder`], so it
68//! never scans the batches in front of the one it wants.
69//!
70//! # The lookup is ragnar's static tree — see [`OidTree`]
71//!
72//! That index used to be a `Vec<(Vec<u8>, Located)>` walked with
73//! `binary_search_by`: one heap allocation per oid — **11.7 million** of them on
74//! `linux.git` — and a `log₂ n` chain of dependent loads over scattered `Vec`
75//! headers. It is now the same `stree` (`znippy-zoomies`) this crate already puts
76//! over oids in [`crate::oid_index`], over flat parallel arrays.
77//! `nornir-workspace.toml`'s performance law names that structure by name; this
78//! is the second place in this crate that obeys it, and it obeys it by reusing
79//! [`crate::oid_index::key_for_oid`] rather than deriving a second key.
80//!
81//! # Droppable, and it is one `rm`
82//!
83//! Every row is re-derivable from the verbatim pack bytes, so the file can be
84//! deleted without consulting a client:
85//! [`crate::git_ops::Absorber::adopt_journal`] compares this table's row count
86//! against the `objects` table's and re-queues every pack if it is short. Absent
87//! means fall back and rebuild; it never means wrong. One file, one delete —
88//! which is the property the engine's own tests assert.
89
90use std::fs::{File, OpenOptions};
91use std::io::{Read, Seek, SeekFrom, Write};
92use std::os::unix::fs::FileExt;
93use std::path::{Path, PathBuf};
94use std::sync::Arc;
95use std::sync::Mutex;
96use std::sync::atomic::{AtomicU64, Ordering};
97
98use anyhow::{Context, Result};
99use znippy_common::arrow::array::{
100    Array, FixedSizeBinaryArray, FixedSizeBinaryBuilder, LargeBinaryArray, LargeBinaryBuilder,
101    StringArray, StringBuilder, UInt8Array, UInt8Builder, UInt32Array, UInt32Builder,
102};
103use znippy_common::arrow::buffer::Buffer;
104use znippy_common::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
105use znippy_common::arrow::ipc::reader::StreamDecoder;
106use znippy_common::arrow::ipc::writer::{
107    DictionaryTracker, IpcDataGenerator, IpcWriteOptions, write_message,
108};
109use znippy_common::arrow::ipc::root_as_message;
110use znippy_common::arrow::record_batch::RecordBatch;
111use znippy_zoomies::stree::STree64Mmap;
112
113use crate::object::GitObjectKind;
114use crate::oid_index::key_for_oid;
115
116pub const COL_OID: &str = "oid";
117pub const COL_TYPE: &str = "object_type";
118pub const COL_MODE: &str = "mode";
119pub const COL_PATH: &str = "path";
120pub const COL_PAYLOAD: &str = "payload";
121
122/// Buffered payload bytes that force a flush.
123///
124/// Not a memory *limit* — one object can exceed it on its own and is written
125/// anyway. It is what keeps a 2 GiB push from holding its whole resolved content
126/// in RAM before the first batch reaches the file.
127pub const FLUSH_BYTES: usize = 64 << 20;
128
129/// `oid | object_type | mode | path | payload`.
130///
131/// `oid_len` is a parameter because this engine serves both sha1 (20 bytes) and
132/// sha256 (32); a hardcoded 20 would silently truncate every sha256 key.
133pub fn exploded_schema(oid_len: usize) -> SchemaRef {
134    Arc::new(Schema::new(vec![
135        Field::new(COL_OID, DataType::FixedSizeBinary(oid_len as i32), false),
136        Field::new(COL_TYPE, DataType::UInt8, false),
137        Field::new(COL_MODE, DataType::UInt32, true),
138        Field::new(COL_PATH, DataType::Utf8, true),
139        Field::new(COL_PAYLOAD, DataType::LargeBinary, false),
140    ]))
141}
142
143/// The pack type codes, the same mapping [`crate::exploded`] uses, so the two
144/// media cannot disagree about what a `2` means.
145pub fn kind_code(k: GitObjectKind) -> u8 {
146    match k {
147        GitObjectKind::Commit => 1,
148        GitObjectKind::Tree => 2,
149        GitObjectKind::Blob => 3,
150        GitObjectKind::Tag => 4,
151    }
152}
153
154pub fn kind_of(code: u8) -> Option<GitObjectKind> {
155    match code {
156        1 => Some(GitObjectKind::Commit),
157        2 => Some(GitObjectKind::Tree),
158        3 => Some(GitObjectKind::Blob),
159        4 => Some(GitObjectKind::Tag),
160        _ => None,
161    }
162}
163
164/// **A row that retires an oid.**
165///
166/// The file is append-only, so `retain` cannot reach back and delete: it appends
167/// rows that say *this oid is gone*, and the index fold applies them in file
168/// order, so the last word about an oid wins. Dropping only from the in-memory
169/// index is what a first cut did, and it was WRONG in the way that matters —
170/// `a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened` came back
171/// with **551 commits instead of 1**, because a reopen rebuilds the index from
172/// the file and resurrected every row `gc` had retired.
173///
174/// `0` and not a sixth column: git's pack type codes are 1–7 and 0 is not one of
175/// them, so [`kind_of`] already answers `None` for it. A tombstone costs an oid
176/// and an empty payload, not a byte on every live row.
177pub const TOMBSTONE: u8 = 0;
178
179/// One buffered row, waiting for the next batch. `code` rather than a
180/// [`GitObjectKind`] because a tombstone is a row with no kind.
181#[derive(Debug, Clone)]
182struct Pending {
183    oid: Vec<u8>,
184    code: u8,
185    mode: Option<u32>,
186    path: Option<String>,
187    payload: Vec<u8>,
188}
189
190/// Where one batch message sits in the file. Recovered from the framing walk,
191/// which reads metadata and seeks over bodies.
192#[derive(Debug, Clone, Copy, PartialEq, Eq)]
193struct Batch {
194    /// Byte offset of the message's continuation marker.
195    at: u64,
196    /// Bytes the whole message occupies: `8 + metadata + body`.
197    len: u64,
198    /// Rows in it — read from the metadata, so counting them costs no body read.
199    rows: u32,
200    /// Absolute byte offset of the message BODY — where the buffer offsets the
201    /// metadata declares are measured from.
202    body: u64,
203    /// The payload DATA buffer's offset within the body, from the metadata's
204    /// own buffer table. [`EXTENT_UNAVAILABLE`] when the body is compressed or
205    /// the buffer layout is not the one [`exploded_schema`] writes — a point
206    /// read then falls back to decoding the batch, which is slow and correct.
207    pay_data: u64,
208}
209
210/// Sentinel for "no byte extent could be computed for this row/batch" — the
211/// fall-back-and-decode marker, never an offset a file could really have.
212const EXTENT_UNAVAILABLE: u64 = u64::MAX;
213
214/// The payload DATA buffer's offset within the message body, read from the
215/// batch metadata alone.
216///
217/// The payload is [`exploded_schema`]'s LAST field and a `LargeBinary`'s data
218/// buffer is its last buffer, so the table's final buffer entry IS the payload
219/// bytes — asserted by also counting the buffers (12: validity+data for oid and
220/// object_type, validity+data for mode, validity+offsets+data for path and for
221/// payload). `None` when the count disagrees or the body is compressed;
222/// [`ExplodedArchive::content`] then decodes the batch instead of guessing.
223fn payload_data_offset(meta: &[u8]) -> Option<u64> {
224    let msg = root_as_message(meta).ok()?;
225    let rb = msg.header_as_record_batch()?;
226    if rb.compression().is_some() {
227        return None;
228    }
229    let bufs = rb.buffers()?;
230    if bufs.len() != 12 {
231        return None;
232    }
233    let last = bufs.get(bufs.len() - 1);
234    (last.offset() >= 0).then(|| last.offset() as u64)
235}
236
237/// Where one object's row is — and where its payload BYTES are.
238///
239/// The extent is what makes a point read one small `pread` instead of a whole
240/// 64 MiB batch decode. MEASURED without it, rust-lang/rust push, 2026-08-19:
241/// the pre-ack connectivity walk visited commits and trees in graph order —
242/// effectively random across ~574 batches against a 4-slot cache — and decoded
243/// 6.7 GB/s of page cache to serve ~98 objects/s, one core pinned for hours, a
244/// ~200 000× read amplification. The extent costs 16 bytes per live oid.
245#[derive(Debug, Clone, Copy, PartialEq, Eq)]
246struct Located {
247    batch: u32,
248    row: u32,
249    kind: GitObjectKind,
250    /// Absolute file offset of this row's payload bytes, or
251    /// [`EXTENT_UNAVAILABLE`] — then the read falls back to the batch decode.
252    pay_at: u64,
253    /// Payload length in bytes. Meaningless when `pay_at` is the sentinel.
254    pay_len: u64,
255}
256
257/// **`oid → Located`, as ragnar's static tree over flat arrays.**
258///
259/// Three parallel arrays in one oid-ascending order, plus an [`STree64Mmap`]
260/// built over the first of them:
261///
262/// ```text
263///   keys  count * 8         i64 little-endian, ascending — the stree keyspace
264///   oids  count * oid_len   the FULL oid, for the verify step
265///   locs  count             batch, row, kind
266/// ```
267///
268/// ## Why this and not the `Vec<(Vec<u8>, Located)>` it replaces
269///
270/// That vector cost one heap allocation *per oid* — 11.7 M of them for
271/// `linux.git` — and its `binary_search_by` chased `log₂ n` dependent loads over
272/// `Vec` headers scattered across the heap before it could compare a byte. Here
273/// the oid bytes are contiguous, and `stree` routes through 64-byte nodes with a
274/// branchless AVX2 compare. `nornir-workspace.toml` names this structure as the
275/// one to use, and [`crate::oid_index`] already uses it over exactly these keys —
276/// so the key function is **imported from there**, not written again.
277///
278/// ## The 8-byte key is a FILTER, not a key — and this is where a wrong answer
279/// would come from
280///
281/// The key is [`key_for_oid`]: the first eight bytes of the oid, big-endian, top
282/// bit flipped so unsigned byte order becomes signed `i64` order (without the
283/// flip every oid starting `0x80..0xff` — half of them — sorts below every oid
284/// starting `0x00..0x7f`, and the keyspace `stree` requires to be ascending is
285/// not). **Twelve** of a sha1 oid's twenty bytes are not in the key at all (24 of
286/// 32 for sha256), so two distinct objects **can** share one key.
287///
288/// How likely, honestly: for `linux.git`'s 11.7 M objects, birthday arithmetic
289/// gives n²/2 · 2⁻⁶⁴ ≈ **3.7 × 10⁻⁶** — one table in ~270 000. So it will not
290/// happen by accident on one repository, and that is exactly why a verify-less
291/// probe would ship and pass every test anyone bothered to write. It does not
292/// stay accidental: finding *some* pair of contents whose oids share eight bytes
293/// is a 2³² birthday search — hours of commodity hashing, not a research result —
294/// and both objects are then perfectly ordinary blobs a client may push. (Hitting
295/// a *specific* stored oid's prefix is 2⁶⁴ and is not a threat.) A hosted forge
296/// serves oids that other people chose, so this is a property to hold by
297/// construction rather than a probability to accept.
298///
299/// So `stree` narrows and never decides. It routes to *a* member of the run of
300/// equal keys — which member is its own business, because its leaf scan starts at
301/// a block boundary that can fall in the middle of the run — so
302/// [`expand_run`](Self::expand_run) walks **both** directions to the run's ends,
303/// and [`find`](Self::find) then compares the **whole oid** against every
304/// candidate. Skipping that comparison would not merely be sloppy: a `have`
305/// negotiation for an oid this table has never seen, whose first eight bytes
306/// happen to match a stored one, would be answered with **the other object's
307/// bytes** — a silently wrong object, which is far worse than a slow one.
308struct OidTree {
309    /// Width of one oid. `0` only when the tree is empty.
310    oid_len: usize,
311    count: usize,
312    keys: Vec<u8>,
313    oids: Vec<u8>,
314    locs: Vec<Located>,
315    /// `None` for an empty table — `STree64Mmap::new_with_stride` asserts
316    /// `count > 0`.
317    tree: Option<STree64Mmap>,
318}
319
320/// Which member of a run of equal oids survives the fold — the two callers
321/// genuinely differ, and the difference is not cosmetic.
322#[derive(Debug, Clone, Copy, PartialEq, Eq)]
323enum Keep {
324    /// [`ExplodedArchive::flush`]'s incremental extension: an oid already in the
325    /// index keeps the [`Located`] it already had. That is what
326    /// [`ExplodedArchive::explode_at`] documents — a blob is reachable at many
327    /// paths and the table records the **first** name it was seen at.
328    First,
329    /// [`ExplodedArchive::index_now`]'s rebuild from the file: batches are walked
330    /// in file order and rows in row order, so the **last** word about an oid
331    /// wins. That is what makes a [`TOMBSTONE`] durable rather than advisory.
332    Last,
333}
334
335/// Accumulates sightings in the order they happen, then folds them.
336///
337/// Sightings, not entries: the same oid may arrive many times, and a `None`
338/// [`Located`] is a [`TOMBSTONE`] row saying the oid is gone. Nothing is
339/// deduplicated until [`finish`](Self::finish), because which sighting wins
340/// depends on their order and on [`Keep`].
341struct OidTreeBuilder {
342    /// Taken from the first oid pushed; every later one must match it.
343    oid_len: usize,
344    oids: Vec<u8>,
345    locs: Vec<Option<Located>>,
346}
347
348impl OidTreeBuilder {
349    fn with_capacity(rows: usize) -> Self {
350        Self { oid_len: 0, oids: Vec::new(), locs: Vec::with_capacity(rows) }
351    }
352
353    /// Record one sighting. `loc` is `None` for a tombstone row.
354    ///
355    /// A width that disagrees with the rows already pushed is refused rather
356    /// than padded: a truncated or extended oid would sort and compare as a
357    /// different object.
358    fn push(&mut self, oid: &[u8], loc: Option<Located>) -> Result<()> {
359        if self.locs.is_empty() && self.oids.is_empty() {
360            self.oid_len = oid.len();
361            self.oids.reserve(self.locs.capacity() * self.oid_len);
362        }
363        anyhow::ensure!(
364            oid.len() == self.oid_len,
365            "the exploded index holds {}-byte oids and a {}-byte one arrived",
366            self.oid_len,
367            oid.len()
368        );
369        self.oids.extend_from_slice(oid);
370        self.locs.push(loc);
371        Ok(())
372    }
373
374    fn oid_at(&self, i: usize) -> &[u8] {
375        &self.oids[i * self.oid_len..(i + 1) * self.oid_len]
376    }
377
378    /// Sort, fold each run of equal oids down to its winner, drop the ones whose
379    /// winner is a tombstone, and build the tree.
380    fn finish(self, keep: Keep) -> OidTree {
381        let n = self.locs.len();
382        let mut order: Vec<u32> = (0..n as u32).collect();
383        // STABLE, and load-bearing: `Keep` is about *insertion* order within a
384        // run of equal oids, so an unstable sort would make "first sighting"
385        // and "last word wins" both mean "whichever one the sort happened to
386        // leave there".
387        order.sort_by(|&a, &b| self.oid_at(a as usize).cmp(self.oid_at(b as usize)));
388
389        let mut keys: Vec<u8> = Vec::with_capacity(n * 8);
390        let mut oids: Vec<u8> = Vec::with_capacity(n * self.oid_len);
391        let mut locs: Vec<Located> = Vec::with_capacity(n);
392        let mut i = 0usize;
393        while i < n {
394            let oid = self.oid_at(order[i] as usize);
395            let mut j = i + 1;
396            while j < n && self.oid_at(order[j] as usize) == oid {
397                j += 1;
398            }
399            let pick = match keep {
400                Keep::First => order[i],
401                Keep::Last => order[j - 1],
402            } as usize;
403            if let Some(loc) = self.locs[pick] {
404                // Ascending oid order gives ascending keys, because
405                // `key_for_oid` is order-preserving over the oid's leading
406                // bytes — the property `stree` needs and the reason for the
407                // sign-bit flip.
408                keys.extend_from_slice(&key_for_oid(oid).to_le_bytes());
409                oids.extend_from_slice(oid);
410                locs.push(loc);
411            }
412            i = j;
413        }
414
415        let count = locs.len();
416        let tree = (count > 0).then(|| STree64Mmap::new_with_stride(&keys, count, 8));
417        OidTree { oid_len: self.oid_len, count, keys, oids, locs, tree }
418    }
419}
420
421impl OidTree {
422    fn len(&self) -> usize {
423        self.count
424    }
425
426    fn oid_at(&self, i: usize) -> &[u8] {
427        &self.oids[i * self.oid_len..(i + 1) * self.oid_len]
428    }
429
430    fn key_at(&self, i: usize) -> i64 {
431        i64::from_le_bytes(self.keys[i * 8..i * 8 + 8].try_into().unwrap())
432    }
433
434    /// Widen a hit to the whole run of equal keys. `stree` routes to *a* member
435    /// of the run — its leaf scan starts at a block boundary, which can fall
436    /// anywhere inside one — so both directions are walked rather than assumed.
437    fn expand_run(&self, pos: usize, key: i64) -> std::ops::Range<usize> {
438        let mut lo = pos;
439        while lo > 0 && self.key_at(lo - 1) == key {
440            lo -= 1;
441        }
442        let mut hi = pos + 1;
443        while hi < self.count && self.key_at(hi) == key {
444            hi += 1;
445        }
446        lo..hi
447    }
448
449    /// Resolve an oid. `None` when absent — never another object's row.
450    fn find(&self, oid: &[u8]) -> Option<Located> {
451        if oid.len() != self.oid_len {
452            return None;
453        }
454        let tree = self.tree.as_ref()?;
455        let key = key_for_oid(oid);
456        let pos = tree.find_exact(key, &self.keys)?;
457        self.verify(pos, key, oid)
458    }
459
460    /// The full-oid comparison the whole structure's correctness rests on.
461    fn verify(&self, pos: usize, key: i64, oid: &[u8]) -> Option<Located> {
462        self.expand_run(pos, key)
463            .find(|&i| self.oid_at(i) == oid)
464            .map(|i| self.locs[i])
465    }
466
467    /// Every live entry, in ascending oid order — what `of_kind` and `retain`
468    /// fold over.
469    fn iter(&self) -> impl Iterator<Item = (&[u8], Located)> + '_ {
470        (0..self.count).map(move |i| (self.oid_at(i), self.locs[i]))
471    }
472
473    /// The **baseline** the tree has to agree with: `binary_search` over the very
474    /// same oid array, followed by the very same run-and-verify. It is the
475    /// finder this file shipped with, and it exists so "the stree returns the
476    /// same row" is a measurement rather than an assertion.
477    #[cfg(test)]
478    fn find_by_binary_search(&self, oid: &[u8]) -> Option<Located> {
479        if oid.len() != self.oid_len || self.count == 0 {
480            return None;
481        }
482        let mut lo = 0usize;
483        let mut hi = self.count;
484        while lo < hi {
485            let mid = lo + (hi - lo) / 2;
486            if self.oid_at(mid) < oid { lo = mid + 1 } else { hi = mid }
487        }
488        (lo < self.count && self.oid_at(lo) == oid).then(|| self.locs[lo])
489    }
490
491    /// The **unverified** candidate run for a key: every entry sharing that
492    /// 8-byte prefix. Normally length 1; length > 1 is a real prefix collision.
493    /// Exposed so a test can prove the collision it constructed is real and that
494    /// the verify step is what tells the candidates apart.
495    #[cfg(test)]
496    fn candidate_run(&self, key: i64) -> std::ops::Range<usize> {
497        let Some(tree) = self.tree.as_ref() else { return 0..0 };
498        let Some(pos) = tree.find_exact(key, &self.keys) else { return 0..0 };
499        self.expand_run(pos, key)
500    }
501}
502
503/// Counters, all of them applied output.
504#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
505pub struct Stats {
506    pub rows: u64,
507    pub written: u64,
508    pub served: u64,
509    pub rederived: u64,
510    pub absent: u64,
511}
512
513/// §14's derived table for one repository.
514pub struct ExplodedArchive {
515    path: PathBuf,
516    /// The schema message: the file's first message, kept so a point read can
517    /// feed a [`StreamDecoder`] without re-reading it from disk each time.
518    schema_msg: Mutex<Option<Vec<u8>>>,
519    /// Every batch message in the file, in order. Cheap — one entry per 64 MiB.
520    batches: Mutex<Vec<Batch>>,
521    /// `oid → Located` as an [`OidTree`]. `None` until the first lookup forces
522    /// the pass that builds it, so a push-only process never pays for it.
523    index: Mutex<Option<OidTree>>,
524    /// The last few decoded batches, most recent first.
525    ///
526    /// Not an optimisation looking for a problem: without it every point read
527    /// decodes its whole batch, and the 2687-object push guard went from 4 s to
528    /// **over 100 s** — 2687 lookups × one 106 MiB batch each. A batch is
529    /// `Arc`-backed, so a hit is a clone of pointers.
530    cache: Mutex<Vec<(u32, RecordBatch)>>,
531    pending: Mutex<Vec<Pending>>,
532    pending_bytes: AtomicU64,
533    /// Bytes in the file. Kept here rather than stat'ed so the append offset and
534    /// the batch table cannot drift apart.
535    end: Mutex<u64>,
536    oid_len: Mutex<Option<usize>>,
537    policy: ExplodePolicy,
538    skipped: AtomicU64,
539    written: AtomicU64,
540    served: AtomicU64,
541    /// The subset of `served` answered by the extent `pread` fast path.
542    pread_served: AtomicU64,
543    rederived: AtomicU64,
544    absent: AtomicU64,
545}
546
547impl ExplodedArchive {
548    /// Open or create the table, with the policy `ZNIPPY_GIT_EXPLODE` names.
549    /// Reads kilobytes, never the payloads.
550    pub fn open(path: &Path) -> Result<Self> {
551        Self::open_with_policy(path, ExplodePolicy::from_env()?)
552    }
553
554    /// Open with an explicit policy — what a test or a per-tenant configuration
555    /// uses, so the setting is not reachable only through the environment.
556    pub fn open_with_policy(path: &Path, policy: ExplodePolicy) -> Result<Self> {
557        if let Some(parent) = path.parent()
558            && !parent.as_os_str().is_empty()
559        {
560            std::fs::create_dir_all(parent)
561                .with_context(|| format!("creating {}", parent.display()))?;
562        }
563        let (schema_msg, batches, end, oid_len) = Self::walk(path)?;
564        Ok(Self {
565            path: path.to_path_buf(),
566            schema_msg: Mutex::new(schema_msg),
567            batches: Mutex::new(batches),
568            index: Mutex::new(None),
569            cache: Mutex::new(Vec::new()),
570            pending: Mutex::new(Vec::new()),
571            pending_bytes: AtomicU64::new(0),
572            end: Mutex::new(end),
573            oid_len: Mutex::new(oid_len),
574            policy,
575            skipped: AtomicU64::new(0),
576            written: AtomicU64::new(0),
577            served: AtomicU64::new(0),
578            pread_served: AtomicU64::new(0),
579            rederived: AtomicU64::new(0),
580            absent: AtomicU64::new(0),
581        })
582    }
583
584    pub fn path(&self) -> &Path {
585        &self.path
586    }
587
588    /// Walk the message framing: `[0xFFFFFFFF][len u32][metadata][body]`.
589    ///
590    /// The metadata flatbuffer carries `bodyLength`, so this SEEKS over every
591    /// payload — opening a 17 GB table touches kilobytes. It also carries a
592    /// record batch's row `length`, which is how `rows()` answers without
593    /// decoding anything.
594    ///
595    /// A torn tail — a half-written message at EOF — stops the walk and keeps
596    /// what came before it, and the file is truncated to that boundary on the
597    /// next append. The table is derived, so a partial batch is re-exploded
598    /// rather than being an error a client ever sees.
599    #[allow(clippy::type_complexity)]
600    fn walk(path: &Path) -> Result<(Option<Vec<u8>>, Vec<Batch>, u64, Option<usize>)> {
601        let Ok(mut f) = File::open(path) else {
602            return Ok((None, Vec::new(), 0, None));
603        };
604        let total = f.metadata()?.len();
605        let mut schema_msg: Option<Vec<u8>> = None;
606        let mut oid_len: Option<usize> = None;
607        let mut batches = Vec::new();
608        let mut at = 0u64;
609        while at + 8 <= total {
610            f.seek(SeekFrom::Start(at))?;
611            let mut hdr = [0u8; 8];
612            if f.read_exact(&mut hdr).is_err() {
613                break;
614            }
615            if u32::from_le_bytes(hdr[0..4].try_into().unwrap()) != 0xFFFF_FFFF {
616                break;
617            }
618            let meta_len = u32::from_le_bytes(hdr[4..8].try_into().unwrap()) as u64;
619            if meta_len == 0 {
620                break; // end-of-stream marker
621            }
622            if at + 8 + meta_len > total {
623                break; // torn metadata
624            }
625            let mut meta = vec![0u8; meta_len as usize];
626            if f.read_exact(&mut meta).is_err() {
627                break;
628            }
629            let Ok(msg) = root_as_message(&meta) else {
630                break;
631            };
632            let body = msg.bodyLength().max(0) as u64;
633            let whole = 8 + meta_len + body;
634            if at + whole > total {
635                break; // torn body
636            }
637            if let Some(rb) = msg.header_as_record_batch() {
638                batches.push(Batch {
639                    at,
640                    len: whole,
641                    rows: rb.length().max(0) as u32,
642                    body: at + 8 + meta_len,
643                    pay_data: payload_data_offset(&meta).unwrap_or(EXTENT_UNAVAILABLE),
644                });
645            } else if msg.header_as_schema().is_some() {
646                schema_msg = Some({
647                    let mut whole_msg = Vec::with_capacity((8 + meta_len) as usize);
648                    whole_msg.extend_from_slice(&hdr);
649                    whole_msg.extend_from_slice(&meta);
650                    whole_msg
651                });
652                oid_len = Self::oid_len_of_schema_msg(schema_msg.as_ref().unwrap());
653            }
654            at += whole;
655        }
656        Ok((schema_msg, batches, at, oid_len))
657    }
658
659    /// The declared oid width, so a reopened table refuses a hash it cannot
660    /// hold instead of writing a batch no reader can line up.
661    fn oid_len_of_schema_msg(msg: &[u8]) -> Option<usize> {
662        let mut dec = StreamDecoder::new();
663        let mut buf = Buffer::from_vec(msg.to_vec());
664        dec.decode(&mut buf).ok()?;
665        let schema = dec.schema()?;
666        match schema.field_with_name(COL_OID).ok()?.data_type() {
667            DataType::FixedSizeBinary(n) => Some(*n as usize),
668            _ => None,
669        }
670    }
671
672    /// One resolved object with no name attached — what [`crate::exploded::PayloadSink`]
673    /// can supply.
674    pub fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
675        self.explode_at(oid, kind, None, None, payload)
676    }
677
678    /// One resolved object **with its name**: the mode and path a tree walk knows.
679    ///
680    /// A blob is reachable at many paths, so this records the first one seen and
681    /// never rewrites it — a row that already exists is not re-keyed by a second
682    /// sighting, because "which of its names" is not a question this table
683    /// pretends to answer.
684    pub fn explode_at(
685        &self,
686        oid: &[u8],
687        kind: GitObjectKind,
688        mode: Option<u32>,
689        path: Option<&str>,
690        payload: &[u8],
691    ) -> Result<()> {
692        if !self.policy.wants(kind) {
693            self.skipped.fetch_add(1, Ordering::Relaxed);
694            return Ok(());
695        }
696        {
697            let mut p = self
698                .pending
699                .lock()
700                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
701            p.push(Pending {
702                oid: oid.to_vec(),
703                code: kind_code(kind),
704                mode,
705                path: path.map(str::to_owned),
706                payload: payload.to_vec(),
707            });
708        }
709        let n = self
710            .pending_bytes
711            .fetch_add(payload.len() as u64, Ordering::AcqRel)
712            + payload.len() as u64;
713        if n >= FLUSH_BYTES as u64 {
714            self.flush()?;
715        }
716        Ok(())
717    }
718
719    fn build_batch(rows: &[Pending], oid_len: usize) -> Result<RecordBatch> {
720        let mut oid_b = FixedSizeBinaryBuilder::with_capacity(rows.len(), oid_len as i32);
721        let mut kind_b = UInt8Builder::with_capacity(rows.len());
722        let mut mode_b = UInt32Builder::with_capacity(rows.len());
723        let mut path_b = StringBuilder::new();
724        let mut pay_b = LargeBinaryBuilder::new();
725        for r in rows {
726            oid_b
727                .append_value(&r.oid)
728                .map_err(|e| anyhow::anyhow!("exploded oid column: {e}"))?;
729            kind_b.append_value(r.code);
730            mode_b.append_option(r.mode);
731            path_b.append_option(r.path.as_deref());
732            pay_b.append_value(&r.payload);
733        }
734        RecordBatch::try_new(
735            exploded_schema(oid_len),
736            vec![
737                Arc::new(oid_b.finish()),
738                Arc::new(kind_b.finish()),
739                Arc::new(mode_b.finish()),
740                Arc::new(path_b.finish()),
741                Arc::new(pay_b.finish()),
742            ],
743        )
744        .context("building the exploded batch")
745    }
746
747    /// Append the buffered rows as ONE batch message. Nothing already in the
748    /// file is read or rewritten.
749    pub fn flush(&self) -> Result<u64> {
750        let rows = {
751            let mut p = self
752                .pending
753                .lock()
754                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
755            if p.is_empty() {
756                return Ok(0);
757            }
758            self.pending_bytes.store(0, Ordering::Release);
759            std::mem::take(&mut *p)
760        };
761        let n = rows.len() as u64;
762        let width = rows[0].oid.len();
763        anyhow::ensure!(
764            rows.iter().all(|r| r.oid.len() == width),
765            "one exploded batch cannot hold two hash widths"
766        );
767        {
768            let mut w = self
769                .oid_len
770                .lock()
771                .map_err(|_| anyhow::anyhow!("exploded oid width poisoned"))?;
772            match *w {
773                Some(have) => anyhow::ensure!(
774                    have == width,
775                    "this table holds {have}-byte oids and a {width}-byte one arrived"
776                ),
777                None => *w = Some(width),
778            }
779        }
780
781        let batch = Self::build_batch(&rows, width)?;
782        let opts = IpcWriteOptions::default();
783        let ipc = IpcDataGenerator::default();
784        let mut tracker = DictionaryTracker::new(false);
785
786        // The file is opened per flush rather than held: an append writes at a
787        // known offset and closing between batches is what makes a crashed
788        // process leave a file whose last complete message is intact.
789        let mut end = self
790            .end
791            .lock()
792            .map_err(|_| anyhow::anyhow!("exploded end poisoned"))?;
793        let mut f = OpenOptions::new()
794            .create(true)
795            .read(true)
796            .write(true)
797            .truncate(false)
798            .open(&self.path)
799            .with_context(|| format!("opening {}", self.path.display()))?;
800        // Drop anything after the last complete message before appending — a
801        // torn tail from a previous crash must not sit in the middle of the
802        // stream, where every later batch would be unreachable.
803        if f.metadata()?.len() != *end {
804            f.set_len(*end)
805                .with_context(|| format!("truncating {} to {}", self.path.display(), *end))?;
806        }
807        f.seek(SeekFrom::Start(*end))?;
808
809        let mut schema_guard = self
810            .schema_msg
811            .lock()
812            .map_err(|_| anyhow::anyhow!("exploded schema poisoned"))?;
813        if schema_guard.is_none() {
814            let enc =
815                ipc.schema_to_bytes_with_dictionary_tracker(&exploded_schema(width), &mut tracker, &opts);
816            let mut msg = Vec::new();
817            let (meta, body) = write_message(&mut msg, enc, &opts)
818                .map_err(|e| anyhow::anyhow!("encoding the exploded schema: {e}"))?;
819            debug_assert_eq!(meta + body, msg.len());
820            f.write_all(&msg).context("writing the exploded schema")?;
821            *end += msg.len() as u64;
822            *schema_guard = Some(msg);
823        }
824        drop(schema_guard);
825
826        let (dicts, enc) = ipc
827            .encode(&batch, &mut tracker, &opts, &mut Default::default())
828            .map_err(|e| anyhow::anyhow!("encoding an exploded batch: {e}"))?;
829        anyhow::ensure!(
830            dicts.is_empty(),
831            "the exploded schema has no dictionary columns, yet {} arrived",
832            dicts.len()
833        );
834        let mut msg = Vec::new();
835        let (meta, body) = write_message(&mut msg, enc, &opts)
836            .map_err(|e| anyhow::anyhow!("encoding an exploded batch: {e}"))?;
837        debug_assert_eq!(meta + body, msg.len());
838        f.write_all(&msg).context("appending an exploded batch")?;
839
840        let placed = Batch {
841            at: *end,
842            len: msg.len() as u64,
843            rows: rows.len() as u32,
844            // `meta` from `write_message` is the 8-byte prefix plus the padded
845            // metadata, so the body begins exactly `meta` bytes in — and the
846            // metadata slice between prefix and body is what the offset helper
847            // parses, the same bytes `walk` reads back after a reopen.
848            body: *end + meta as u64,
849            pay_data: payload_data_offset(&msg[8..meta]).unwrap_or(EXTENT_UNAVAILABLE),
850        };
851        *end += msg.len() as u64;
852
853        // Recorded while the append lock is still held, so the batch table stays
854        // in file order under N concurrent exploders. Correctness would survive
855        // it being out of order — a lookup addresses a batch by its slot, not by
856        // its offset — but a table that reads back in file order is one a human
857        // can check against the file with `od`.
858        let batch_no = {
859            let mut b = self
860                .batches
861                .lock()
862                .map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
863            b.push(placed);
864            (b.len() - 1) as u32
865        };
866        drop(end);
867
868        // Only extend an index that already exists. Building one here would make
869        // the first push pay for a structure no reader has asked for.
870        let mut idx = self
871            .index
872            .lock()
873            .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
874        if let Some(old) = idx.take() {
875            // Retired oids are collected and applied in ONE pass. Removing them
876            // one at a time is quadratic, and a `gc` retires millions.
877            let mut retired: std::collections::HashSet<&[u8]> = std::collections::HashSet::new();
878            for r in rows.iter() {
879                if kind_of(r.code).is_none() {
880                    retired.insert(r.oid.as_slice());
881                }
882            }
883            let mut b = OidTreeBuilder::with_capacity(old.len() + rows.len());
884            for (oid, loc) in old.iter() {
885                if !retired.contains(oid) {
886                    b.push(oid, Some(loc))?;
887                }
888            }
889            let pay = batch
890                .column_by_name(COL_PAYLOAD)
891                .and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
892                .context("the exploded table has no payload column")?;
893            // Freshly built, never sliced, so these offsets are byte-for-byte
894            // what the encoder just wrote into the file — the same equivalence
895            // `index_now` relies on after a reopen.
896            let off = pay.value_offsets();
897            for (i, r) in rows.iter().enumerate() {
898                if let Some(kind) = kind_of(r.code)
899                    && !retired.contains(r.oid.as_slice())
900                {
901                    let (pay_at, pay_len) = if placed.pay_data == EXTENT_UNAVAILABLE {
902                        (EXTENT_UNAVAILABLE, 0)
903                    } else {
904                        (
905                            placed.body + placed.pay_data + off[i] as u64,
906                            (off[i + 1] - off[i]) as u64,
907                        )
908                    };
909                    b.push(
910                        &r.oid,
911                        Some(Located { batch: batch_no, row: i as u32, kind, pay_at, pay_len }),
912                    )?;
913                }
914            }
915            // `Keep::First`, because the surviving entries were pushed before
916            // this batch's rows: an oid already in the index keeps the row it
917            // already had, which is what `explode_at` documents about names.
918            *idx = Some(b.finish(Keep::First));
919        }
920        self.written.fetch_add(n, Ordering::Relaxed);
921        Ok(n)
922    }
923
924    /// How many decoded batches stay resident. Four × [`FLUSH_BYTES`] is the
925    /// worst case, and a batch is `Arc`-backed so a hit copies pointers.
926    const CACHE: usize = 4;
927
928    /// Read one batch message off disk and decode it — or hand back the decoded
929    /// copy if it is one of the last [`Self::CACHE`].
930    ///
931    /// Two `pread`s on a miss: the schema message and the batch. It never reads
932    /// the batches in front of the one it wants — the property a stream reader
933    /// iterating from byte zero does not have.
934    fn read_batch(&self, no: u32) -> Result<RecordBatch> {
935        {
936            let mut c = self
937                .cache
938                .lock()
939                .map_err(|_| anyhow::anyhow!("exploded batch cache poisoned"))?;
940            if let Some(at) = c.iter().position(|(n, _)| *n == no) {
941                let hit = c.remove(at);
942                let batch = hit.1.clone();
943                c.insert(0, hit);
944                return Ok(batch);
945            }
946        }
947        let placed = {
948            let b = self
949                .batches
950                .lock()
951                .map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
952            *b.get(no as usize)
953                .with_context(|| format!("exploded batch {no} is not in this table"))?
954        };
955        let schema_msg = {
956            let s = self
957                .schema_msg
958                .lock()
959                .map_err(|_| anyhow::anyhow!("exploded schema poisoned"))?;
960            s.clone()
961                .context("the exploded table has batches but no schema message")?
962        };
963        let f = File::open(&self.path)
964            .with_context(|| format!("opening {}", self.path.display()))?;
965        let mut raw = vec![0u8; placed.len as usize];
966        f.read_exact_at(&mut raw, placed.at)
967            .with_context(|| format!("reading exploded batch {no}"))?;
968
969        let mut dec = StreamDecoder::new();
970        let mut head = Buffer::from_vec(schema_msg);
971        dec.decode(&mut head)
972            .map_err(|e| anyhow::anyhow!("decoding the exploded schema: {e}"))?;
973        let mut body = Buffer::from_vec(raw);
974        let batch = dec
975            .decode(&mut body)
976            .map_err(|e| anyhow::anyhow!("decoding exploded batch {no}: {e}"))?
977            .with_context(|| format!("exploded batch {no} decoded to no rows"))?;
978        {
979            let mut c = self
980                .cache
981                .lock()
982                .map_err(|_| anyhow::anyhow!("exploded batch cache poisoned"))?;
983            c.retain(|(n, _)| *n != no);
984            c.insert(0, (no, batch.clone()));
985            c.truncate(Self::CACHE);
986        }
987        Ok(batch)
988    }
989
990    /// Build `oid → Located` with one sequential pass, applying every
991    /// [`TOMBSTONE`] in file order so a retired oid stays retired across a
992    /// reopen. Called on the first lookup, never on open.
993    fn index_now(&self) -> Result<()> {
994        {
995            let idx = self
996                .index
997                .lock()
998                .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
999            if idx.is_some() {
1000                return Ok(());
1001            }
1002        }
1003        let (count, rows, placed) = {
1004            let b = self
1005                .batches
1006                .lock()
1007                .map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
1008            // The row total is in the batch metadata the framing walk already
1009            // read, so the builder's arrays are sized once instead of doubling
1010            // their way to 11.7 million.
1011            (b.len() as u32, b.iter().map(|x| x.rows as usize).sum::<usize>(), b.clone())
1012        };
1013        // Every row is recorded as a sighting and the fold happens once, at
1014        // `finish`. The `HashMap<Vec<u8>, Located>` this replaced allocated —
1015        // and then dropped — one `Vec<u8>` per oid on a corpus with 11.7 M of
1016        // them, to compute an answer a single stable sort gives.
1017        let mut fold = OidTreeBuilder::with_capacity(rows);
1018        for no in 0..count {
1019            let batch = self.read_batch(no)?;
1020            let oid = batch
1021                .column_by_name(COL_OID)
1022                .and_then(|c| c.as_any().downcast_ref::<FixedSizeBinaryArray>())
1023                .context("the exploded table has no oid column")?;
1024            let kinds = batch
1025                .column_by_name(COL_TYPE)
1026                .and_then(|c| c.as_any().downcast_ref::<UInt8Array>())
1027                .context("the exploded table has no object_type column")?;
1028            let pay = batch
1029                .column_by_name(COL_PAYLOAD)
1030                .and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
1031                .context("the exploded table has no payload column")?;
1032            // The decoded offsets carry the SAME values the file's offsets
1033            // buffer holds — a StreamDecoder copies buffers verbatim — so the
1034            // extent needs no second parse of the body.
1035            let off = pay.value_offsets();
1036            let b = placed[no as usize];
1037            for i in 0..batch.num_rows() {
1038                // A tombstone is pushed as a sighting with no location, not
1039                // skipped: batches are walked in file order and rows in row
1040                // order, so `Keep::Last` gives the last word about an oid — and
1041                // when that word is a tombstone the oid is dropped, which is
1042                // what makes the tombstone durable rather than advisory.
1043                fold.push(
1044                    oid.value(i),
1045                    kind_of(kinds.value(i)).map(|kind| {
1046                        let (pay_at, pay_len) = if b.pay_data == EXTENT_UNAVAILABLE {
1047                            (EXTENT_UNAVAILABLE, 0)
1048                        } else {
1049                            (
1050                                b.body + b.pay_data + off[i] as u64,
1051                                (off[i + 1] - off[i]) as u64,
1052                            )
1053                        };
1054                        Located { batch: no, row: i as u32, kind, pay_at, pay_len }
1055                    }),
1056                )?;
1057            }
1058        }
1059        let out = fold.finish(Keep::Last);
1060        let mut idx = self
1061            .index
1062            .lock()
1063            .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1064        if idx.is_none() {
1065            *idx = Some(out);
1066        }
1067        Ok(())
1068    }
1069
1070    fn find(&self, oid: &[u8]) -> Result<Option<Located>> {
1071        self.index_now()?;
1072        let idx = self
1073            .index
1074            .lock()
1075            .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1076        Ok(idx.as_ref().expect("index_now built it").find(oid))
1077    }
1078
1079    /// The object's kind and bytes, or `None` — *fall back and rebuild*, never
1080    /// *wrong*.
1081    pub fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
1082        // A buffered row has not been written yet, and a read must not miss it:
1083        // an "absent" that is really "not flushed" sends the caller off to
1084        // re-resolve a whole pack for an object we are holding in RAM.
1085        {
1086            let p = self
1087                .pending
1088                .lock()
1089                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1090            if let Some(r) = p.iter().rev().find(|r| r.oid == oid) {
1091                let Some(kind) = kind_of(r.code) else {
1092                    // The newest buffered word about this oid retires it.
1093                    self.absent.fetch_add(1, Ordering::Relaxed);
1094                    return Ok(None);
1095                };
1096                self.served.fetch_add(1, Ordering::Relaxed);
1097                return Ok(Some((kind, r.payload.clone())));
1098            }
1099        }
1100        let Some(loc) = self.find(oid)? else {
1101            self.absent.fetch_add(1, Ordering::Relaxed);
1102            return Ok(None);
1103        };
1104        // ── the point read IS a point read ──────────────────────────────────
1105        //
1106        // One pread of exactly the payload's bytes, addressed by the extent the
1107        // index carries. The batch-decode below is the FALLBACK — correct for a
1108        // compressed or foreign-layout batch — not the path a healthy table
1109        // takes: taking it per object is what turned a rust-lang/rust
1110        // connectivity walk into hours of one-core batch decoding (2026-08-19).
1111        if loc.pay_at != EXTENT_UNAVAILABLE {
1112            let f = File::open(&self.path)
1113                .with_context(|| format!("opening {}", self.path.display()))?;
1114            let mut buf = vec![0u8; loc.pay_len as usize];
1115            f.read_exact_at(&mut buf, loc.pay_at)
1116                .with_context(|| format!("preading payload at {}", loc.pay_at))?;
1117            self.served.fetch_add(1, Ordering::Relaxed);
1118            self.pread_served.fetch_add(1, Ordering::Relaxed);
1119            return Ok(Some((loc.kind, buf)));
1120        }
1121        let batch = self.read_batch(loc.batch)?;
1122        let pay = batch
1123            .column_by_name(COL_PAYLOAD)
1124            .and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
1125            .context("the exploded table has no payload column")?;
1126        self.served.fetch_add(1, Ordering::Relaxed);
1127        Ok(Some((loc.kind, pay.value(loc.row as usize).to_vec())))
1128    }
1129
1130    /// Point reads served by the extent `pread` rather than a batch decode —
1131    /// how a test proves the fast path is the one actually taken, instead of
1132    /// timing something.
1133    pub fn pread_served(&self) -> u64 {
1134        self.pread_served.load(Ordering::Relaxed)
1135    }
1136
1137    /// The name this object was first exploded at, if any caller knew one.
1138    /// `None` is "not recorded", never "at the root".
1139    pub fn name(&self, oid: &[u8]) -> Result<Option<(Option<u32>, Option<String>)>> {
1140        {
1141            let p = self
1142                .pending
1143                .lock()
1144                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1145            if let Some(r) = p.iter().rev().find(|r| r.oid == oid) {
1146                return match kind_of(r.code) {
1147                    Some(_) => Ok(Some((r.mode, r.path.clone()))),
1148                    None => Ok(None),
1149                };
1150            }
1151        }
1152        let Some(loc) = self.find(oid)? else {
1153            return Ok(None);
1154        };
1155        let batch = self.read_batch(loc.batch)?;
1156        let mode = batch
1157            .column_by_name(COL_MODE)
1158            .and_then(|c| c.as_any().downcast_ref::<UInt32Array>())
1159            .context("the exploded table has no mode column")?;
1160        let path = batch
1161            .column_by_name(COL_PATH)
1162            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
1163            .context("the exploded table has no path column")?;
1164        let i = loc.row as usize;
1165        Ok(Some((
1166            (!mode.is_null(i)).then(|| mode.value(i)),
1167            (!path.is_null(i)).then(|| path.value(i).to_owned()),
1168        )))
1169    }
1170
1171    pub fn has(&self, oid: &[u8]) -> Result<bool> {
1172        {
1173            let p = self
1174                .pending
1175                .lock()
1176                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1177            if let Some(r) = p.iter().rev().find(|r| r.oid == oid) {
1178                return Ok(kind_of(r.code).is_some());
1179            }
1180        }
1181        Ok(self.find(oid)?.is_some())
1182    }
1183
1184    /// Every oid of one kind, with its payload. The graph fold's input.
1185    ///
1186    /// Driven by the **index**, not by a scan of the file, because the file
1187    /// holds superseded and retired rows too. A scan that read the file directly
1188    /// is exactly how `gc`'d commits came back after a reopen.
1189    pub fn of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
1190        self.flush()?;
1191        self.index_now()?;
1192        // Grouped by batch so each one is decoded once, in file order.
1193        let mut by_batch: std::collections::BTreeMap<u32, Vec<(u32, Vec<u8>)>> =
1194            std::collections::BTreeMap::new();
1195        {
1196            let idx = self
1197                .index
1198                .lock()
1199                .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1200            for (oid, loc) in idx.as_ref().expect("index_now built it").iter() {
1201                if loc.kind == kind {
1202                    by_batch.entry(loc.batch).or_default().push((loc.row, oid.to_vec()));
1203                }
1204            }
1205        }
1206        let mut out = Vec::new();
1207        for (no, mut wanted) in by_batch {
1208            wanted.sort_unstable_by_key(|(r, _)| *r);
1209            let batch = self.read_batch(no)?;
1210            let pay = batch
1211                .column_by_name(COL_PAYLOAD)
1212                .and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
1213                .context("the exploded table has no payload column")?;
1214            for (row, oid) in wanted {
1215                out.push((oid, pay.value(row as usize).to_vec()));
1216            }
1217        }
1218        Ok(out)
1219    }
1220
1221    /// Retire every oid `live` rejects, **durably**.
1222    ///
1223    /// The file is append-only, so this appends a [`TOMBSTONE`] row per retired
1224    /// oid rather than deleting anything. The bytes of the dead rows stay until
1225    /// the table is rebuilt — reclaiming them means rewriting up to 17 GB, which
1226    /// is a job with its own proof obligations and not something to smuggle into
1227    /// `gc` — but the *rows* are gone from every reader, including a reopened
1228    /// one.
1229    ///
1230    /// Dropping only from the in-memory index is what a first cut did. Seen RED
1231    /// by `store::tests::a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened`:
1232    /// **"the graph after gc + reopen holds 551 commits, not the one live one"**.
1233    pub fn retain(&self, live: &dyn Fn(&[u8]) -> bool) -> Result<u64> {
1234        self.flush()?;
1235        self.index_now()?;
1236        let dead: Vec<Vec<u8>> = {
1237            let idx = self
1238                .index
1239                .lock()
1240                .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1241            idx.as_ref()
1242                .expect("index_now built it")
1243                .iter()
1244                .filter(|(oid, _)| !live(oid))
1245                .map(|(oid, _)| oid.to_vec())
1246                .collect()
1247        };
1248        if dead.is_empty() {
1249            return Ok(0);
1250        }
1251        {
1252            let mut p = self
1253                .pending
1254                .lock()
1255                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1256            for oid in &dead {
1257                p.push(Pending {
1258                    oid: oid.clone(),
1259                    code: TOMBSTONE,
1260                    mode: None,
1261                    path: None,
1262                    payload: Vec::new(),
1263                });
1264            }
1265        }
1266        // Written before returning, so a `gc` that says it retired rows has
1267        // already said so on disk.
1268        self.flush()?;
1269        Ok(dead.len() as u64)
1270    }
1271
1272    /// Rows in the table. Answered from batch metadata when no index has been
1273    /// built, so startup does not read a payload.
1274    pub fn rows(&self) -> Result<u64> {
1275        let pending = {
1276            let p = self
1277                .pending
1278                .lock()
1279                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1280            p.len() as u64
1281        };
1282        {
1283            let idx = self
1284                .index
1285                .lock()
1286                .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1287            if let Some(idx) = idx.as_ref() {
1288                return Ok(idx.len() as u64 + pending);
1289            }
1290        }
1291        let b = self
1292            .batches
1293            .lock()
1294            .map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
1295        Ok(b.iter().map(|x| x.rows as u64).sum::<u64>() + pending)
1296    }
1297
1298    /// Bytes the table occupies — the number redb could not keep anywhere near
1299    /// its payload total.
1300    pub fn disk_bytes(&self) -> u64 {
1301        std::fs::metadata(&self.path).map(|m| m.len()).unwrap_or(0)
1302    }
1303
1304    pub fn note_rederived(&self) {
1305        self.rederived.fetch_add(1, Ordering::Relaxed);
1306    }
1307
1308    /// What this table is configured to keep.
1309    pub fn policy(&self) -> ExplodePolicy {
1310        self.policy
1311    }
1312
1313    /// Objects the policy declined, COUNTED: a table short because of policy and
1314    /// one short because of a fault must not look alike to `adopt_journal`.
1315    pub fn skipped(&self) -> u64 {
1316        self.skipped.load(Ordering::Relaxed)
1317    }
1318
1319    pub fn stats(&self) -> Stats {
1320        Stats {
1321            rows: self.rows().unwrap_or(0),
1322            written: self.written.load(Ordering::Relaxed),
1323            served: self.served.load(Ordering::Relaxed),
1324            rederived: self.rederived.load(Ordering::Relaxed),
1325            absent: self.absent.load(Ordering::Relaxed),
1326        }
1327    }
1328
1329    /// The engine-wide counter type, so no caller can tell the medium apart by
1330    /// its instruments.
1331    pub fn engine_stats(&self) -> crate::exploded::ExplodedStats {
1332        let s = self.stats();
1333        crate::exploded::ExplodedStats {
1334            rows: s.rows,
1335            written: s.written,
1336            served: s.served,
1337            rederived: s.rederived,
1338            absent: s.absent,
1339        }
1340    }
1341}
1342
1343impl crate::exploded::PayloadSink for ExplodedArchive {
1344    fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
1345        ExplodedArchive::explode(self, oid, kind, payload)
1346    }
1347}
1348
1349// ════════════════════════════════════════════════════════════════════════════
1350// The policy: what gets exploded at all
1351// ════════════════════════════════════════════════════════════════════════════
1352
1353/// **How much of a pack is resolved into the exploded table.**
1354///
1355/// Three values and not a boolean, because a flat off-switch reintroduces a
1356/// correctness bug this table exists to fix. Commit and tree payloads used to
1357/// live only in RAM, so a **clean** shutdown came back with every pack's
1358/// `indexed` bit legitimately set and an EMPTY graph — MEASURED on oden
1359/// 2026-08-08: 2687 rows and **0 of 551 commits**, `reachable()` returning a
1360/// commit instead of its closure and `gc` seeing an empty live set.
1361///
1362/// MEASURED on `linux.git`: resolved content is 16.8 GB against a 6.4 GB pack,
1363/// and blobs are nearly all of it. So [`Graph`](ExplodePolicy::Graph) buys the
1364/// correctness fix for a small fraction of the disk, and
1365/// [`Full`](ExplodePolicy::Full) — one-lookup content reads, thin-pack bases,
1366/// and the path columns that make the table a filesystem — is the part worth
1367/// charging for.
1368///
1369/// **The default is [`Full`](ExplodePolicy::Full), and that is a compatibility
1370/// choice rather than a recommendation.** §14's `indexed` bit is derived by
1371/// comparing this table's row count against the index's, so a table that is
1372/// *deliberately* short is indistinguishable, on disk, from one that was
1373/// dropped — see [`crate::git_ops::Absorber::adopt_journal`], which weakens that
1374/// comparison under `Graph` and cannot make it exact without a durable
1375/// commit-and-tree count. Changing the default would change what a reopen does
1376/// on every existing store; the setting is the knob, not the default.
1377#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1378pub enum ExplodePolicy {
1379    /// Store nothing. Reads re-resolve, and every pack is re-queued on reopen —
1380    /// which is what rebuilds the graph, since nothing it is folded from is on
1381    /// disk.
1382    Off,
1383    /// Commits and trees only — the graph's inputs, which is everything a reopen
1384    /// needs to be *correct*. Blobs re-resolve.
1385    Graph,
1386    /// Every object.
1387    #[default]
1388    Full,
1389}
1390
1391impl ExplodePolicy {
1392    pub const fn wants(self, kind: GitObjectKind) -> bool {
1393        match self {
1394            ExplodePolicy::Off => false,
1395            ExplodePolicy::Graph => matches!(kind, GitObjectKind::Commit | GitObjectKind::Tree),
1396            ExplodePolicy::Full => true,
1397        }
1398    }
1399    pub const fn as_str(self) -> &'static str {
1400        match self {
1401            ExplodePolicy::Off => "off",
1402            ExplodePolicy::Graph => "graph",
1403            ExplodePolicy::Full => "full",
1404        }
1405    }
1406    /// An unknown value is an ERROR, never a silent fall back to the default: a
1407    /// typo that quietly disables a paid tier is not noticed until a bill is
1408    /// wrong.
1409    pub fn parse(s: &str) -> Result<Self> {
1410        match s.trim().to_ascii_lowercase().as_str() {
1411            "off" | "none" => Ok(ExplodePolicy::Off),
1412            "graph" | "commits-and-trees" => Ok(ExplodePolicy::Graph),
1413            "full" | "all" => Ok(ExplodePolicy::Full),
1414            other => anyhow::bail!("unknown explode policy {other:?}; expected off, graph or full"),
1415        }
1416    }
1417    /// Reads `ZNIPPY_GIT_EXPLODE` ([`crate::arms::ENV_EXPLODE`]) through the
1418    /// crate's one counted door, once per store open.
1419    pub fn from_env() -> Result<Self> {
1420        match crate::arms::read_env(crate::arms::ENV_EXPLODE) {
1421            Some(v) => Self::parse(&v),
1422            None => Ok(Self::default()),
1423        }
1424    }
1425}
1426
1427#[cfg(test)]
1428mod tests {
1429    use super::*;
1430    use crate::exploded::PayloadSink as _;
1431
1432    fn tmp(name: &str) -> PathBuf {
1433        let d = std::env::temp_dir().join(format!("exploded-arrow-{}-{name}", std::process::id()));
1434        let _ = std::fs::remove_dir_all(&d);
1435        std::fs::create_dir_all(&d).unwrap();
1436        d.join("objects.exploded")
1437    }
1438
1439    fn oid(n: u8) -> [u8; 20] {
1440        [n; 20]
1441    }
1442
1443    /// ★ ONE TABLE, and the payload is IN the row.
1444    ///
1445    /// Red-first: drop the payload column and nothing below compiles; widen
1446    /// `payload` to `Binary` and this fails on the type it asserts.
1447    #[test]
1448    fn the_table_holds_the_payload_and_a_name_in_one_row() {
1449        let s = exploded_schema(20);
1450        let names: Vec<&str> = s.fields().iter().map(|f| f.name().as_str()).collect();
1451        assert_eq!(names, vec![COL_OID, COL_TYPE, COL_MODE, COL_PATH, COL_PAYLOAD]);
1452        assert_eq!(
1453            s.field_with_name(COL_PAYLOAD).unwrap().data_type(),
1454            &DataType::LargeBinary,
1455            "Binary offsets are i32 and would cap one batch at 2 GiB of payload"
1456        );
1457        assert!(
1458            s.field_with_name(COL_PATH).unwrap().is_nullable(),
1459            "a bare pack resolve knows no path; null must be a legal answer"
1460        );
1461    }
1462
1463    /// ★ THE WHOLE POINT: the file stays near the payload total.
1464    ///
1465    /// redb held 16.8 GB of kernel payload in a 204 GB file — 12×. An Arrow IPC
1466    /// stream has no pages to rewrite, so the overhead is framing. Red-first:
1467    /// put this back on a copy-on-write B-tree and it fails.
1468    #[test]
1469    fn the_file_stays_close_to_the_payload_total() {
1470        let t = ExplodedArchive::open_with_policy(&tmp("size"), ExplodePolicy::Full).unwrap();
1471        let payload = vec![7u8; 4096];
1472        let n = 2000u64;
1473        for i in 0..n {
1474            let mut o = [0u8; 20];
1475            o[..8].copy_from_slice(&i.to_le_bytes());
1476            t.explode(&o, GitObjectKind::Blob, &payload).unwrap();
1477        }
1478        t.flush().unwrap();
1479        let payload_total = n * payload.len() as u64;
1480        let on_disk = t.disk_bytes();
1481        assert!(
1482            on_disk < payload_total + payload_total / 4,
1483            "{on_disk} bytes on disk against {payload_total} of payload — the medium is amplifying"
1484        );
1485        assert_eq!(t.rows().unwrap(), n);
1486    }
1487
1488    /// ★ A point read is one small `pread` of the payload's own bytes, and it
1489    /// is the path ACTUALLY taken — counted, not timed.
1490    ///
1491    /// Red-first, measured on a rust-lang/rust push (2026-08-19): without the
1492    /// extent, every `content` decoded its whole 64 MiB batch against a 4-slot
1493    /// cache, and the pre-ack connectivity walk — graph order, so effectively
1494    /// random across ~574 batches — decoded 6.7 GB/s of page cache to serve
1495    /// ~98 objects/s, one core pinned for hours. Both index-building paths are
1496    /// exercised: the in-process flush extension AND `index_now` after a
1497    /// reopen, against multi-batch tables with varied payload sizes, so a
1498    /// mis-computed extent returns the WRONG BYTES here rather than in a
1499    /// customer's clone.
1500    #[test]
1501    fn a_point_read_preads_the_exact_payload_without_decoding_the_batch() {
1502        let path = tmp("pread");
1503        let pay = |b: u8, i: u8| vec![b ^ i; 100 + i as usize * 7];
1504        let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
1505        // Batch 0, then force the index to exist so batch 1 goes through the
1506        // flush EXTENSION path rather than a later index_now.
1507        for i in 0..40u8 {
1508            t.explode(&[i; 20], GitObjectKind::Blob, &pay(0, i)).unwrap();
1509        }
1510        t.flush().unwrap();
1511        assert!(t.content(&[0u8; 20]).unwrap().is_some(), "build the index");
1512        for i in 40..80u8 {
1513            t.explode(&[i; 20], GitObjectKind::Blob, &pay(1, i)).unwrap();
1514        }
1515        t.flush().unwrap();
1516
1517        let before = t.pread_served();
1518        for i in 0..80u8 {
1519            let (kind, got) = t.content(&[i; 20]).unwrap().expect("every oid is live");
1520            assert_eq!(kind, GitObjectKind::Blob);
1521            let want = if i < 40 { pay(0, i) } else { pay(1, i) };
1522            assert_eq!(got, want, "oid {i}: the pread returned some OTHER bytes");
1523        }
1524        assert_eq!(
1525            t.pread_served() - before,
1526            80,
1527            "a read fell back to the batch decode — the extent was not computed"
1528        );
1529
1530        // Reopen: the framing walk + index_now must reproduce the same extents
1531        // from the file alone.
1532        drop(t);
1533        let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
1534        for i in 0..80u8 {
1535            let (_, got) = t.content(&[i; 20]).unwrap().expect("survives a reopen");
1536            let want = if i < 40 { pay(0, i) } else { pay(1, i) };
1537            assert_eq!(got, want, "oid {i} after reopen");
1538        }
1539        assert_eq!(t.pread_served(), 80, "the reopened table must pread too");
1540    }
1541
1542    #[test]
1543    fn a_payload_round_trips_through_the_table() {
1544        let t = ExplodedArchive::open_with_policy(&tmp("rt"), ExplodePolicy::Full).unwrap();
1545        t.explode(&oid(1), GitObjectKind::Commit, b"tree deadbeef\n")
1546            .unwrap();
1547        t.explode(&oid(2), GitObjectKind::Blob, b"hello world")
1548            .unwrap();
1549        t.flush().unwrap();
1550        assert_eq!(
1551            t.content(&oid(2)).unwrap().unwrap(),
1552            (GitObjectKind::Blob, b"hello world".to_vec())
1553        );
1554        assert_eq!(
1555            t.content(&oid(1)).unwrap().unwrap(),
1556            (GitObjectKind::Commit, b"tree deadbeef\n".to_vec())
1557        );
1558        assert!(t.has(&oid(1)).unwrap());
1559        assert!(!t.has(&oid(9)).unwrap());
1560    }
1561
1562    /// The row is a FILE: mode and path survive the round trip.
1563    #[test]
1564    fn a_row_can_carry_the_name_it_was_seen_at() {
1565        let t = ExplodedArchive::open_with_policy(&tmp("named"), ExplodePolicy::Full).unwrap();
1566        t.explode_at(
1567            &oid(1),
1568            GitObjectKind::Blob,
1569            Some(0o100755),
1570            Some("scripts/build.sh"),
1571            b"#!/bin/sh\n",
1572        )
1573        .unwrap();
1574        t.explode(&oid(2), GitObjectKind::Blob, b"anonymous").unwrap();
1575        t.flush().unwrap();
1576        assert_eq!(
1577            t.name(&oid(1)).unwrap().unwrap(),
1578            (Some(0o100755), Some("scripts/build.sh".to_owned()))
1579        );
1580        assert_eq!(
1581            t.name(&oid(2)).unwrap().unwrap(),
1582            (None, None),
1583            "null is 'not recorded', and must not read as a real name"
1584        );
1585    }
1586
1587    /// Reopening finds every row without reading a payload, and appends continue
1588    /// the SAME stream rather than starting a second one.
1589    #[test]
1590    fn the_table_survives_a_reopen_and_keeps_appending() {
1591        let p = tmp("reopen");
1592        {
1593            let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1594            t.explode(&oid(5), GitObjectKind::Tree, b"100644 f\0").unwrap();
1595            t.flush().unwrap();
1596        }
1597        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1598        assert_eq!(t.rows().unwrap(), 1, "row count comes from batch metadata");
1599        t.explode(&oid(6), GitObjectKind::Blob, b"second session").unwrap();
1600        t.flush().unwrap();
1601        assert_eq!(t.rows().unwrap(), 2);
1602        assert_eq!(
1603            t.content(&oid(5)).unwrap().unwrap(),
1604            (GitObjectKind::Tree, b"100644 f\0".to_vec())
1605        );
1606        assert_eq!(
1607            t.content(&oid(6)).unwrap().unwrap(),
1608            (GitObjectKind::Blob, b"second session".to_vec())
1609        );
1610
1611        // And a third open sees both — one schema message, two batches.
1612        let again = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1613        assert_eq!(again.rows().unwrap(), 2);
1614        assert_eq!(again.batches.lock().unwrap().len(), 2);
1615    }
1616
1617    /// A read must not miss a row that is buffered but not yet written — an
1618    /// "absent" that is really "not flushed" sends the caller off to re-resolve
1619    /// a whole pack for an object we are holding.
1620    #[test]
1621    fn an_unflushed_row_is_still_served() {
1622        let t = ExplodedArchive::open_with_policy(&tmp("unflushed"), ExplodePolicy::Full).unwrap();
1623        t.explode(&oid(4), GitObjectKind::Commit, b"buffered").unwrap();
1624        assert_eq!(
1625            t.content(&oid(4)).unwrap().unwrap(),
1626            (GitObjectKind::Commit, b"buffered".to_vec())
1627        );
1628        assert!(t.has(&oid(4)).unwrap());
1629    }
1630
1631    #[test]
1632    fn an_absent_oid_is_none_and_counted() {
1633        let t = ExplodedArchive::open_with_policy(&tmp("absent"), ExplodePolicy::Full).unwrap();
1634        assert!(t.content(&oid(3)).unwrap().is_none());
1635        assert_eq!(t.stats().absent, 1);
1636        assert_eq!(t.stats().served, 0);
1637    }
1638
1639    #[test]
1640    fn of_kind_selects_by_kind() {
1641        let t = ExplodedArchive::open_with_policy(&tmp("kind"), ExplodePolicy::Full).unwrap();
1642        t.explode(&oid(1), GitObjectKind::Commit, b"c1").unwrap();
1643        t.explode(&oid(2), GitObjectKind::Blob, b"bb").unwrap();
1644        t.explode(&oid(3), GitObjectKind::Commit, b"c2").unwrap();
1645        let commits = t.of_kind(GitObjectKind::Commit).unwrap();
1646        assert_eq!(commits.len(), 2);
1647        assert!(commits.iter().all(|(_, p)| p[0] == b'c'));
1648    }
1649
1650    /// ★ A RETIRED ROW STAYS RETIRED ACROSS A REOPEN.
1651    ///
1652    /// The file is append-only, so `retain` writes a [`TOMBSTONE`] rather than
1653    /// deleting. Seen RED with an index-only `retain`: the reopen below rebuilds
1654    /// from the file and brought the dead row back — which is
1655    /// `store::tests::a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened`
1656    /// failing with "551 commits, not the one live one", reached from here in
1657    /// three lines instead of a whole gc.
1658    #[test]
1659    fn retain_drops_only_what_is_dead_and_the_drop_survives_a_reopen() {
1660        let p = tmp("retain");
1661        {
1662            let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1663            t.explode(&oid(1), GitObjectKind::Blob, b"a").unwrap();
1664            t.explode(&oid(2), GitObjectKind::Blob, b"b").unwrap();
1665            t.flush().unwrap();
1666            assert_eq!(t.retain(&|o: &[u8]| o[0] == 1).unwrap(), 1);
1667            assert!(t.has(&oid(1)).unwrap());
1668            assert!(!t.has(&oid(2)).unwrap());
1669            assert_eq!(
1670                t.of_kind(GitObjectKind::Blob).unwrap().len(),
1671                1,
1672                "a scan must not resurrect what retain dropped"
1673            );
1674        }
1675        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1676        assert!(t.has(&oid(1)).unwrap(), "the live row did not survive");
1677        assert!(
1678            !t.has(&oid(2)).unwrap(),
1679            "a reopen resurrected a retired row — the tombstone is not durable"
1680        );
1681        assert!(t.content(&oid(2)).unwrap().is_none());
1682        assert_eq!(t.of_kind(GitObjectKind::Blob).unwrap().len(), 1);
1683    }
1684
1685    /// sha256 keys are not truncated to sha1 width, and a table refuses a width
1686    /// it cannot hold.
1687    #[test]
1688    fn a_sha256_oid_keeps_all_thirty_two_bytes() {
1689        let p = tmp("sha256");
1690        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1691        let o = [3u8; 32];
1692        t.explode(&o, GitObjectKind::Tree, b"t").unwrap();
1693        t.flush().unwrap();
1694        assert!(t.has(&o).unwrap());
1695        assert_eq!(t.content(&o).unwrap().unwrap().1, b"t");
1696
1697        t.explode(&oid(1), GitObjectKind::Blob, b"narrow").unwrap();
1698        let err = t.flush().expect_err("a 20-byte oid cannot join a 32-byte table");
1699        assert!(format!("{err}").contains("32-byte oids"), "{err}");
1700    }
1701
1702    /// A torn tail is truncated away rather than left mid-stream, where every
1703    /// later batch would be unreachable.
1704    #[test]
1705    fn a_torn_tail_is_dropped_and_the_table_keeps_working() {
1706        let p = tmp("torn");
1707        {
1708            let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1709            t.explode(&oid(1), GitObjectKind::Blob, b"complete").unwrap();
1710            t.flush().unwrap();
1711        }
1712        // Half a message, as a killed process leaves.
1713        {
1714            let mut f = OpenOptions::new().append(true).open(&p).unwrap();
1715            f.write_all(&[0xFF, 0xFF, 0xFF, 0xFF, 0x40, 0x00, 0x00, 0x00, 1, 2, 3])
1716                .unwrap();
1717        }
1718        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1719        assert_eq!(t.rows().unwrap(), 1, "the torn message contributes nothing");
1720        t.explode(&oid(2), GitObjectKind::Blob, b"after").unwrap();
1721        t.flush().unwrap();
1722        assert_eq!(t.rows().unwrap(), 2);
1723        assert_eq!(
1724            t.content(&oid(2)).unwrap().unwrap().1,
1725            b"after".to_vec(),
1726            "the append landed at the last good boundary, not after the garbage"
1727        );
1728    }
1729
1730    #[test]
1731    fn graph_keeps_commits_and_trees_and_drops_blobs() {
1732        let p = ExplodePolicy::Graph;
1733        assert!(p.wants(GitObjectKind::Commit));
1734        assert!(p.wants(GitObjectKind::Tree));
1735        assert!(!p.wants(GitObjectKind::Blob));
1736        assert!(!ExplodePolicy::Off.wants(GitObjectKind::Commit));
1737        assert!(ExplodePolicy::Full.wants(GitObjectKind::Blob));
1738        assert_eq!(
1739            ExplodePolicy::default(),
1740            ExplodePolicy::Full,
1741            "the default keeps §14's eager table whole, so `adopt_journal`'s \
1742             row-count check keeps meaning what it did"
1743        );
1744    }
1745
1746    /// ★ THE SETTING: a declined object is not stored, and is COUNTED — so a
1747    /// table short by policy and one short by fault do not look alike.
1748    #[test]
1749    fn a_declined_object_is_not_stored_and_is_counted() {
1750        let t = ExplodedArchive::open_with_policy(&tmp("policy"), ExplodePolicy::Graph).unwrap();
1751        t.explode(&oid(1), GitObjectKind::Commit, b"c").unwrap();
1752        t.explode(&oid(2), GitObjectKind::Blob, b"bbbb").unwrap();
1753        t.flush().unwrap();
1754        assert!(t.has(&oid(1)).unwrap());
1755        assert!(!t.has(&oid(2)).unwrap());
1756        assert_eq!(t.skipped(), 1);
1757        assert_eq!(t.rows().unwrap(), 1);
1758    }
1759
1760    /// `Off` writes no file at all — the whole point of the free tier.
1761    #[test]
1762    fn off_stores_nothing_and_leaves_no_file() {
1763        let p = tmp("off");
1764        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Off).unwrap();
1765        t.explode(&oid(1), GitObjectKind::Commit, b"c").unwrap();
1766        t.explode(&oid(2), GitObjectKind::Blob, b"b").unwrap();
1767        assert_eq!(t.flush().unwrap(), 0);
1768        assert_eq!(t.rows().unwrap(), 0);
1769        assert_eq!(t.skipped(), 2);
1770        assert!(!p.exists(), "a disabled table must not create a file");
1771    }
1772
1773    // ════════════════════════════════════════════════════════════════════════
1774    // The stree in front of the lookup
1775    // ════════════════════════════════════════════════════════════════════════
1776
1777    /// `n` oids in `groups` of `per` that share their first EIGHT bytes and
1778    /// differ only after them — the collision is CONSTRUCTED, never hoped for.
1779    ///
1780    /// The prefixes deliberately straddle `0x80`: half the oids in a real repo
1781    /// start there, and they are the half a non-order-preserving key would sort
1782    /// onto the wrong side of the keyspace.
1783    fn colliding_oids(groups: usize, per: usize) -> Vec<[u8; 20]> {
1784        let mut out = Vec::with_capacity(groups * per);
1785        for g in 0..groups {
1786            let mut z = (g as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15);
1787            z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
1788            z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
1789            let prefix = (z ^ (z >> 31)).to_be_bytes();
1790            for k in 0..per {
1791                let mut o = [0u8; 20];
1792                o[..8].copy_from_slice(&prefix);
1793                o[8] = k as u8;
1794                o[9..13].copy_from_slice(&(g as u32).to_be_bytes());
1795                out.push(o);
1796            }
1797        }
1798        out
1799    }
1800
1801    /// ★ THE STREE RETURNS THE ROW THE BINARY SEARCH RETURNED — including on
1802    /// 8-byte prefix collisions, of which this table has 1000.
1803    ///
1804    /// Three answers are compared for every oid: the `stree` probe, the
1805    /// `binary_search` over the identical oid array that this file shipped with,
1806    /// and the **applied output** — the payload `content` hands back, which is
1807    /// unique per oid, so a wrong row is a wrong payload and not a passing
1808    /// assertion about a `Located` nobody read.
1809    ///
1810    /// Seen RED by deleting the full-oid comparison from `OidTree::verify` (i.e.
1811    /// `.find(|&i| self.oid_at(i) == oid)` → `.next()`, the "stree already found
1812    /// it" mistake): **"the stree and the binary search disagree at
1813    /// 0000000000000000010000000000000000000000: Some(Located { batch: 0, row: 0,
1814    /// kind: Blob }) vs Some(Located { batch: 0, row: 1, kind: Blob })"** — the
1815    /// probe was handed the row of the prefix-mate stored one slot earlier.
1816    #[test]
1817    fn the_stree_agrees_with_the_binary_search_it_replaced_on_colliding_prefixes() {
1818        let t = ExplodedArchive::open_with_policy(&tmp("stree-agree"), ExplodePolicy::Full).unwrap();
1819        let oids = colliding_oids(1000, 4);
1820        for (i, o) in oids.iter().enumerate() {
1821            t.explode(o, GitObjectKind::Blob, format!("payload-{i}").as_bytes())
1822                .unwrap();
1823        }
1824        t.flush().unwrap();
1825        // Force the index, then reach into it: the two finders must sit over the
1826        // SAME arrays, or this compares two indexes rather than two finders.
1827        t.index_now().unwrap();
1828        let idx = t.index.lock().unwrap();
1829        let tree = idx.as_ref().expect("index_now built it");
1830        assert_eq!(tree.len(), oids.len());
1831
1832        // The premise: the collisions are real in the built tree.
1833        let mut colliding_runs = 0usize;
1834        for o in &oids {
1835            let run = tree.candidate_run(key_for_oid(o));
1836            assert_eq!(
1837                run.len(),
1838                4,
1839                "expected a 4-entry candidate run for {}, got {run:?} — the collision \
1840                 the rest of this test rests on is not there",
1841                hex::encode(o)
1842            );
1843            colliding_runs += 1;
1844        }
1845        assert_eq!(colliding_runs, oids.len());
1846        // And the sign boundary is really crossed, or the key's top-bit flip is
1847        // untested by this corpus.
1848        assert!(
1849            oids.iter().any(|o| o[0] >= 0x80) && oids.iter().any(|o| o[0] < 0x80),
1850            "premise: the prefixes must straddle 0x80"
1851        );
1852
1853        for o in &oids {
1854            assert_eq!(
1855                tree.find(o),
1856                tree.find_by_binary_search(o),
1857                "the stree and the binary search disagree at {}: {:?} vs {:?}",
1858                hex::encode(o),
1859                tree.find(o),
1860                tree.find_by_binary_search(o)
1861            );
1862            assert!(tree.find(o).is_some(), "{} vanished", hex::encode(o));
1863        }
1864        // An oid on a colliding prefix that was never stored must MISS, in both
1865        // finders — this is the query a verify-less probe answers with somebody
1866        // else's row.
1867        for o in oids.iter().step_by(4) {
1868            let mut absent = *o;
1869            absent[8] = 0xff;
1870            assert_eq!(tree.find(&absent), None, "{} was never stored", hex::encode(absent));
1871            assert_eq!(tree.find_by_binary_search(&absent), None);
1872        }
1873        drop(idx);
1874
1875        // Applied output: each oid's own payload, not its prefix-mate's.
1876        for (i, o) in oids.iter().enumerate() {
1877            assert_eq!(
1878                t.content(o).unwrap().unwrap().1,
1879                format!("payload-{i}").into_bytes(),
1880                "{} came back with another object's bytes",
1881                hex::encode(o)
1882            );
1883        }
1884    }
1885
1886    /// ★ A TOMBSTONE STILL RETIRES AN OID WITH THE STREE IN FRONT — and its
1887    /// prefix-mates do not go with it, nor stand in for it.
1888    ///
1889    /// Two failure modes live here and both return a wrong answer rather than an
1890    /// error: a fold that keeps the *first* sighting rather than the last leaves
1891    /// the retired oid resolvable, and a probe that trusts the 8-byte key answers
1892    /// the retired oid with the row of the live oid it collides with.
1893    ///
1894    /// Seen RED three times:
1895    ///
1896    /// * `Keep::Last` → `Keep::First` in `index_now`: **"a reopen resurrected a
1897    ///   retired oid: 0000000000000000000000000000000000000000"**. The same break
1898    ///   also reds the older, coarser
1899    ///   `retain_drops_only_what_is_dead_and_the_drop_survives_a_reopen`.
1900    /// * the full-oid comparison dropped from `OidTree::verify`:
1901    ///   **"0000000000000000000000000000000000000000 was retired"** — the dead
1902    ///   oid answered `has` with its live prefix-mate's row, in-process, before
1903    ///   any reopen.
1904    /// * the incremental fold in `flush` no longer filtering the retired set
1905    ///   (`if !retired.contains(oid)` → `if true`): the same message, from the
1906    ///   live process rather than from a reopen.
1907    #[test]
1908    fn a_tombstoned_oid_stays_absent_when_a_live_oid_shares_its_prefix() {
1909        let p = tmp("stree-tomb");
1910        // Four oids per prefix: one is retired, three stay.
1911        let oids = colliding_oids(64, 4);
1912        let dead: Vec<[u8; 20]> = oids.iter().copied().step_by(4).collect();
1913        {
1914            let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1915            for (i, o) in oids.iter().enumerate() {
1916                t.explode(o, GitObjectKind::Blob, format!("live-{i}").as_bytes())
1917                    .unwrap();
1918            }
1919            t.flush().unwrap();
1920            let retired = t.retain(&|o: &[u8]| !dead.iter().any(|d| d == o)).unwrap();
1921            assert_eq!(retired, dead.len() as u64);
1922            for d in &dead {
1923                assert!(!t.has(d).unwrap(), "{} was retired", hex::encode(d));
1924                assert!(
1925                    t.content(d).unwrap().is_none(),
1926                    "the retired oid {} was answered with a prefix-mate's row",
1927                    hex::encode(d)
1928                );
1929            }
1930            for (i, o) in oids.iter().enumerate() {
1931                if i % 4 == 0 {
1932                    continue;
1933                }
1934                assert_eq!(
1935                    t.content(o).unwrap().unwrap().1,
1936                    format!("live-{i}").into_bytes(),
1937                    "{} is alive and must still answer with its OWN bytes",
1938                    hex::encode(o)
1939                );
1940            }
1941        }
1942        // ★ And a reopen — which rebuilds the tree from the file, tombstones and
1943        // all — finds every live row and none of the dead.
1944        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1945        for (i, o) in oids.iter().enumerate() {
1946            if i % 4 == 0 {
1947                assert!(
1948                    !t.has(o).unwrap(),
1949                    "a reopen resurrected a retired oid: {}",
1950                    hex::encode(o)
1951                );
1952            } else {
1953                assert_eq!(
1954                    t.content(o).unwrap().unwrap().1,
1955                    format!("live-{i}").into_bytes(),
1956                    "a reopen lost or misrouted the live oid {}",
1957                    hex::encode(o)
1958                );
1959            }
1960        }
1961        assert_eq!(
1962            t.of_kind(GitObjectKind::Blob).unwrap().len(),
1963            oids.len() - dead.len()
1964        );
1965        // Only now — `rows()` answers from batch metadata until an index exists,
1966        // and the file still holds the retired rows and their tombstones. This
1967        // assertion measured 320 against 192 when it was made before the reads
1968        // above, which is the documented behaviour of `rows()` and not the
1969        // tombstone leaking.
1970        assert_eq!(t.rows().unwrap(), (oids.len() - dead.len()) as u64);
1971    }
1972
1973    /// The incremental extension in `flush` and the from-scratch rebuild in
1974    /// `index_now` must land on the same tree — one of them runs while a push is
1975    /// hot and the other after a restart, and a difference between them is a
1976    /// bug that only appears to the second process.
1977    ///
1978    /// Seen RED twice, once from each side, and the two reds are mirror images —
1979    /// which is what tells you the test compares the folds rather than one of
1980    /// them against itself:
1981    ///
1982    /// * incremental path drops the `retired` filter over the entries it carries
1983    ///   forward (`if !retired.contains(oid)` → `if true`): **"the two folds kept
1984    ///   a different number of oids: left: 150, right: 200"** — the live index
1985    ///   kept the 50 it had been told to retire.
1986    /// * `Keep::Last` → `Keep::First` in `index_now`: **"left: 200, right: 150"**
1987    ///   — the rebuild kept them instead.
1988    #[test]
1989    fn the_incremental_index_and_a_rebuilt_one_agree() {
1990        let p = tmp("stree-incr");
1991        let oids = colliding_oids(50, 4);
1992        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
1993        // Build the index FIRST, so every flush below takes the incremental path.
1994        t.index_now().unwrap();
1995        for (i, o) in oids.iter().enumerate() {
1996            t.explode(o, GitObjectKind::Blob, format!("v-{i}").as_bytes())
1997                .unwrap();
1998            if i % 37 == 0 {
1999                t.flush().unwrap();
2000            }
2001        }
2002        t.flush().unwrap();
2003        t.retain(&|o: &[u8]| o[8] != 3).unwrap();
2004        let live: Vec<(Vec<u8>, Vec<u8>)> = {
2005            let idx = t.index.lock().unwrap();
2006            idx.as_ref()
2007                .unwrap()
2008                .iter()
2009                .map(|(o, l)| (o.to_vec(), vec![l.batch as u8, l.row as u8]))
2010                .collect()
2011        };
2012        drop(t);
2013
2014        let again = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2015        again.index_now().unwrap();
2016        let rebuilt = again.index.lock().unwrap();
2017        let rebuilt = rebuilt.as_ref().unwrap();
2018        assert_eq!(rebuilt.len(), live.len(), "the two folds kept a different number of oids");
2019        for (o, _) in &live {
2020            assert!(
2021                rebuilt.find(o).is_some(),
2022                "the live process and a reopen disagree about {}: Some(..) vs None",
2023                hex::encode(o)
2024            );
2025        }
2026        for o in oids.iter().filter(|o| o[8] == 3) {
2027            assert!(rebuilt.find(o).is_none(), "{} was retired", hex::encode(o));
2028        }
2029    }
2030
2031    /// The measurement behind the change, re-takeable: three finders over the
2032    /// SAME 1 000 000 entries — the `Vec<(Vec<u8>, Located)>` this file shipped
2033    /// with, a binary search over the flat array, and the stree.
2034    ///
2035    /// `--ignored` because it is a measurement, not a guard: it asserts only that
2036    /// the three agree, and prints the times. Run it with
2037    /// `cargo test --release -p znippy-plugin-git --lib exploded_arrow -- --ignored --nocapture`
2038    /// — a debug build measures the borrow checker, not the machine.
2039    ///
2040    /// ## Measured, oden 2026-08-14, `--release`, 1 000 000 entries, 200 000
2041    /// probes (half misses), three runs
2042    ///
2043    /// | arm | ns/probe | vs shipped |
2044    /// |---|---:|---:|
2045    /// | `Vec<(Vec<u8>, Located)>` + `binary_search` (what shipped) | 836 / 862 / 961 | 1.00 |
2046    /// | flat oid array + `binary_search` | 339 / 339 / 348 | 0.40 |
2047    /// | **stree** | **164 / 175 / 185** | **0.21** |
2048    ///
2049    /// So the structure is worth **~2×** on its own and the flattening another
2050    /// **~2.5×**, and the two together take a point lookup to a fifth of what it
2051    /// cost. **The box was NOT quiet** — `/proc/loadavg` 1-min 29.7–31.5 on 32
2052    /// cores, other agents building throughout — so the absolute figures are
2053    /// upper bounds. The ratios are reported instead, and each arm's own spread
2054    /// across the three runs (14% / 2.7% / 12%) is far narrower than the 5×
2055    /// and 2× between them.
2056    #[test]
2057    #[ignore = "measurement; run under --release with --ignored"]
2058    fn lookup_cost_before_and_after() {
2059        use std::time::Instant;
2060        let n = 1_000_000usize;
2061        // 4 oids per prefix, so ~25% of probes land in a run of four and the
2062        // verify step is exercised rather than skipped.
2063        let oids = colliding_oids(n / 4, 4);
2064        let mut b = OidTreeBuilder::with_capacity(oids.len());
2065        for (i, o) in oids.iter().enumerate() {
2066            b.push(o, Some(Located { batch: (i / 100_000) as u32, row: i as u32, kind: GitObjectKind::Blob, pay_at: EXTENT_UNAVAILABLE, pay_len: 0 }))
2067                .unwrap();
2068        }
2069        let build = Instant::now();
2070        let tree = b.finish(Keep::Last);
2071        let build = build.elapsed();
2072
2073        // The shape this replaced: one heap allocation per oid.
2074        let mut old: Vec<(Vec<u8>, Located)> =
2075            tree.iter().map(|(o, l)| (o.to_vec(), l)).collect();
2076        old.sort_by(|a, b| a.0.cmp(&b.0));
2077
2078        // A miss-heavy probe order, and NOT in oid order — a `have` negotiation
2079        // is random and half of it misses.
2080        let mut probes: Vec<[u8; 20]> = Vec::with_capacity(200_000);
2081        for i in (0..oids.len()).step_by(oids.len() / 100_000) {
2082            probes.push(oids[i]);
2083            let mut m = oids[i];
2084            m[8] = 0xfe;
2085            probes.push(m);
2086        }
2087
2088        let mut sink = 0u64;
2089        let t0 = Instant::now();
2090        for p in &probes {
2091            if let Ok(i) = old.binary_search_by(|(o, _)| o.as_slice().cmp(&p[..])) {
2092                sink += old[i].1.row as u64;
2093            }
2094        }
2095        let vec_of_vec = t0.elapsed();
2096        let t0 = Instant::now();
2097        for p in &probes {
2098            if let Some(l) = tree.find_by_binary_search(p) {
2099                sink += l.row as u64;
2100            }
2101        }
2102        let flat_bsearch = t0.elapsed();
2103        let t0 = Instant::now();
2104        for p in &probes {
2105            if let Some(l) = tree.find(p) {
2106                sink += l.row as u64;
2107            }
2108        }
2109        let stree = t0.elapsed();
2110
2111        // The arms must be the same question.
2112        for p in &probes {
2113            let want = tree.find_by_binary_search(p);
2114            assert_eq!(tree.find(p), want, "arms disagree at {}", hex::encode(p));
2115        }
2116        let per = |d: std::time::Duration| d.as_secs_f64() * 1e9 / probes.len() as f64;
2117        println!(
2118            "exploded oid lookup, {} entries ({} groups of 4 colliding prefixes), \
2119             {} probes half of them misses:\n  \
2120             Vec<(Vec<u8>,Located)> binary_search  {:>7.1} ns/probe\n  \
2121             flat array binary_search             {:>7.1} ns/probe\n  \
2122             stree                                {:>7.1} ns/probe\n  \
2123             tree build (sort + fold + stree)      {:.3} s   sink={sink}",
2124            tree.len(),
2125            oids.len() / 4,
2126            probes.len(),
2127            per(vec_of_vec),
2128            per(flat_bsearch),
2129            per(stree),
2130            build.as_secs_f64(),
2131        );
2132    }
2133
2134    #[test]
2135    fn an_unknown_policy_is_refused() {
2136        assert_eq!(ExplodePolicy::parse("full").unwrap(), ExplodePolicy::Full);
2137        let err = ExplodePolicy::parse("ful").expect_err("a typo must be refused");
2138        assert!(format!("{err}").contains("unknown explode policy"), "{err}");
2139    }
2140}