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.
133/// **How each row's payload is stored** — `ENC_RAW` or `ENC_ZLIB`. Lever 3
134/// (2026-09-04): the eager pool was 1.67 GB of inflated commits and trees, 61 %
135/// of everything the store kept above the pack. Arrow IPC's own body
136/// compression is per BUFFER, so a point read could not lift one row out of a
137/// compressed batch and the `pread` extent fast path — the whole reason the
138/// pool is fast — would have died with it (`payload_data_offset` refuses
139/// compressed batches for exactly that reason; MEASURED 200 000× read
140/// amplification without the extent). So the compression is per ROW: the
141/// extent pread returns one compressed payload and inflates just that one.
142/// zlib, because the crate already has it (`flate2`, zlib-rs — pure Rust), and
143/// because it is the codec git itself uses per object, so the ratios are known
144/// ground.
145pub const COL_ENC: &str = "enc";
146pub const ENC_RAW: u8 = 0;
147pub const ENC_ZLIB: u8 = 1;
148/// Rows shorter than this are never compressed: a zlib header and trailer are
149/// 8 bytes, and a tiny tree does not earn them back. Set from
150/// `examples/explode_compress_ratio.rs` over the kernel mirror's pool.
151pub const MIN_ROW_COMPRESS: usize = 128;
152/// A compressed row must be at least this much smaller than the raw one to be
153/// kept compressed, or the inflate on every read buys nothing.
154const COMPRESS_KEEP_IF_UNDER_PERCENT: usize = 90;
155
156/// Schema v2 (2026-09-04): `enc` sits BEFORE `payload`, so the payload's data
157/// buffer stays the LAST buffer of the batch — which is what
158/// [`payload_data_offset`] reads the extent off. A v1 file has no `enc` column;
159/// every reader treats a missing column as [`ENC_RAW`], so pools written before
160/// this open unchanged and are never rewritten for it.
161pub fn exploded_schema(oid_len: usize) -> SchemaRef {
162    Arc::new(Schema::new(vec![
163        Field::new(COL_OID, DataType::FixedSizeBinary(oid_len as i32), false),
164        Field::new(COL_TYPE, DataType::UInt8, false),
165        Field::new(COL_MODE, DataType::UInt32, true),
166        Field::new(COL_PATH, DataType::Utf8, true),
167        Field::new(COL_ENC, DataType::UInt8, false),
168        Field::new(COL_PAYLOAD, DataType::LargeBinary, false),
169    ]))
170}
171
172/// Schema v1 — what every pool written before 2026-09-04 carries, and what this
173/// build keeps appending to such a pool (see `ExplodedArchive::legacy_v1`).
174pub fn exploded_schema_v1(oid_len: usize) -> SchemaRef {
175    Arc::new(Schema::new(vec![
176        Field::new(COL_OID, DataType::FixedSizeBinary(oid_len as i32), false),
177        Field::new(COL_TYPE, DataType::UInt8, false),
178        Field::new(COL_MODE, DataType::UInt32, true),
179        Field::new(COL_PATH, DataType::Utf8, true),
180        Field::new(COL_PAYLOAD, DataType::LargeBinary, false),
181    ]))
182}
183
184/// Compress one row's payload for storage, or `None` when it is not worth it
185/// (too short, or zlib did not get it under the keep threshold).
186///
187/// **`default` (level 6), not `fast`, and that is a measurement**
188/// (`examples/explode_compress_ratio.rs`, the kernel mirror's eager pool,
189/// 1 722 249 rows / 1 609 688 036 B of commits and tags, 2026-09-04):
190///
191/// | level | pool after | ratio | saved |
192/// |---|---:|---:|---:|
193/// | fast | 1 138 092 314 B | 1.41× | 471 595 722 B |
194/// | **default** | **889 639 690 B** | **1.81×** | **720 048 346 B** |
195///
196/// Inflate cost is the SAME at both — 3.07 µs/row mean over 20 000 sampled
197/// rows — because zlib decode speed does not depend on the level. So the read
198/// path pays +3 µs on a point read (first-read p50 was 3.9 µs) whichever level
199/// is chosen, and the only thing the level decides is how much disk that buys:
200/// 44.7 % of the pool at `default` against 29.3 % at `fast`. The write pays
201/// more CPU at `default`, on the indexer's threads, once per object.
202fn compress_row(raw: &[u8]) -> Option<Vec<u8>> {
203    use std::io::Write as _;
204    if raw.len() < MIN_ROW_COMPRESS {
205        return None;
206    }
207    let mut e = flate2::write::ZlibEncoder::new(
208        Vec::with_capacity(raw.len() / 2 + 16),
209        flate2::Compression::default(),
210    );
211    e.write_all(raw).ok()?;
212    let z = e.finish().ok()?;
213    (z.len() * 100 < raw.len() * COMPRESS_KEEP_IF_UNDER_PERCENT).then_some(z)
214}
215
216/// The row's payload as the caller wants it — inflated if it was stored
217/// compressed. `bytes` is exactly what the column (or the extent pread) held.
218fn row_payload(enc: u8, bytes: &[u8]) -> Result<Vec<u8>> {
219    use std::io::Read as _;
220    match enc {
221        ENC_RAW => Ok(bytes.to_vec()),
222        ENC_ZLIB => {
223            let mut out = Vec::with_capacity(bytes.len() * 3);
224            flate2::read::ZlibDecoder::new(bytes)
225                .read_to_end(&mut out)
226                .context("inflating a compressed exploded row")?;
227            Ok(out)
228        }
229        other => anyhow::bail!("exploded row carries unknown encoding {other}"),
230    }
231}
232
233/// The `enc` column of a decoded batch, or `None` on a v1 batch (every row raw).
234fn enc_column(batch: &RecordBatch) -> Option<&UInt8Array> {
235    batch
236        .column_by_name(COL_ENC)
237        .and_then(|c| c.as_any().downcast_ref::<UInt8Array>())
238}
239
240#[inline]
241fn enc_at(enc: Option<&UInt8Array>, row: usize) -> u8 {
242    enc.map_or(ENC_RAW, |e| e.value(row))
243}
244
245/// The pack type codes, the same mapping [`crate::exploded`] uses, so the two
246/// media cannot disagree about what a `2` means.
247pub fn kind_code(k: GitObjectKind) -> u8 {
248    match k {
249        GitObjectKind::Commit => 1,
250        GitObjectKind::Tree => 2,
251        GitObjectKind::Blob => 3,
252        GitObjectKind::Tag => 4,
253    }
254}
255
256pub fn kind_of(code: u8) -> Option<GitObjectKind> {
257    match code {
258        1 => Some(GitObjectKind::Commit),
259        2 => Some(GitObjectKind::Tree),
260        3 => Some(GitObjectKind::Blob),
261        4 => Some(GitObjectKind::Tag),
262        _ => None,
263    }
264}
265
266/// **A row that retires an oid.**
267///
268/// The file is append-only, so `retain` cannot reach back and delete: it appends
269/// rows that say *this oid is gone*, and the index fold applies them in file
270/// order, so the last word about an oid wins. Dropping only from the in-memory
271/// index is what a first cut did, and it was WRONG in the way that matters —
272/// `a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened` came back
273/// with **551 commits instead of 1**, because a reopen rebuilds the index from
274/// the file and resurrected every row `gc` had retired.
275///
276/// `0` and not a sixth column: git's pack type codes are 1–7 and 0 is not one of
277/// them, so [`kind_of`] already answers `None` for it. A tombstone costs an oid
278/// and an empty payload, not a byte on every live row.
279pub const TOMBSTONE: u8 = 0;
280
281/// One buffered row, waiting for the next batch. `code` rather than a
282/// [`GitObjectKind`] because a tombstone is a row with no kind.
283#[derive(Debug, Clone)]
284struct Pending {
285    oid: Vec<u8>,
286    code: u8,
287    mode: Option<u32>,
288    path: Option<String>,
289    /// Where this row's bytes are in [`PendingBatch::arena`]. **Not a
290    /// `Vec<u8>`** — see [`PendingBatch`].
291    pay: std::ops::Range<usize>,
292}
293
294/// The rows buffered for the next batch, and **one arena holding their bytes**.
295///
296/// # Why the bytes are not in the row
297///
298/// `payload: Vec<u8>` cost one heap allocation and one `memcpy` of the whole
299/// object **per buffered row**, and on the lazy write-back path that allocation
300/// is a second copy of bytes the reader is holding anyway. MEASURED by lane AA,
301/// 2026-08-30: reading one 3 145 728 B object through `gunnar-cli` allocated
302/// 6 340 693 B — *exactly two copies* — and the second was this one.
303///
304/// An arena makes the write an `extend_from_slice` into a buffer that grows
305/// geometrically and is reused after every flush, so the bytes attributable to
306/// one admitted read amortise to zero and a **declined** read costs nothing at
307/// all (the admission gate returns before this is reached). LAW: zero copy.
308#[derive(Default)]
309struct PendingBatch {
310    rows: Vec<Pending>,
311    arena: Vec<u8>,
312}
313
314impl PendingBatch {
315    fn is_empty(&self) -> bool {
316        self.rows.is_empty()
317    }
318    /// Buffer one row's bytes and return the range they landed in.
319    fn stash(&mut self, payload: &[u8]) -> std::ops::Range<usize> {
320        let start = self.arena.len();
321        self.arena.extend_from_slice(payload);
322        start..self.arena.len()
323    }
324    fn bytes(&self, r: &Pending) -> &[u8] {
325        &self.arena[r.pay.clone()]
326    }
327    /// Hand the batch over for encoding and keep the arena's capacity: the next
328    /// batch reuses the same allocation, which is what makes the per-read cost
329    /// amortise.
330    fn take(&mut self) -> (Vec<Pending>, Vec<u8>) {
331        let rows = std::mem::take(&mut self.rows);
332        let arena = std::mem::replace(&mut self.arena, Vec::with_capacity(0));
333        (rows, arena)
334    }
335}
336
337/// Where one batch message sits in the file. Recovered from the framing walk,
338/// which reads metadata and seeks over bodies.
339#[derive(Debug, Clone, Copy, PartialEq, Eq)]
340struct Batch {
341    /// Byte offset of the message's continuation marker.
342    at: u64,
343    /// Bytes the whole message occupies: `8 + metadata + body`.
344    len: u64,
345    /// Rows in it — read from the metadata, so counting them costs no body read.
346    rows: u32,
347    /// Absolute byte offset of the message BODY — where the buffer offsets the
348    /// metadata declares are measured from.
349    body: u64,
350    /// The payload DATA buffer's offset within the body, from the metadata's
351    /// own buffer table. [`EXTENT_UNAVAILABLE`] when the body is compressed or
352    /// the buffer layout is not the one [`exploded_schema`] writes — a point
353    /// read then falls back to decoding the batch, which is slow and correct.
354    pay_data: u64,
355}
356
357/// Sentinel for "no byte extent could be computed for this row/batch" — the
358/// fall-back-and-decode marker, never an offset a file could really have.
359const EXTENT_UNAVAILABLE: u64 = u64::MAX;
360
361/// The payload DATA buffer's offset within the message body, read from the
362/// batch metadata alone.
363///
364/// The payload is [`exploded_schema`]'s LAST field and a `LargeBinary`'s data
365/// buffer is its last buffer, so the table's final buffer entry IS the payload
366/// bytes — asserted by also counting the buffers (12: validity+data for oid and
367/// object_type, validity+data for mode, validity+offsets+data for path and for
368/// payload). `None` when the count disagrees or the body is compressed;
369/// [`ExplodedArchive::content`] then decodes the batch instead of guessing.
370fn payload_data_offset(meta: &[u8]) -> Option<u64> {
371    let msg = root_as_message(meta).ok()?;
372    let rb = msg.header_as_record_batch()?;
373    if rb.compression().is_some() {
374        return None;
375    }
376    let bufs = rb.buffers()?;
377    // 12 buffers is schema v1 (oid 2, type 2, mode 2, path 3, payload 3); 14 is
378    // v2 with the `enc` column's two. In both the payload's data buffer is the
379    // last one, which is the only fact used below.
380    if bufs.len() != 12 && bufs.len() != 14 {
381        return None;
382    }
383    let last = bufs.get(bufs.len() - 1);
384    (last.offset() >= 0).then(|| last.offset() as u64)
385}
386
387/// Where one object's row is — and where its payload BYTES are.
388///
389/// The extent is what makes a point read one small `pread` instead of a whole
390/// 64 MiB batch decode. MEASURED without it, rust-lang/rust push, 2026-08-19:
391/// the pre-ack connectivity walk visited commits and trees in graph order —
392/// effectively random across ~574 batches against a 4-slot cache — and decoded
393/// 6.7 GB/s of page cache to serve ~98 objects/s, one core pinned for hours, a
394/// ~200 000× read amplification. The extent costs 16 bytes per live oid.
395#[derive(Debug, Clone, Copy, PartialEq, Eq)]
396struct Located {
397    batch: u32,
398    row: u32,
399    kind: GitObjectKind,
400    /// Absolute file offset of this row's payload bytes, or
401    /// [`EXTENT_UNAVAILABLE`] — then the read falls back to the batch decode.
402    pay_at: u64,
403    /// Payload length in bytes. Meaningless when `pay_at` is the sentinel.
404    pay_len: u64,
405    /// [`ENC_RAW`] or [`ENC_ZLIB`] — carried in the index so the extent fast
406    /// path knows whether to inflate without decoding the batch.
407    enc: u8,
408}
409
410/// **`oid → Located`, as ragnar's static tree over flat arrays.**
411///
412/// Three parallel arrays in one oid-ascending order, plus an [`STree64Mmap`]
413/// built over the first of them:
414///
415/// ```text
416///   keys  count * 8         i64 little-endian, ascending — the stree keyspace
417///   oids  count * oid_len   the FULL oid, for the verify step
418///   locs  count             batch, row, kind
419/// ```
420///
421/// ## Why this and not the `Vec<(Vec<u8>, Located)>` it replaces
422///
423/// That vector cost one heap allocation *per oid* — 11.7 M of them for
424/// `linux.git` — and its `binary_search_by` chased `log₂ n` dependent loads over
425/// `Vec` headers scattered across the heap before it could compare a byte. Here
426/// the oid bytes are contiguous, and `stree` routes through 64-byte nodes with a
427/// branchless AVX2 compare. `nornir-workspace.toml` names this structure as the
428/// one to use, and [`crate::oid_index`] already uses it over exactly these keys —
429/// so the key function is **imported from there**, not written again.
430///
431/// ## The 8-byte key is a FILTER, not a key — and this is where a wrong answer
432/// would come from
433///
434/// The key is [`key_for_oid`]: the first eight bytes of the oid, big-endian, top
435/// bit flipped so unsigned byte order becomes signed `i64` order (without the
436/// flip every oid starting `0x80..0xff` — half of them — sorts below every oid
437/// starting `0x00..0x7f`, and the keyspace `stree` requires to be ascending is
438/// not). **Twelve** of a sha1 oid's twenty bytes are not in the key at all (24 of
439/// 32 for sha256), so two distinct objects **can** share one key.
440///
441/// How likely, honestly: for `linux.git`'s 11.7 M objects, birthday arithmetic
442/// gives n²/2 · 2⁻⁶⁴ ≈ **3.7 × 10⁻⁶** — one table in ~270 000. So it will not
443/// happen by accident on one repository, and that is exactly why a verify-less
444/// probe would ship and pass every test anyone bothered to write. It does not
445/// stay accidental: finding *some* pair of contents whose oids share eight bytes
446/// is a 2³² birthday search — hours of commodity hashing, not a research result —
447/// and both objects are then perfectly ordinary blobs a client may push. (Hitting
448/// a *specific* stored oid's prefix is 2⁶⁴ and is not a threat.) A hosted forge
449/// serves oids that other people chose, so this is a property to hold by
450/// construction rather than a probability to accept.
451///
452/// So `stree` narrows and never decides. It routes to *a* member of the run of
453/// equal keys — which member is its own business, because its leaf scan starts at
454/// a block boundary that can fall in the middle of the run — so
455/// [`expand_run`](Self::expand_run) walks **both** directions to the run's ends,
456/// and [`find`](Self::find) then compares the **whole oid** against every
457/// candidate. Skipping that comparison would not merely be sloppy: a `have`
458/// negotiation for an oid this table has never seen, whose first eight bytes
459/// happen to match a stored one, would be answered with **the other object's
460/// bytes** — a silently wrong object, which is far worse than a slow one.
461struct OidTree {
462    /// Width of one oid. `0` only when the tree is empty.
463    oid_len: usize,
464    count: usize,
465    keys: Vec<u8>,
466    oids: Vec<u8>,
467    locs: Vec<Located>,
468    /// `None` for an empty table — `STree64Mmap::new_with_stride` asserts
469    /// `count > 0`.
470    tree: Option<STree64Mmap>,
471}
472
473/// Which member of a run of equal oids survives the fold — the two callers
474/// genuinely differ, and the difference is not cosmetic.
475#[derive(Debug, Clone, Copy, PartialEq, Eq)]
476enum Keep {
477    /// [`ExplodedArchive::flush`]'s incremental extension: an oid already in the
478    /// index keeps the [`Located`] it already had. That is what
479    /// [`ExplodedArchive::explode_at`] documents — a blob is reachable at many
480    /// paths and the table records the **first** name it was seen at.
481    First,
482    /// [`ExplodedArchive::index_now`]'s rebuild from the file: batches are walked
483    /// in file order and rows in row order, so the **last** word about an oid
484    /// wins. That is what makes a [`TOMBSTONE`] durable rather than advisory.
485    Last,
486}
487
488/// Accumulates sightings in the order they happen, then folds them.
489///
490/// Sightings, not entries: the same oid may arrive many times, and a `None`
491/// [`Located`] is a [`TOMBSTONE`] row saying the oid is gone. Nothing is
492/// deduplicated until [`finish`](Self::finish), because which sighting wins
493/// depends on their order and on [`Keep`].
494struct OidTreeBuilder {
495    /// Taken from the first oid pushed; every later one must match it.
496    oid_len: usize,
497    oids: Vec<u8>,
498    locs: Vec<Option<Located>>,
499}
500
501impl OidTreeBuilder {
502    fn with_capacity(rows: usize) -> Self {
503        Self { oid_len: 0, oids: Vec::new(), locs: Vec::with_capacity(rows) }
504    }
505
506    /// Record one sighting. `loc` is `None` for a tombstone row.
507    ///
508    /// A width that disagrees with the rows already pushed is refused rather
509    /// than padded: a truncated or extended oid would sort and compare as a
510    /// different object.
511    fn push(&mut self, oid: &[u8], loc: Option<Located>) -> Result<()> {
512        if self.locs.is_empty() && self.oids.is_empty() {
513            self.oid_len = oid.len();
514            self.oids.reserve(self.locs.capacity() * self.oid_len);
515        }
516        anyhow::ensure!(
517            oid.len() == self.oid_len,
518            "the exploded index holds {}-byte oids and a {}-byte one arrived",
519            self.oid_len,
520            oid.len()
521        );
522        self.oids.extend_from_slice(oid);
523        self.locs.push(loc);
524        Ok(())
525    }
526
527    fn oid_at(&self, i: usize) -> &[u8] {
528        &self.oids[i * self.oid_len..(i + 1) * self.oid_len]
529    }
530
531    /// Sort, fold each run of equal oids down to its winner, drop the ones whose
532    /// winner is a tombstone, and build the tree.
533    fn finish(self, keep: Keep) -> OidTree {
534        let n = self.locs.len();
535        let mut order: Vec<u32> = (0..n as u32).collect();
536        // STABLE, and load-bearing: `Keep` is about *insertion* order within a
537        // run of equal oids, so an unstable sort would make "first sighting"
538        // and "last word wins" both mean "whichever one the sort happened to
539        // leave there".
540        order.sort_by(|&a, &b| self.oid_at(a as usize).cmp(self.oid_at(b as usize)));
541
542        let mut keys: Vec<u8> = Vec::with_capacity(n * 8);
543        let mut oids: Vec<u8> = Vec::with_capacity(n * self.oid_len);
544        let mut locs: Vec<Located> = Vec::with_capacity(n);
545        let mut i = 0usize;
546        while i < n {
547            let oid = self.oid_at(order[i] as usize);
548            let mut j = i + 1;
549            while j < n && self.oid_at(order[j] as usize) == oid {
550                j += 1;
551            }
552            let pick = match keep {
553                Keep::First => order[i],
554                Keep::Last => order[j - 1],
555            } as usize;
556            if let Some(loc) = self.locs[pick] {
557                // Ascending oid order gives ascending keys, because
558                // `key_for_oid` is order-preserving over the oid's leading
559                // bytes — the property `stree` needs and the reason for the
560                // sign-bit flip.
561                keys.extend_from_slice(&key_for_oid(oid).to_le_bytes());
562                oids.extend_from_slice(oid);
563                locs.push(loc);
564            }
565            i = j;
566        }
567
568        let count = locs.len();
569        let tree = (count > 0).then(|| STree64Mmap::new_with_stride(&keys, count, 8));
570        OidTree { oid_len: self.oid_len, count, keys, oids, locs, tree }
571    }
572}
573
574impl OidTree {
575    fn len(&self) -> usize {
576        self.count
577    }
578
579    fn oid_at(&self, i: usize) -> &[u8] {
580        &self.oids[i * self.oid_len..(i + 1) * self.oid_len]
581    }
582
583    fn key_at(&self, i: usize) -> i64 {
584        i64::from_le_bytes(self.keys[i * 8..i * 8 + 8].try_into().unwrap())
585    }
586
587    /// Widen a hit to the whole run of equal keys. `stree` routes to *a* member
588    /// of the run — its leaf scan starts at a block boundary, which can fall
589    /// anywhere inside one — so both directions are walked rather than assumed.
590    fn expand_run(&self, pos: usize, key: i64) -> std::ops::Range<usize> {
591        let mut lo = pos;
592        while lo > 0 && self.key_at(lo - 1) == key {
593            lo -= 1;
594        }
595        let mut hi = pos + 1;
596        while hi < self.count && self.key_at(hi) == key {
597            hi += 1;
598        }
599        lo..hi
600    }
601
602    /// Resolve an oid. `None` when absent — never another object's row.
603    fn find(&self, oid: &[u8]) -> Option<Located> {
604        if oid.len() != self.oid_len {
605            return None;
606        }
607        let tree = self.tree.as_ref()?;
608        let key = key_for_oid(oid);
609        let pos = tree.find_exact(key, &self.keys)?;
610        self.verify(pos, key, oid)
611    }
612
613    /// The full-oid comparison the whole structure's correctness rests on.
614    fn verify(&self, pos: usize, key: i64, oid: &[u8]) -> Option<Located> {
615        self.expand_run(pos, key)
616            .find(|&i| self.oid_at(i) == oid)
617            .map(|i| self.locs[i])
618    }
619
620    /// Every live entry, in ascending oid order — what `of_kind` and `retain`
621    /// fold over.
622    fn iter(&self) -> impl Iterator<Item = (&[u8], Located)> + '_ {
623        (0..self.count).map(move |i| (self.oid_at(i), self.locs[i]))
624    }
625
626    /// The **baseline** the tree has to agree with: `binary_search` over the very
627    /// same oid array, followed by the very same run-and-verify. It is the
628    /// finder this file shipped with, and it exists so "the stree returns the
629    /// same row" is a measurement rather than an assertion.
630    #[cfg(test)]
631    fn find_by_binary_search(&self, oid: &[u8]) -> Option<Located> {
632        if oid.len() != self.oid_len || self.count == 0 {
633            return None;
634        }
635        let mut lo = 0usize;
636        let mut hi = self.count;
637        while lo < hi {
638            let mid = lo + (hi - lo) / 2;
639            if self.oid_at(mid) < oid { lo = mid + 1 } else { hi = mid }
640        }
641        (lo < self.count && self.oid_at(lo) == oid).then(|| self.locs[lo])
642    }
643
644    /// The **unverified** candidate run for a key: every entry sharing that
645    /// 8-byte prefix. Normally length 1; length > 1 is a real prefix collision.
646    /// Exposed so a test can prove the collision it constructed is real and that
647    /// the verify step is what tells the candidates apart.
648    #[cfg(test)]
649    fn candidate_run(&self, key: i64) -> std::ops::Range<usize> {
650        let Some(tree) = self.tree.as_ref() else { return 0..0 };
651        let Some(pos) = tree.find_exact(key, &self.keys) else { return 0..0 };
652        self.expand_run(pos, key)
653    }
654}
655
656/// Counters, all of them applied output.
657#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
658pub struct Stats {
659    pub rows: u64,
660    pub written: u64,
661    pub served: u64,
662    pub rederived: u64,
663    pub absent: u64,
664}
665
666/// §14's derived table for one repository.
667pub struct ExplodedArchive {
668    path: PathBuf,
669    /// The schema message: the file's first message, kept so a point read can
670    /// feed a [`StreamDecoder`] without re-reading it from disk each time.
671    schema_msg: Mutex<Option<Vec<u8>>>,
672    /// Every batch message in the file, in order. Cheap — one entry per 64 MiB.
673    batches: Mutex<Vec<Batch>>,
674    /// `oid → Located` as an [`OidTree`]. `None` until the first lookup forces
675    /// the pass that builds it, so a push-only process never pays for it.
676    index: Mutex<Option<OidTree>>,
677    /// The last few decoded batches, most recent first.
678    ///
679    /// Not an optimisation looking for a problem: without it every point read
680    /// decodes its whole batch, and the 2687-object push guard went from 4 s to
681    /// **over 100 s** — 2687 lookups × one 106 MiB batch each. A batch is
682    /// `Arc`-backed, so a hit is a clone of pointers.
683    cache: Mutex<Vec<(u32, RecordBatch)>>,
684    pending: Mutex<PendingBatch>,
685    pending_bytes: AtomicU64,
686    /// Bytes in the file. Kept here rather than stat'ed so the append offset and
687    /// the batch table cannot drift apart.
688    end: Mutex<u64>,
689    oid_len: Mutex<Option<usize>>,
690    policy: ExplodePolicy,
691    skipped: AtomicU64,
692    written: AtomicU64,
693    served: AtomicU64,
694    /// The subset of `served` answered by the extent `pread` fast path.
695    pread_served: AtomicU64,
696    rederived: AtomicU64,
697    absent: AtomicU64,
698    /// The file was written with schema v1 (no `enc` column). Every batch this
699    /// process appends to it is then written in v1 shape too — raw, no `enc` —
700    /// because an IPC stream has ONE schema message and a six-column batch
701    /// under a five-column schema does not decode. The pool is a cache; it moves
702    /// to v2 the day it is dropped and rebuilt, never in place.
703    legacy_v1: std::sync::atomic::AtomicBool,
704}
705
706impl ExplodedArchive {
707    /// Open or create the table, with the policy `ZNIPPY_GIT_EXPLODE` names.
708    /// Reads kilobytes, never the payloads.
709    pub fn open(path: &Path) -> Result<Self> {
710        Self::open_with_policy(path, ExplodePolicy::from_env()?)
711    }
712
713    /// Open with an explicit policy — what a test or a per-tenant configuration
714    /// uses, so the setting is not reachable only through the environment.
715    pub fn open_with_policy(path: &Path, policy: ExplodePolicy) -> Result<Self> {
716        if let Some(parent) = path.parent()
717            && !parent.as_os_str().is_empty()
718        {
719            std::fs::create_dir_all(parent)
720                .with_context(|| format!("creating {}", parent.display()))?;
721        }
722        let (schema_msg, batches, end, oid_len) = Self::walk(path)?;
723        let legacy_v1 = schema_msg.as_ref().is_some_and(|m| !Self::schema_has_enc(m));
724        Ok(Self {
725            legacy_v1: std::sync::atomic::AtomicBool::new(legacy_v1),
726            path: path.to_path_buf(),
727            schema_msg: Mutex::new(schema_msg),
728            batches: Mutex::new(batches),
729            index: Mutex::new(None),
730            cache: Mutex::new(Vec::new()),
731            pending: Mutex::new(PendingBatch::default()),
732            pending_bytes: AtomicU64::new(0),
733            end: Mutex::new(end),
734            oid_len: Mutex::new(oid_len),
735            policy,
736            skipped: AtomicU64::new(0),
737            written: AtomicU64::new(0),
738            served: AtomicU64::new(0),
739            pread_served: AtomicU64::new(0),
740            rederived: AtomicU64::new(0),
741            absent: AtomicU64::new(0),
742        })
743    }
744
745    pub fn path(&self) -> &Path {
746        &self.path
747    }
748
749    /// **Drop every derived, rebuildable structure**: the `oid → Located`
750    /// tree (~50 B per row — 700 MB on linux, and re-extended on every flush
751    /// while a reader keeps it alive) and the decoded-batch cache. What the
752    /// memory watchdog's flush reaches for ([`crate::memguard`]); the next
753    /// lookup rebuilds the tree from the file in one pass.
754    pub fn drop_caches(&self) {
755        if let Ok(mut i) = self.index.lock() {
756            *i = None;
757        }
758        if let Ok(mut c) = self.cache.lock() {
759            c.clear();
760        }
761    }
762
763    /// Walk the message framing: `[0xFFFFFFFF][len u32][metadata][body]`.
764    ///
765    /// The metadata flatbuffer carries `bodyLength`, so this SEEKS over every
766    /// payload — opening a 17 GB table touches kilobytes. It also carries a
767    /// record batch's row `length`, which is how `rows()` answers without
768    /// decoding anything.
769    ///
770    /// A torn tail — a half-written message at EOF — stops the walk and keeps
771    /// what came before it, and the file is truncated to that boundary on the
772    /// next append. The table is derived, so a partial batch is re-exploded
773    /// rather than being an error a client ever sees.
774    #[allow(clippy::type_complexity)]
775    fn walk(path: &Path) -> Result<(Option<Vec<u8>>, Vec<Batch>, u64, Option<usize>)> {
776        let Ok(mut f) = File::open(path) else {
777            return Ok((None, Vec::new(), 0, None));
778        };
779        let total = f.metadata()?.len();
780        let mut schema_msg: Option<Vec<u8>> = None;
781        let mut oid_len: Option<usize> = None;
782        let mut batches = Vec::new();
783        let mut at = 0u64;
784        while at + 8 <= total {
785            f.seek(SeekFrom::Start(at))?;
786            let mut hdr = [0u8; 8];
787            if f.read_exact(&mut hdr).is_err() {
788                break;
789            }
790            if u32::from_le_bytes(hdr[0..4].try_into().unwrap()) != 0xFFFF_FFFF {
791                break;
792            }
793            let meta_len = u32::from_le_bytes(hdr[4..8].try_into().unwrap()) as u64;
794            if meta_len == 0 {
795                break; // end-of-stream marker
796            }
797            if at + 8 + meta_len > total {
798                break; // torn metadata
799            }
800            let mut meta = vec![0u8; meta_len as usize];
801            if f.read_exact(&mut meta).is_err() {
802                break;
803            }
804            let Ok(msg) = root_as_message(&meta) else {
805                break;
806            };
807            let body = msg.bodyLength().max(0) as u64;
808            let whole = 8 + meta_len + body;
809            if at + whole > total {
810                break; // torn body
811            }
812            if let Some(rb) = msg.header_as_record_batch() {
813                batches.push(Batch {
814                    at,
815                    len: whole,
816                    rows: rb.length().max(0) as u32,
817                    body: at + 8 + meta_len,
818                    pay_data: payload_data_offset(&meta).unwrap_or(EXTENT_UNAVAILABLE),
819                });
820            } else if msg.header_as_schema().is_some() {
821                schema_msg = Some({
822                    let mut whole_msg = Vec::with_capacity((8 + meta_len) as usize);
823                    whole_msg.extend_from_slice(&hdr);
824                    whole_msg.extend_from_slice(&meta);
825                    whole_msg
826                });
827                oid_len = Self::oid_len_of_schema_msg(schema_msg.as_ref().unwrap());
828            }
829            at += whole;
830        }
831        Ok((schema_msg, batches, at, oid_len))
832    }
833
834    /// The declared oid width, so a reopened table refuses a hash it cannot
835    /// hold instead of writing a batch no reader can line up.
836    /// The schema inside a stream schema message (`msg` = the 8-byte
837    /// continuation header + the flatbuffer), read off the flatbuffer directly.
838    ///
839    /// Not through `StreamDecoder`: fed a schema message alone it returns
840    /// `Ok(None)` and its `schema()` stays `None` — MEASURED 2026-09-05 with a
841    /// probe on a file this writer had just produced (`decode=Ok(false)
842    /// schema=None`). `oid_len_of_schema_msg` had used exactly that path since
843    /// it was written and therefore always answered `None`; nothing depended on
844    /// it, so nothing noticed. `legacy_v1` does depend on it, which is how it
845    /// surfaced: a fresh v2 file was read as v1 and the next append was written
846    /// five columns wide under a six-column schema.
847    fn schema_of_msg(msg: &[u8]) -> Option<Schema> {
848        let meta = msg.get(8..)?;
849        let m = root_as_message(meta).ok()?;
850        let fb = m.header_as_schema()?;
851        Some(znippy_common::arrow::ipc::convert::fb_to_schema(fb))
852    }
853
854    /// Whether the file's schema message carries the v2 `enc` column.
855    fn schema_has_enc(msg: &[u8]) -> bool {
856        Self::schema_of_msg(msg).is_some_and(|s| s.field_with_name(COL_ENC).is_ok())
857    }
858
859    fn oid_len_of_schema_msg(msg: &[u8]) -> Option<usize> {
860        let schema = Self::schema_of_msg(msg)?;
861        match schema.field_with_name(COL_OID).ok()?.data_type() {
862            DataType::FixedSizeBinary(n) => Some(*n as usize),
863            _ => None,
864        }
865    }
866
867    /// One resolved object with no name attached — what [`crate::exploded::PayloadSink`]
868    /// can supply.
869    pub fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
870        self.explode_at(oid, kind, None, None, payload)
871    }
872
873    /// One resolved object **with its name**: the mode and path a tree walk knows.
874    ///
875    /// A blob is reachable at many paths, so this records the first one seen and
876    /// never rewrites it — a row that already exists is not re-keyed by a second
877    /// sighting, because "which of its names" is not a question this table
878    /// pretends to answer.
879    pub fn explode_at(
880        &self,
881        oid: &[u8],
882        kind: GitObjectKind,
883        mode: Option<u32>,
884        path: Option<&str>,
885        payload: &[u8],
886    ) -> Result<()> {
887        if !self.policy.wants(kind) {
888            self.skipped.fetch_add(1, Ordering::Relaxed);
889            return Ok(());
890        }
891        {
892            let mut p = self
893                .pending
894                .lock()
895                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
896            // **One `extend_from_slice` into the arena, no allocation of its
897            // own.** This line was `payload.to_vec()` — a whole second copy of
898            // the object the caller is holding, which is the byte lane AA
899            // measured on 2026-08-30.
900            let pay = p.stash(payload);
901            p.rows.push(Pending {
902                oid: oid.to_vec(),
903                code: kind_code(kind),
904                mode,
905                path: path.map(str::to_owned),
906                pay,
907            });
908        }
909        let n = self
910            .pending_bytes
911            .fetch_add(payload.len() as u64, Ordering::AcqRel)
912            + payload.len() as u64;
913        if n >= FLUSH_BYTES as u64 {
914            self.flush()?;
915        }
916        Ok(())
917    }
918
919    /// `v1`: write the batch in schema v1 — no `enc` column, every row raw —
920    /// because the file it is appended to was opened with a v1 schema message.
921    fn build_batch(rows: &[Pending], arena: &[u8], oid_len: usize, v1: bool) -> Result<RecordBatch> {
922        if v1 {
923            let mut oid_b = FixedSizeBinaryBuilder::with_capacity(rows.len(), oid_len as i32);
924            let mut kind_b = UInt8Builder::with_capacity(rows.len());
925            let mut mode_b = UInt32Builder::with_capacity(rows.len());
926            let mut path_b = StringBuilder::new();
927            let mut pay_b = LargeBinaryBuilder::new();
928            for r in rows {
929                oid_b
930                    .append_value(&r.oid)
931                    .map_err(|e| anyhow::anyhow!("exploded oid column: {e}"))?;
932                kind_b.append_value(r.code);
933                mode_b.append_option(r.mode);
934                path_b.append_option(r.path.as_deref());
935                pay_b.append_value(&arena[r.pay.clone()]);
936            }
937            return RecordBatch::try_new(
938                exploded_schema_v1(oid_len),
939                vec![
940                    Arc::new(oid_b.finish()),
941                    Arc::new(kind_b.finish()),
942                    Arc::new(mode_b.finish()),
943                    Arc::new(path_b.finish()),
944                    Arc::new(pay_b.finish()),
945                ],
946            )
947            .context("building a v1 exploded batch");
948        }
949        let mut oid_b = FixedSizeBinaryBuilder::with_capacity(rows.len(), oid_len as i32);
950        let mut kind_b = UInt8Builder::with_capacity(rows.len());
951        let mut mode_b = UInt32Builder::with_capacity(rows.len());
952        let mut path_b = StringBuilder::new();
953        let mut enc_b = UInt8Builder::with_capacity(rows.len());
954        let mut pay_b = LargeBinaryBuilder::new();
955        for r in rows {
956            oid_b
957                .append_value(&r.oid)
958                .map_err(|e| anyhow::anyhow!("exploded oid column: {e}"))?;
959            kind_b.append_value(r.code);
960            mode_b.append_option(r.mode);
961            path_b.append_option(r.path.as_deref());
962            // Per row, at build time: the pending arena keeps rows raw (so the
963            // pending-serve path hands them out untouched), and what lands in
964            // the file is whatever is smaller.
965            let raw = &arena[r.pay.clone()];
966            match compress_row(raw) {
967                Some(z) => {
968                    enc_b.append_value(ENC_ZLIB);
969                    pay_b.append_value(&z);
970                }
971                None => {
972                    enc_b.append_value(ENC_RAW);
973                    pay_b.append_value(raw);
974                }
975            }
976        }
977        RecordBatch::try_new(
978            exploded_schema(oid_len),
979            vec![
980                Arc::new(oid_b.finish()),
981                Arc::new(kind_b.finish()),
982                Arc::new(mode_b.finish()),
983                Arc::new(path_b.finish()),
984                Arc::new(enc_b.finish()),
985                Arc::new(pay_b.finish()),
986            ],
987        )
988        .context("building the exploded batch")
989    }
990
991    /// Append the buffered rows as ONE batch message. Nothing already in the
992    /// file is read or rewritten.
993    pub fn flush(&self) -> Result<u64> {
994        let (rows, arena) = {
995            let mut p = self
996                .pending
997                .lock()
998                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
999            if p.is_empty() {
1000                return Ok(0);
1001            }
1002            self.pending_bytes.store(0, Ordering::Release);
1003            p.take()
1004        };
1005        let n = rows.len() as u64;
1006        let width = rows[0].oid.len();
1007        anyhow::ensure!(
1008            rows.iter().all(|r| r.oid.len() == width),
1009            "one exploded batch cannot hold two hash widths"
1010        );
1011        {
1012            let mut w = self
1013                .oid_len
1014                .lock()
1015                .map_err(|_| anyhow::anyhow!("exploded oid width poisoned"))?;
1016            match *w {
1017                Some(have) => anyhow::ensure!(
1018                    have == width,
1019                    "this table holds {have}-byte oids and a {width}-byte one arrived"
1020                ),
1021                None => *w = Some(width),
1022            }
1023        }
1024
1025        let batch = Self::build_batch(&rows, &arena, width, self.legacy_v1.load(Ordering::Relaxed))?;
1026        let opts = IpcWriteOptions::default();
1027        let ipc = IpcDataGenerator::default();
1028        let mut tracker = DictionaryTracker::new(false);
1029
1030        // The file is opened per flush rather than held: an append writes at a
1031        // known offset and closing between batches is what makes a crashed
1032        // process leave a file whose last complete message is intact.
1033        let mut end = self
1034            .end
1035            .lock()
1036            .map_err(|_| anyhow::anyhow!("exploded end poisoned"))?;
1037        let mut f = OpenOptions::new()
1038            .create(true)
1039            .read(true)
1040            .write(true)
1041            .truncate(false)
1042            .open(&self.path)
1043            .with_context(|| format!("opening {}", self.path.display()))?;
1044        // Drop anything after the last complete message before appending — a
1045        // torn tail from a previous crash must not sit in the middle of the
1046        // stream, where every later batch would be unreachable.
1047        if f.metadata()?.len() != *end {
1048            f.set_len(*end)
1049                .with_context(|| format!("truncating {} to {}", self.path.display(), *end))?;
1050        }
1051        f.seek(SeekFrom::Start(*end))?;
1052
1053        let mut schema_guard = self
1054            .schema_msg
1055            .lock()
1056            .map_err(|_| anyhow::anyhow!("exploded schema poisoned"))?;
1057        if schema_guard.is_none() {
1058            let enc =
1059                ipc.schema_to_bytes_with_dictionary_tracker(&exploded_schema(width), &mut tracker, &opts);
1060            let mut msg = Vec::new();
1061            let (meta, body) = write_message(&mut msg, enc, &opts)
1062                .map_err(|e| anyhow::anyhow!("encoding the exploded schema: {e}"))?;
1063            debug_assert_eq!(meta + body, msg.len());
1064            f.write_all(&msg).context("writing the exploded schema")?;
1065            *end += msg.len() as u64;
1066            *schema_guard = Some(msg);
1067        }
1068        drop(schema_guard);
1069
1070        let (dicts, enc) = ipc
1071            .encode(&batch, &mut tracker, &opts, &mut Default::default())
1072            .map_err(|e| anyhow::anyhow!("encoding an exploded batch: {e}"))?;
1073        anyhow::ensure!(
1074            dicts.is_empty(),
1075            "the exploded schema has no dictionary columns, yet {} arrived",
1076            dicts.len()
1077        );
1078        let mut msg = Vec::new();
1079        let (meta, body) = write_message(&mut msg, enc, &opts)
1080            .map_err(|e| anyhow::anyhow!("encoding an exploded batch: {e}"))?;
1081        debug_assert_eq!(meta + body, msg.len());
1082        f.write_all(&msg).context("appending an exploded batch")?;
1083
1084        let placed = Batch {
1085            at: *end,
1086            len: msg.len() as u64,
1087            rows: rows.len() as u32,
1088            // `meta` from `write_message` is the 8-byte prefix plus the padded
1089            // metadata, so the body begins exactly `meta` bytes in — and the
1090            // metadata slice between prefix and body is what the offset helper
1091            // parses, the same bytes `walk` reads back after a reopen.
1092            body: *end + meta as u64,
1093            pay_data: payload_data_offset(&msg[8..meta]).unwrap_or(EXTENT_UNAVAILABLE),
1094        };
1095        *end += msg.len() as u64;
1096
1097        // Recorded while the append lock is still held, so the batch table stays
1098        // in file order under N concurrent exploders. Correctness would survive
1099        // it being out of order — a lookup addresses a batch by its slot, not by
1100        // its offset — but a table that reads back in file order is one a human
1101        // can check against the file with `od`.
1102        let batch_no = {
1103            let mut b = self
1104                .batches
1105                .lock()
1106                .map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
1107            b.push(placed);
1108            (b.len() - 1) as u32
1109        };
1110        drop(end);
1111
1112        // Only extend an index that already exists. Building one here would make
1113        // the first push pay for a structure no reader has asked for.
1114        let mut idx = self
1115            .index
1116            .lock()
1117            .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1118        if let Some(old) = idx.take() {
1119            // Retired oids are collected and applied in ONE pass. Removing them
1120            // one at a time is quadratic, and a `gc` retires millions.
1121            let mut retired: std::collections::HashSet<&[u8]> = std::collections::HashSet::new();
1122            for r in rows.iter() {
1123                if kind_of(r.code).is_none() {
1124                    retired.insert(r.oid.as_slice());
1125                }
1126            }
1127            let mut b = OidTreeBuilder::with_capacity(old.len() + rows.len());
1128            for (oid, loc) in old.iter() {
1129                if !retired.contains(oid) {
1130                    b.push(oid, Some(loc))?;
1131                }
1132            }
1133            let pay = batch
1134                .column_by_name(COL_PAYLOAD)
1135                .and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
1136                .context("the exploded table has no payload column")?;
1137            // Freshly built, never sliced, so these offsets are byte-for-byte
1138            // what the encoder just wrote into the file — the same equivalence
1139            // `index_now` relies on after a reopen.
1140            let off = pay.value_offsets();
1141            let enc = enc_column(&batch);
1142            for (i, r) in rows.iter().enumerate() {
1143                if let Some(kind) = kind_of(r.code)
1144                    && !retired.contains(r.oid.as_slice())
1145                {
1146                    let (pay_at, pay_len) = if placed.pay_data == EXTENT_UNAVAILABLE {
1147                        (EXTENT_UNAVAILABLE, 0)
1148                    } else {
1149                        (
1150                            placed.body + placed.pay_data + off[i] as u64,
1151                            (off[i + 1] - off[i]) as u64,
1152                        )
1153                    };
1154                    b.push(
1155                        &r.oid,
1156                        Some(Located {
1157                            batch: batch_no,
1158                            row: i as u32,
1159                            kind,
1160                            pay_at,
1161                            pay_len,
1162                            enc: enc_at(enc, i),
1163                        }),
1164                    )?;
1165                }
1166            }
1167            // `Keep::First`, because the surviving entries were pushed before
1168            // this batch's rows: an oid already in the index keeps the row it
1169            // already had, which is what `explode_at` documents about names.
1170            *idx = Some(b.finish(Keep::First));
1171        }
1172        self.written.fetch_add(n, Ordering::Relaxed);
1173        Ok(n)
1174    }
1175
1176    /// How many decoded batches stay resident. Four × [`FLUSH_BYTES`] is the
1177    /// worst case, and a batch is `Arc`-backed so a hit copies pointers.
1178    const CACHE: usize = 4;
1179
1180    /// Read one batch message off disk and decode it — or hand back the decoded
1181    /// copy if it is one of the last [`Self::CACHE`].
1182    ///
1183    /// Two `pread`s on a miss: the schema message and the batch. It never reads
1184    /// the batches in front of the one it wants — the property a stream reader
1185    /// iterating from byte zero does not have.
1186    fn read_batch(&self, no: u32) -> Result<RecordBatch> {
1187        {
1188            let mut c = self
1189                .cache
1190                .lock()
1191                .map_err(|_| anyhow::anyhow!("exploded batch cache poisoned"))?;
1192            if let Some(at) = c.iter().position(|(n, _)| *n == no) {
1193                let hit = c.remove(at);
1194                let batch = hit.1.clone();
1195                c.insert(0, hit);
1196                return Ok(batch);
1197            }
1198        }
1199        let placed = {
1200            let b = self
1201                .batches
1202                .lock()
1203                .map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
1204            *b.get(no as usize)
1205                .with_context(|| format!("exploded batch {no} is not in this table"))?
1206        };
1207        let schema_msg = {
1208            let s = self
1209                .schema_msg
1210                .lock()
1211                .map_err(|_| anyhow::anyhow!("exploded schema poisoned"))?;
1212            s.clone()
1213                .context("the exploded table has batches but no schema message")?
1214        };
1215        let f = File::open(&self.path)
1216            .with_context(|| format!("opening {}", self.path.display()))?;
1217        let mut raw = vec![0u8; placed.len as usize];
1218        f.read_exact_at(&mut raw, placed.at)
1219            .with_context(|| format!("reading exploded batch {no}"))?;
1220
1221        let mut dec = StreamDecoder::new();
1222        let mut head = Buffer::from_vec(schema_msg);
1223        dec.decode(&mut head)
1224            .map_err(|e| anyhow::anyhow!("decoding the exploded schema: {e}"))?;
1225        let mut body = Buffer::from_vec(raw);
1226        let batch = dec
1227            .decode(&mut body)
1228            .map_err(|e| anyhow::anyhow!("decoding exploded batch {no}: {e}"))?
1229            .with_context(|| format!("exploded batch {no} decoded to no rows"))?;
1230        {
1231            let mut c = self
1232                .cache
1233                .lock()
1234                .map_err(|_| anyhow::anyhow!("exploded batch cache poisoned"))?;
1235            c.retain(|(n, _)| *n != no);
1236            c.insert(0, (no, batch.clone()));
1237            c.truncate(Self::CACHE);
1238        }
1239        Ok(batch)
1240    }
1241
1242    /// Build `oid → Located` with one sequential pass, applying every
1243    /// [`TOMBSTONE`] in file order so a retired oid stays retired across a
1244    /// reopen. Called on the first lookup, never on open.
1245    fn index_now(&self) -> Result<()> {
1246        {
1247            let idx = self
1248                .index
1249                .lock()
1250                .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1251            if idx.is_some() {
1252                return Ok(());
1253            }
1254        }
1255        let (count, rows, placed) = {
1256            let b = self
1257                .batches
1258                .lock()
1259                .map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
1260            // The row total is in the batch metadata the framing walk already
1261            // read, so the builder's arrays are sized once instead of doubling
1262            // their way to 11.7 million.
1263            (b.len() as u32, b.iter().map(|x| x.rows as usize).sum::<usize>(), b.clone())
1264        };
1265        // Every row is recorded as a sighting and the fold happens once, at
1266        // `finish`. The `HashMap<Vec<u8>, Located>` this replaced allocated —
1267        // and then dropped — one `Vec<u8>` per oid on a corpus with 11.7 M of
1268        // them, to compute an answer a single stable sort gives.
1269        let mut fold = OidTreeBuilder::with_capacity(rows);
1270        for no in 0..count {
1271            let batch = self.read_batch(no)?;
1272            let oid = batch
1273                .column_by_name(COL_OID)
1274                .and_then(|c| c.as_any().downcast_ref::<FixedSizeBinaryArray>())
1275                .context("the exploded table has no oid column")?;
1276            let kinds = batch
1277                .column_by_name(COL_TYPE)
1278                .and_then(|c| c.as_any().downcast_ref::<UInt8Array>())
1279                .context("the exploded table has no object_type column")?;
1280            let pay = batch
1281                .column_by_name(COL_PAYLOAD)
1282                .and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
1283                .context("the exploded table has no payload column")?;
1284            // The decoded offsets carry the SAME values the file's offsets
1285            // buffer holds — a StreamDecoder copies buffers verbatim — so the
1286            // extent needs no second parse of the body.
1287            let off = pay.value_offsets();
1288            let enc = enc_column(&batch);
1289            let b = placed[no as usize];
1290            for i in 0..batch.num_rows() {
1291                // A tombstone is pushed as a sighting with no location, not
1292                // skipped: batches are walked in file order and rows in row
1293                // order, so `Keep::Last` gives the last word about an oid — and
1294                // when that word is a tombstone the oid is dropped, which is
1295                // what makes the tombstone durable rather than advisory.
1296                fold.push(
1297                    oid.value(i),
1298                    kind_of(kinds.value(i)).map(|kind| {
1299                        let (pay_at, pay_len) = if b.pay_data == EXTENT_UNAVAILABLE {
1300                            (EXTENT_UNAVAILABLE, 0)
1301                        } else {
1302                            (
1303                                b.body + b.pay_data + off[i] as u64,
1304                                (off[i + 1] - off[i]) as u64,
1305                            )
1306                        };
1307                        Located { batch: no, row: i as u32, kind, pay_at, pay_len, enc: enc_at(enc, i) }
1308                    }),
1309                )?;
1310            }
1311        }
1312        let out = fold.finish(Keep::Last);
1313        let mut idx = self
1314            .index
1315            .lock()
1316            .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1317        if idx.is_none() {
1318            *idx = Some(out);
1319        }
1320        Ok(())
1321    }
1322
1323    fn find(&self, oid: &[u8]) -> Result<Option<Located>> {
1324        self.index_now()?;
1325        let idx = self
1326            .index
1327            .lock()
1328            .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1329        Ok(idx.as_ref().expect("index_now built it").find(oid))
1330    }
1331
1332    /// The object's kind and bytes, or `None` — *fall back and rebuild*, never
1333    /// *wrong*.
1334    pub fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
1335        // A buffered row has not been written yet, and a read must not miss it:
1336        // an "absent" that is really "not flushed" sends the caller off to
1337        // re-resolve a whole pack for an object we are holding in RAM.
1338        {
1339            let p = self
1340                .pending
1341                .lock()
1342                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1343            if let Some(r) = p.rows.iter().rev().find(|r| r.oid == oid) {
1344                let Some(kind) = kind_of(r.code) else {
1345                    // The newest buffered word about this oid retires it.
1346                    self.absent.fetch_add(1, Ordering::Relaxed);
1347                    return Ok(None);
1348                };
1349                self.served.fetch_add(1, Ordering::Relaxed);
1350                return Ok(Some((kind, p.bytes(r).to_vec())));
1351            }
1352        }
1353        let Some(loc) = self.find(oid)? else {
1354            self.absent.fetch_add(1, Ordering::Relaxed);
1355            return Ok(None);
1356        };
1357        // ── the point read IS a point read ──────────────────────────────────
1358        //
1359        // One pread of exactly the payload's bytes, addressed by the extent the
1360        // index carries. The batch-decode below is the FALLBACK — correct for a
1361        // compressed or foreign-layout batch — not the path a healthy table
1362        // takes: taking it per object is what turned a rust-lang/rust
1363        // connectivity walk into hours of one-core batch decoding (2026-08-19).
1364        if loc.pay_at != EXTENT_UNAVAILABLE {
1365            let f = File::open(&self.path)
1366                .with_context(|| format!("opening {}", self.path.display()))?;
1367            let mut buf = vec![0u8; loc.pay_len as usize];
1368            f.read_exact_at(&mut buf, loc.pay_at)
1369                .with_context(|| format!("preading payload at {}", loc.pay_at))?;
1370            self.served.fetch_add(1, Ordering::Relaxed);
1371            self.pread_served.fetch_add(1, Ordering::Relaxed);
1372            // The extent held the row as stored; a compressed row is inflated
1373            // here, from the index's `enc`, without decoding the batch.
1374            return Ok(Some((loc.kind, row_payload(loc.enc, &buf)?)));
1375        }
1376        let batch = self.read_batch(loc.batch)?;
1377        let pay = batch
1378            .column_by_name(COL_PAYLOAD)
1379            .and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
1380            .context("the exploded table has no payload column")?;
1381        let enc = enc_at(enc_column(&batch), loc.row as usize);
1382        self.served.fetch_add(1, Ordering::Relaxed);
1383        Ok(Some((loc.kind, row_payload(enc, pay.value(loc.row as usize))?)))
1384    }
1385
1386    /// Point reads served by the extent `pread` rather than a batch decode —
1387    /// how a test proves the fast path is the one actually taken, instead of
1388    /// timing something.
1389    pub fn pread_served(&self) -> u64 {
1390        self.pread_served.load(Ordering::Relaxed)
1391    }
1392
1393    /// The name this object was first exploded at, if any caller knew one.
1394    /// `None` is "not recorded", never "at the root".
1395    pub fn name(&self, oid: &[u8]) -> Result<Option<(Option<u32>, Option<String>)>> {
1396        {
1397            let p = self
1398                .pending
1399                .lock()
1400                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1401            if let Some(r) = p.rows.iter().rev().find(|r| r.oid == oid) {
1402                return match kind_of(r.code) {
1403                    Some(_) => Ok(Some((r.mode, r.path.clone()))),
1404                    None => Ok(None),
1405                };
1406            }
1407        }
1408        let Some(loc) = self.find(oid)? else {
1409            return Ok(None);
1410        };
1411        let batch = self.read_batch(loc.batch)?;
1412        let mode = batch
1413            .column_by_name(COL_MODE)
1414            .and_then(|c| c.as_any().downcast_ref::<UInt32Array>())
1415            .context("the exploded table has no mode column")?;
1416        let path = batch
1417            .column_by_name(COL_PATH)
1418            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
1419            .context("the exploded table has no path column")?;
1420        let i = loc.row as usize;
1421        Ok(Some((
1422            (!mode.is_null(i)).then(|| mode.value(i)),
1423            (!path.is_null(i)).then(|| path.value(i).to_owned()),
1424        )))
1425    }
1426
1427    pub fn has(&self, oid: &[u8]) -> Result<bool> {
1428        {
1429            let p = self
1430                .pending
1431                .lock()
1432                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1433            if let Some(r) = p.rows.iter().rev().find(|r| r.oid == oid) {
1434                return Ok(kind_of(r.code).is_some());
1435            }
1436        }
1437        Ok(self.find(oid)?.is_some())
1438    }
1439
1440    /// Every oid of one kind, with its payload. The graph fold's input.
1441    ///
1442    /// Driven by the **index**, not by a scan of the file, because the file
1443    /// holds superseded and retired rows too. A scan that read the file directly
1444    /// is exactly how `gc`'d commits came back after a reopen.
1445    pub fn of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
1446        self.flush()?;
1447        self.index_now()?;
1448        // Grouped by batch so each one is decoded once, in file order.
1449        let mut by_batch: std::collections::BTreeMap<u32, Vec<(u32, Vec<u8>)>> =
1450            std::collections::BTreeMap::new();
1451        {
1452            let idx = self
1453                .index
1454                .lock()
1455                .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1456            for (oid, loc) in idx.as_ref().expect("index_now built it").iter() {
1457                if loc.kind == kind {
1458                    by_batch.entry(loc.batch).or_default().push((loc.row, oid.to_vec()));
1459                }
1460            }
1461        }
1462        let mut out = Vec::new();
1463        for (no, mut wanted) in by_batch {
1464            wanted.sort_unstable_by_key(|(r, _)| *r);
1465            let batch = self.read_batch(no)?;
1466            let pay = batch
1467                .column_by_name(COL_PAYLOAD)
1468                .and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
1469                .context("the exploded table has no payload column")?;
1470            let enc = enc_column(&batch);
1471            for (row, oid) in wanted {
1472                out.push((oid, row_payload(enc_at(enc, row as usize), pay.value(row as usize))?));
1473            }
1474        }
1475        Ok(out)
1476    }
1477
1478    /// Retire every oid `live` rejects, **durably**.
1479    ///
1480    /// The file is append-only, so this appends a [`TOMBSTONE`] row per retired
1481    /// oid rather than deleting anything. The bytes of the dead rows stay until
1482    /// the table is rebuilt — reclaiming them means rewriting up to 17 GB, which
1483    /// is a job with its own proof obligations and not something to smuggle into
1484    /// `gc` — but the *rows* are gone from every reader, including a reopened
1485    /// one.
1486    ///
1487    /// Dropping only from the in-memory index is what a first cut did. Seen RED
1488    /// by `store::tests::a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened`:
1489    /// **"the graph after gc + reopen holds 551 commits, not the one live one"**.
1490    pub fn retain(&self, live: &dyn Fn(&[u8]) -> bool) -> Result<u64> {
1491        self.flush()?;
1492        self.index_now()?;
1493        let dead: Vec<Vec<u8>> = {
1494            let idx = self
1495                .index
1496                .lock()
1497                .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1498            idx.as_ref()
1499                .expect("index_now built it")
1500                .iter()
1501                .filter(|(oid, _)| !live(oid))
1502                .map(|(oid, _)| oid.to_vec())
1503                .collect()
1504        };
1505        if dead.is_empty() {
1506            return Ok(0);
1507        }
1508        {
1509            let mut p = self
1510                .pending
1511                .lock()
1512                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1513            for oid in &dead {
1514                p.rows.push(Pending {
1515                    oid: oid.clone(),
1516                    code: TOMBSTONE,
1517                    mode: None,
1518                    path: None,
1519                    pay: 0..0,
1520                });
1521            }
1522        }
1523        // Written before returning, so a `gc` that says it retired rows has
1524        // already said so on disk.
1525        self.flush()?;
1526        Ok(dead.len() as u64)
1527    }
1528
1529    /// Rows in the table. Answered from batch metadata when no index has been
1530    /// built, so startup does not read a payload.
1531    pub fn rows(&self) -> Result<u64> {
1532        let pending = {
1533            let p = self
1534                .pending
1535                .lock()
1536                .map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
1537            p.rows.len() as u64
1538        };
1539        {
1540            let idx = self
1541                .index
1542                .lock()
1543                .map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
1544            if let Some(idx) = idx.as_ref() {
1545                return Ok(idx.len() as u64 + pending);
1546            }
1547        }
1548        let b = self
1549            .batches
1550            .lock()
1551            .map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
1552        Ok(b.iter().map(|x| x.rows as u64).sum::<u64>() + pending)
1553    }
1554
1555    /// Bytes the table occupies — the number redb could not keep anywhere near
1556    /// its payload total.
1557    pub fn disk_bytes(&self) -> u64 {
1558        std::fs::metadata(&self.path).map(|m| m.len()).unwrap_or(0)
1559    }
1560
1561    pub fn note_rederived(&self) {
1562        self.rederived.fetch_add(1, Ordering::Relaxed);
1563    }
1564
1565    /// What this table is configured to keep.
1566    pub fn policy(&self) -> ExplodePolicy {
1567        self.policy
1568    }
1569
1570    /// Objects the policy declined, COUNTED: a table short because of policy and
1571    /// one short because of a fault must not look alike to `adopt_journal`.
1572    pub fn skipped(&self) -> u64 {
1573        self.skipped.load(Ordering::Relaxed)
1574    }
1575
1576    pub fn stats(&self) -> Stats {
1577        Stats {
1578            rows: self.rows().unwrap_or(0),
1579            written: self.written.load(Ordering::Relaxed),
1580            served: self.served.load(Ordering::Relaxed),
1581            rederived: self.rederived.load(Ordering::Relaxed),
1582            absent: self.absent.load(Ordering::Relaxed),
1583        }
1584    }
1585
1586    /// The engine-wide counter type, so no caller can tell the medium apart by
1587    /// its instruments.
1588    pub fn engine_stats(&self) -> crate::exploded::ExplodedStats {
1589        let s = self.stats();
1590        crate::exploded::ExplodedStats {
1591            rows: s.rows,
1592            written: s.written,
1593            served: s.served,
1594            rederived: s.rederived,
1595            absent: s.absent,
1596        }
1597    }
1598}
1599
1600impl crate::exploded::PayloadSink for ExplodedArchive {
1601    /// The policy, asked before the payload exists rather than after — and the
1602    /// `skipped` counter is bumped here, which is the only place the object is
1603    /// ever seen when the answer is no.
1604    fn wants(&self, kind: GitObjectKind) -> bool {
1605        if self.policy.wants(kind) {
1606            return true;
1607        }
1608        self.skipped.fetch_add(1, Ordering::Relaxed);
1609        false
1610    }
1611
1612    fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
1613        ExplodedArchive::explode(self, oid, kind, payload)
1614    }
1615
1616    fn flush(&self) -> Result<()> {
1617        ExplodedArchive::flush(self).map(|_| ())
1618    }
1619}
1620
1621// ════════════════════════════════════════════════════════════════════════════
1622// The policy: what gets exploded at all
1623// ════════════════════════════════════════════════════════════════════════════
1624
1625/// **How much of a pack is resolved into the exploded table.**
1626///
1627/// Three values and not a boolean, because a flat off-switch reintroduces a
1628/// correctness bug this table exists to fix. Commit and tree payloads used to
1629/// live only in RAM, so a **clean** shutdown came back with every pack's
1630/// `indexed` bit legitimately set and an EMPTY graph — MEASURED on oden
1631/// 2026-08-08: 2687 rows and **0 of 551 commits**, `reachable()` returning a
1632/// commit instead of its closure and `gc` seeing an empty live set.
1633///
1634/// MEASURED on `linux.git`: resolved content is 16.8 GB against a 6.4 GB pack,
1635/// and blobs are nearly all of it. So [`Graph`](ExplodePolicy::Graph) buys the
1636/// correctness fix for a small fraction of the disk, and
1637/// [`Full`](ExplodePolicy::Full) — one-lookup content reads, thin-pack bases,
1638/// and the path columns that make the table a filesystem — is the part worth
1639/// charging for.
1640///
1641/// **The default is [`Commits`](ExplodePolicy::Commits) since lane Q
1642/// (2026-08-30), and the change is a measurement.** §14's `indexed` bit is
1643/// derived by comparing this table's row count against the index's, so a table
1644/// that is *deliberately* short is indistinguishable, on disk, from one that was
1645/// dropped — see [`crate::git_ops::Absorber::adopt_journal`], which weakens that
1646/// comparison for every deliberately-short policy and cannot make it exact
1647/// without a durable per-kind count. That weakening is the price of the
1648/// default, it is stated here rather than hidden, and it costs a store that
1649/// really did lose its table one extra fall-back read per object rather than a
1650/// re-ingest.
1651#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1652pub enum ExplodePolicy {
1653    /// Store nothing. Reads re-resolve, and every pack is re-queued on reopen —
1654    /// which is what rebuilds the graph, since nothing it is folded from is on
1655    /// disk.
1656    Off,
1657    /// Commits and trees, both eager. The shape lane Q shipped first and
1658    /// **measured**: on the real `torvalds/linux` mirror the tree payloads
1659    /// alone are **28.24 GB — 3.24× the 8.13 GiB pack** — and folding them into
1660    /// RAM took peak RSS to 50.9 GiB. Kept as a setting, because a small
1661    /// repository that is read constantly wants exactly this; not the default,
1662    /// because an appliance ships with 16 GB.
1663    Graph,
1664    /// **Commits and tags only — the SHIPPED DEFAULT since lane Q
1665    /// (2026-08-30).**
1666    ///
1667    /// The index-only-at-push shape of
1668    /// `ideas/lazy-explode-index-only-at-push.md`, in full. Commits are what
1669    /// `graph` and `reach` fold and what a visibility or pull check reads; a
1670    /// tag is what a clone peels. Both are small and both stay eager, so
1671    /// neither ever stalls. **Trees and blobs** — 90 % of the objects and
1672    /// essentially all of the bytes — are written into
1673    /// [`crate::exploded_cache`]'s budgeted pool by the FIRST reader that
1674    /// inflates one, and evicted from it when the budget says so.
1675    ///
1676    /// Trees became lazy when [`crate::reach::TreeSource`] replaced the
1677    /// resident `HashMap<String, Vec<u8>>` that [`crate::serve`]'s closure walk
1678    /// and the reachability builder both read. That map was the 28.24 GB above.
1679    ///
1680    /// MEASURED, `torvalds/linux`, oden 2026-08-30, /data bytes per pack byte:
1681    ///
1682    /// | | eager (`Full`) | `Graph` | `Commits` |
1683    /// |---|---:|---:|---:|
1684    /// | at ack | 1.05× | 1.05× | 1.05× |
1685    /// | at index-complete | 22.4× (lane P, 78 % done) | 4.58× | see the lane Q rows |
1686    #[default]
1687    Commits,
1688    /// Every object, eagerly, with no bound. What shipped before lane Q and
1689    /// what the 17× measurement is of; kept so a bench can measure the
1690    /// unbounded shape and so a store written under it still opens.
1691    Full,
1692}
1693
1694impl ExplodePolicy {
1695    pub const fn wants(self, kind: GitObjectKind) -> bool {
1696        match self {
1697            ExplodePolicy::Off => false,
1698            ExplodePolicy::Graph => matches!(kind, GitObjectKind::Commit | GitObjectKind::Tree),
1699            ExplodePolicy::Commits => matches!(kind, GitObjectKind::Commit | GitObjectKind::Tag),
1700            ExplodePolicy::Full => true,
1701        }
1702    }
1703    pub const fn as_str(self) -> &'static str {
1704        match self {
1705            ExplodePolicy::Off => "off",
1706            ExplodePolicy::Graph => "graph",
1707            ExplodePolicy::Commits => "commits",
1708            ExplodePolicy::Full => "full",
1709        }
1710    }
1711    /// An unknown value is an ERROR, never a silent fall back to the default: a
1712    /// typo that quietly disables a paid tier is not noticed until a bill is
1713    /// wrong.
1714    pub fn parse(s: &str) -> Result<Self> {
1715        match s.trim().to_ascii_lowercase().as_str() {
1716            "off" | "none" => Ok(ExplodePolicy::Off),
1717            "graph" | "commits-and-trees" => Ok(ExplodePolicy::Graph),
1718            "commits" | "lazy" | "commits-only" => Ok(ExplodePolicy::Commits),
1719            "full" | "all" => Ok(ExplodePolicy::Full),
1720            other => anyhow::bail!(
1721                "unknown explode policy {other:?}; expected off, commits, graph or full"
1722            ),
1723        }
1724    }
1725    /// Reads `ZNIPPY_GIT_EXPLODE` ([`crate::arms::ENV_EXPLODE`]) through the
1726    /// crate's one counted door, once per store open.
1727    pub fn from_env() -> Result<Self> {
1728        match crate::arms::read_env(crate::arms::ENV_EXPLODE) {
1729            Some(v) => Self::parse(&v),
1730            None => Ok(Self::default()),
1731        }
1732    }
1733}
1734
1735#[cfg(test)]
1736mod tests {
1737    use super::*;
1738    use crate::exploded::PayloadSink as _;
1739
1740    fn tmp(name: &str) -> PathBuf {
1741        let d = std::env::temp_dir().join(format!("exploded-arrow-{}-{name}", std::process::id()));
1742        let _ = std::fs::remove_dir_all(&d);
1743        std::fs::create_dir_all(&d).unwrap();
1744        d.join("objects.exploded")
1745    }
1746
1747    fn oid(n: u8) -> [u8; 20] {
1748        [n; 20]
1749    }
1750
1751    /// ★ ONE TABLE, and the payload is IN the row.
1752    ///
1753    /// Red-first: drop the payload column and nothing below compiles; widen
1754    /// `payload` to `Binary` and this fails on the type it asserts.
1755    #[test]
1756    fn the_table_holds_the_payload_and_a_name_in_one_row() {
1757        let s = exploded_schema(20);
1758        let names: Vec<&str> = s.fields().iter().map(|f| f.name().as_str()).collect();
1759        // `enc` sits before `payload` so the payload data buffer stays the last
1760        // buffer of the batch — `payload_data_offset` depends on that.
1761        assert_eq!(names, vec![COL_OID, COL_TYPE, COL_MODE, COL_PATH, COL_ENC, COL_PAYLOAD]);
1762        assert_eq!(
1763            s.field_with_name(COL_PAYLOAD).unwrap().data_type(),
1764            &DataType::LargeBinary,
1765            "Binary offsets are i32 and would cap one batch at 2 GiB of payload"
1766        );
1767        assert!(
1768            s.field_with_name(COL_PATH).unwrap().is_nullable(),
1769            "a bare pack resolve knows no path; null must be a legal answer"
1770        );
1771    }
1772
1773    /// ★ THE WHOLE POINT: the file stays near the payload total.
1774    ///
1775    /// redb held 16.8 GB of kernel payload in a 204 GB file — 12×. An Arrow IPC
1776    /// stream has no pages to rewrite, so the overhead is framing. Red-first:
1777    /// put this back on a copy-on-write B-tree and it fails.
1778    #[test]
1779    fn the_file_stays_close_to_the_payload_total() {
1780        let t = ExplodedArchive::open_with_policy(&tmp("size"), ExplodePolicy::Full).unwrap();
1781        let payload = vec![7u8; 4096];
1782        let n = 2000u64;
1783        for i in 0..n {
1784            let mut o = [0u8; 20];
1785            o[..8].copy_from_slice(&i.to_le_bytes());
1786            t.explode(&o, GitObjectKind::Blob, &payload).unwrap();
1787        }
1788        t.flush().unwrap();
1789        let payload_total = n * payload.len() as u64;
1790        let on_disk = t.disk_bytes();
1791        assert!(
1792            on_disk < payload_total + payload_total / 4,
1793            "{on_disk} bytes on disk against {payload_total} of payload — the medium is amplifying"
1794        );
1795        assert_eq!(t.rows().unwrap(), n);
1796    }
1797
1798    /// ★ A point read is one small `pread` of the payload's own bytes, and it
1799    /// is the path ACTUALLY taken — counted, not timed.
1800    ///
1801    /// Red-first, measured on a rust-lang/rust push (2026-08-19): without the
1802    /// extent, every `content` decoded its whole 64 MiB batch against a 4-slot
1803    /// cache, and the pre-ack connectivity walk — graph order, so effectively
1804    /// random across ~574 batches — decoded 6.7 GB/s of page cache to serve
1805    /// ~98 objects/s, one core pinned for hours. Both index-building paths are
1806    /// exercised: the in-process flush extension AND `index_now` after a
1807    /// reopen, against multi-batch tables with varied payload sizes, so a
1808    /// mis-computed extent returns the WRONG BYTES here rather than in a
1809    /// customer's clone.
1810    #[test]
1811    fn a_point_read_preads_the_exact_payload_without_decoding_the_batch() {
1812        let path = tmp("pread");
1813        let pay = |b: u8, i: u8| vec![b ^ i; 100 + i as usize * 7];
1814        let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
1815        // Batch 0, then force the index to exist so batch 1 goes through the
1816        // flush EXTENSION path rather than a later index_now.
1817        for i in 0..40u8 {
1818            t.explode(&[i; 20], GitObjectKind::Blob, &pay(0, i)).unwrap();
1819        }
1820        t.flush().unwrap();
1821        assert!(t.content(&[0u8; 20]).unwrap().is_some(), "build the index");
1822        for i in 40..80u8 {
1823            t.explode(&[i; 20], GitObjectKind::Blob, &pay(1, i)).unwrap();
1824        }
1825        t.flush().unwrap();
1826
1827        let before = t.pread_served();
1828        for i in 0..80u8 {
1829            let (kind, got) = t.content(&[i; 20]).unwrap().expect("every oid is live");
1830            assert_eq!(kind, GitObjectKind::Blob);
1831            let want = if i < 40 { pay(0, i) } else { pay(1, i) };
1832            assert_eq!(got, want, "oid {i}: the pread returned some OTHER bytes");
1833        }
1834        assert_eq!(
1835            t.pread_served() - before,
1836            80,
1837            "a read fell back to the batch decode — the extent was not computed"
1838        );
1839
1840        // Reopen: the framing walk + index_now must reproduce the same extents
1841        // from the file alone.
1842        drop(t);
1843        let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
1844        for i in 0..80u8 {
1845            let (_, got) = t.content(&[i; 20]).unwrap().expect("survives a reopen");
1846            let want = if i < 40 { pay(0, i) } else { pay(1, i) };
1847            assert_eq!(got, want, "oid {i} after reopen");
1848        }
1849        assert_eq!(t.pread_served(), 80, "the reopened table must pread too");
1850    }
1851
1852    /// **Lever 3: a compressible row is stored compressed and read back whole, on
1853    /// every read path.** Three rows: a 4 KiB compressible one (stored zlib), a
1854    /// row under [`MIN_ROW_COMPRESS`] (stored raw), and a 4 KiB incompressible
1855    /// one (zlib cannot get it under the keep threshold — stored raw). Asserted
1856    /// on the `enc` column read straight off the file, on the file being smaller
1857    /// than the payloads, and on `content` (the pread fast path — `pread_served`
1858    /// moves) and `of_kind` (the batch-decode path) both handing back the
1859    /// original bytes.
1860    ///
1861    /// Seen RED by returning `bytes.to_vec()` for `ENC_ZLIB` in `row_payload`:
1862    /// "the pread path returned the stored bytes, not the payload" — 4096 bytes
1863    /// expected, ~60 came back. Restored.
1864    #[test]
1865    fn a_compressible_row_is_stored_compressed_and_read_back_whole() {
1866        let path = tmp("lever3");
1867        let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
1868        let big: Vec<u8> = b"tree 0123456789abcdef0123456789abcdef01234567\nparent x\n"
1869            .iter()
1870            .cycle()
1871            .take(4096)
1872            .copied()
1873            .collect();
1874        let tiny = b"tree x\n".to_vec();
1875        let noise: Vec<u8> = {
1876            let mut x: u32 = 0x1234_5678;
1877            (0..4096)
1878                .map(|_| {
1879                    x = x.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
1880                    (x >> 24) as u8
1881                })
1882                .collect()
1883        };
1884        assert!(tiny.len() < MIN_ROW_COMPRESS && big.len() >= MIN_ROW_COMPRESS);
1885        t.explode(&oid(1), GitObjectKind::Commit, &big).unwrap();
1886        t.explode(&oid(2), GitObjectKind::Commit, &tiny).unwrap();
1887        t.explode(&oid(3), GitObjectKind::Blob, &noise).unwrap();
1888        t.flush().unwrap();
1889
1890        // The file holds the compressed row: smaller than the payloads it
1891        // carries, which a raw pool never is.
1892        let file = std::fs::metadata(&path).unwrap().len() as usize;
1893        assert!(
1894            file < big.len() + tiny.len() + noise.len(),
1895            "the pool did not compress: {file} B for {} B of payload",
1896            big.len() + tiny.len() + noise.len()
1897        );
1898
1899        // The `enc` column, read off the file as the plain IPC stream it is.
1900        let rd = znippy_common::arrow::ipc::reader::StreamReader::try_new(
1901            std::fs::File::open(&path).unwrap(),
1902            None,
1903        )
1904        .unwrap();
1905        let mut encs = std::collections::BTreeMap::new();
1906        for b in rd {
1907            let b = b.unwrap();
1908            let oids = b
1909                .column_by_name(COL_OID)
1910                .and_then(|c| c.as_any().downcast_ref::<FixedSizeBinaryArray>())
1911                .unwrap();
1912            let enc = enc_column(&b).expect("a v2 pool has the enc column");
1913            for i in 0..b.num_rows() {
1914                encs.insert(oids.value(i)[0], enc.value(i));
1915            }
1916        }
1917        assert_eq!(encs[&1], ENC_ZLIB, "the compressible row is stored compressed");
1918        assert_eq!(encs[&2], ENC_RAW, "a row under the floor stays raw");
1919        assert_eq!(encs[&3], ENC_RAW, "an incompressible row stays raw");
1920
1921        // The pread fast path inflates.
1922        let before = t.pread_served();
1923        let (k, got) = t.content(&oid(1)).unwrap().unwrap();
1924        assert_eq!(k, GitObjectKind::Commit);
1925        assert_eq!(got, big, "the pread path returned the stored bytes, not the payload");
1926        assert_eq!(t.content(&oid(2)).unwrap().unwrap().1, tiny);
1927        assert_eq!(t.content(&oid(3)).unwrap().unwrap().1, noise);
1928        assert_eq!(t.pread_served() - before, 3, "all three went through the extent pread");
1929
1930        // The batch-decode path (`of_kind`) inflates too.
1931        let commits = t.of_kind(GitObjectKind::Commit).unwrap();
1932        let by: std::collections::HashMap<u8, Vec<u8>> =
1933            commits.into_iter().map(|(o, p)| (o[0], p)).collect();
1934        assert_eq!(by[&1], big, "of_kind returned the stored bytes, not the payload");
1935        assert_eq!(by[&2], tiny);
1936
1937        // And across a reopen, where the index is rebuilt from the file.
1938        drop(t);
1939        let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
1940        assert_eq!(t.content(&oid(1)).unwrap().unwrap().1, big, "after reopen");
1941        assert_eq!(t.content(&oid(3)).unwrap().unwrap().1, noise, "after reopen");
1942    }
1943
1944    /// **A v1 pool — no `enc` column — still reads, every row raw.** Written by
1945    /// hand in v1's shape (five columns, the same framing `flush` uses), then
1946    /// opened by this build: the extent is computed (12 buffers is accepted),
1947    /// `enc` is absent and read as [`ENC_RAW`], and the bytes come back as
1948    /// written. This is the promise that lets an existing appliance's pools
1949    /// open unchanged.
1950    ///
1951    /// Seen RED by making `payload_data_offset` accept only 14 buffers: the
1952    /// pread count stayed 0 ("a v1 pool must still take the extent fast path").
1953    /// Restored.
1954    #[test]
1955    fn a_v1_pool_without_the_enc_column_reads_raw() {
1956        use znippy_common::arrow::array::{
1957            FixedSizeBinaryBuilder, LargeBinaryBuilder, StringBuilder, UInt32Builder,
1958        };
1959        use znippy_common::arrow::ipc::writer::{write_message, DictionaryTracker, IpcDataGenerator, IpcWriteOptions};
1960        let path = tmp("v1pool");
1961        let v1_schema: SchemaRef = Arc::new(Schema::new(vec![
1962            Field::new(COL_OID, DataType::FixedSizeBinary(20), false),
1963            Field::new(COL_TYPE, DataType::UInt8, false),
1964            Field::new(COL_MODE, DataType::UInt32, true),
1965            Field::new(COL_PATH, DataType::Utf8, true),
1966            Field::new(COL_PAYLOAD, DataType::LargeBinary, false),
1967        ]));
1968        let payloads: Vec<Vec<u8>> = (1..=5u8)
1969            .map(|i| b"tree abc\n".iter().cycle().take(300 + i as usize * 100).copied().collect())
1970            .collect();
1971        let mut oid_b = FixedSizeBinaryBuilder::with_capacity(5, 20);
1972        let mut kind_b = UInt8Builder::with_capacity(5);
1973        let mut mode_b = UInt32Builder::with_capacity(5);
1974        let mut path_b = StringBuilder::new();
1975        let mut pay_b = LargeBinaryBuilder::new();
1976        for (i, p) in payloads.iter().enumerate() {
1977            oid_b.append_value(oid(i as u8 + 1)).unwrap();
1978            kind_b.append_value(kind_code(GitObjectKind::Commit));
1979            mode_b.append_null();
1980            path_b.append_null();
1981            pay_b.append_value(p);
1982        }
1983        let batch = RecordBatch::try_new(
1984            v1_schema.clone(),
1985            vec![
1986                Arc::new(oid_b.finish()),
1987                Arc::new(kind_b.finish()),
1988                Arc::new(mode_b.finish()),
1989                Arc::new(path_b.finish()),
1990                Arc::new(pay_b.finish()),
1991            ],
1992        )
1993        .unwrap();
1994        let opts = IpcWriteOptions::default();
1995        let ipc = IpcDataGenerator::default();
1996        let mut tracker = DictionaryTracker::new(false);
1997        let mut file = Vec::new();
1998        {
1999            let enc = ipc.schema_to_bytes_with_dictionary_tracker(&v1_schema, &mut tracker, &opts);
2000            let mut msg = Vec::new();
2001            write_message(&mut msg, enc, &opts).unwrap();
2002            file.extend_from_slice(&msg);
2003        }
2004        {
2005            let (_, enc) = ipc.encode(&batch, &mut tracker, &opts, &mut Default::default()).unwrap();
2006            let mut msg = Vec::new();
2007            write_message(&mut msg, enc, &opts).unwrap();
2008            file.extend_from_slice(&msg);
2009        }
2010        std::fs::write(&path, &file).unwrap();
2011
2012        let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
2013        for (i, p) in payloads.iter().enumerate() {
2014            let (k, got) = t.content(&oid(i as u8 + 1)).unwrap().expect("a v1 row is live");
2015            assert_eq!(k, GitObjectKind::Commit);
2016            assert_eq!(&got, p, "v1 row {} must read back raw", i + 1);
2017        }
2018        assert_eq!(t.pread_served(), 5, "a v1 pool must still take the extent fast path");
2019        // And it keeps working when this build appends to it: the appended
2020        // batch must be v1-shaped, because the stream's one schema message is
2021        // v1 — a six-column batch under it would not decode. Proven by reading
2022        // every batch back through Arrow's own stream reader afterwards.
2023        t.explode(&oid(9), GitObjectKind::Commit, &payloads[0]).unwrap();
2024        t.flush().unwrap();
2025        assert_eq!(t.content(&oid(9)).unwrap().unwrap().1, payloads[0]);
2026        assert_eq!(t.content(&oid(1)).unwrap().unwrap().1, payloads[0]);
2027        let rd = znippy_common::arrow::ipc::reader::StreamReader::try_new(
2028            std::fs::File::open(&path).unwrap(),
2029            None,
2030        )
2031        .unwrap();
2032        let mut rows = 0;
2033        for b in rd {
2034            let b = b.expect("every batch in a v1 file decodes under the v1 schema");
2035            assert!(enc_column(&b).is_none(), "a batch appended to a v1 pool must be v1-shaped");
2036            rows += b.num_rows();
2037        }
2038        assert_eq!(rows, 6, "five v1 rows and the one appended");
2039    }
2040
2041    #[test]
2042    fn a_payload_round_trips_through_the_table() {
2043        let t = ExplodedArchive::open_with_policy(&tmp("rt"), ExplodePolicy::Full).unwrap();
2044        t.explode(&oid(1), GitObjectKind::Commit, b"tree deadbeef\n")
2045            .unwrap();
2046        t.explode(&oid(2), GitObjectKind::Blob, b"hello world")
2047            .unwrap();
2048        t.flush().unwrap();
2049        assert_eq!(
2050            t.content(&oid(2)).unwrap().unwrap(),
2051            (GitObjectKind::Blob, b"hello world".to_vec())
2052        );
2053        assert_eq!(
2054            t.content(&oid(1)).unwrap().unwrap(),
2055            (GitObjectKind::Commit, b"tree deadbeef\n".to_vec())
2056        );
2057        assert!(t.has(&oid(1)).unwrap());
2058        assert!(!t.has(&oid(9)).unwrap());
2059    }
2060
2061    /// The row is a FILE: mode and path survive the round trip.
2062    #[test]
2063    fn a_row_can_carry_the_name_it_was_seen_at() {
2064        let t = ExplodedArchive::open_with_policy(&tmp("named"), ExplodePolicy::Full).unwrap();
2065        t.explode_at(
2066            &oid(1),
2067            GitObjectKind::Blob,
2068            Some(0o100755),
2069            Some("scripts/build.sh"),
2070            b"#!/bin/sh\n",
2071        )
2072        .unwrap();
2073        t.explode(&oid(2), GitObjectKind::Blob, b"anonymous").unwrap();
2074        t.flush().unwrap();
2075        assert_eq!(
2076            t.name(&oid(1)).unwrap().unwrap(),
2077            (Some(0o100755), Some("scripts/build.sh".to_owned()))
2078        );
2079        assert_eq!(
2080            t.name(&oid(2)).unwrap().unwrap(),
2081            (None, None),
2082            "null is 'not recorded', and must not read as a real name"
2083        );
2084    }
2085
2086    /// Reopening finds every row without reading a payload, and appends continue
2087    /// the SAME stream rather than starting a second one.
2088    #[test]
2089    fn the_table_survives_a_reopen_and_keeps_appending() {
2090        let p = tmp("reopen");
2091        {
2092            let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2093            t.explode(&oid(5), GitObjectKind::Tree, b"100644 f\0").unwrap();
2094            t.flush().unwrap();
2095        }
2096        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2097        assert_eq!(t.rows().unwrap(), 1, "row count comes from batch metadata");
2098        t.explode(&oid(6), GitObjectKind::Blob, b"second session").unwrap();
2099        t.flush().unwrap();
2100        assert_eq!(t.rows().unwrap(), 2);
2101        assert_eq!(
2102            t.content(&oid(5)).unwrap().unwrap(),
2103            (GitObjectKind::Tree, b"100644 f\0".to_vec())
2104        );
2105        assert_eq!(
2106            t.content(&oid(6)).unwrap().unwrap(),
2107            (GitObjectKind::Blob, b"second session".to_vec())
2108        );
2109
2110        // And a third open sees both — one schema message, two batches.
2111        let again = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2112        assert_eq!(again.rows().unwrap(), 2);
2113        assert_eq!(again.batches.lock().unwrap().len(), 2);
2114    }
2115
2116    /// A read must not miss a row that is buffered but not yet written — an
2117    /// "absent" that is really "not flushed" sends the caller off to re-resolve
2118    /// a whole pack for an object we are holding.
2119    #[test]
2120    fn an_unflushed_row_is_still_served() {
2121        let t = ExplodedArchive::open_with_policy(&tmp("unflushed"), ExplodePolicy::Full).unwrap();
2122        t.explode(&oid(4), GitObjectKind::Commit, b"buffered").unwrap();
2123        assert_eq!(
2124            t.content(&oid(4)).unwrap().unwrap(),
2125            (GitObjectKind::Commit, b"buffered".to_vec())
2126        );
2127        assert!(t.has(&oid(4)).unwrap());
2128    }
2129
2130    #[test]
2131    fn an_absent_oid_is_none_and_counted() {
2132        let t = ExplodedArchive::open_with_policy(&tmp("absent"), ExplodePolicy::Full).unwrap();
2133        assert!(t.content(&oid(3)).unwrap().is_none());
2134        assert_eq!(t.stats().absent, 1);
2135        assert_eq!(t.stats().served, 0);
2136    }
2137
2138    #[test]
2139    fn of_kind_selects_by_kind() {
2140        let t = ExplodedArchive::open_with_policy(&tmp("kind"), ExplodePolicy::Full).unwrap();
2141        t.explode(&oid(1), GitObjectKind::Commit, b"c1").unwrap();
2142        t.explode(&oid(2), GitObjectKind::Blob, b"bb").unwrap();
2143        t.explode(&oid(3), GitObjectKind::Commit, b"c2").unwrap();
2144        let commits = t.of_kind(GitObjectKind::Commit).unwrap();
2145        assert_eq!(commits.len(), 2);
2146        assert!(commits.iter().all(|(_, p)| p[0] == b'c'));
2147    }
2148
2149    /// ★ A RETIRED ROW STAYS RETIRED ACROSS A REOPEN.
2150    ///
2151    /// The file is append-only, so `retain` writes a [`TOMBSTONE`] rather than
2152    /// deleting. Seen RED with an index-only `retain`: the reopen below rebuilds
2153    /// from the file and brought the dead row back — which is
2154    /// `store::tests::a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened`
2155    /// failing with "551 commits, not the one live one", reached from here in
2156    /// three lines instead of a whole gc.
2157    #[test]
2158    fn retain_drops_only_what_is_dead_and_the_drop_survives_a_reopen() {
2159        let p = tmp("retain");
2160        {
2161            let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2162            t.explode(&oid(1), GitObjectKind::Blob, b"a").unwrap();
2163            t.explode(&oid(2), GitObjectKind::Blob, b"b").unwrap();
2164            t.flush().unwrap();
2165            assert_eq!(t.retain(&|o: &[u8]| o[0] == 1).unwrap(), 1);
2166            assert!(t.has(&oid(1)).unwrap());
2167            assert!(!t.has(&oid(2)).unwrap());
2168            assert_eq!(
2169                t.of_kind(GitObjectKind::Blob).unwrap().len(),
2170                1,
2171                "a scan must not resurrect what retain dropped"
2172            );
2173        }
2174        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2175        assert!(t.has(&oid(1)).unwrap(), "the live row did not survive");
2176        assert!(
2177            !t.has(&oid(2)).unwrap(),
2178            "a reopen resurrected a retired row — the tombstone is not durable"
2179        );
2180        assert!(t.content(&oid(2)).unwrap().is_none());
2181        assert_eq!(t.of_kind(GitObjectKind::Blob).unwrap().len(), 1);
2182    }
2183
2184    /// sha256 keys are not truncated to sha1 width, and a table refuses a width
2185    /// it cannot hold.
2186    #[test]
2187    fn a_sha256_oid_keeps_all_thirty_two_bytes() {
2188        let p = tmp("sha256");
2189        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2190        let o = [3u8; 32];
2191        t.explode(&o, GitObjectKind::Tree, b"t").unwrap();
2192        t.flush().unwrap();
2193        assert!(t.has(&o).unwrap());
2194        assert_eq!(t.content(&o).unwrap().unwrap().1, b"t");
2195
2196        t.explode(&oid(1), GitObjectKind::Blob, b"narrow").unwrap();
2197        let err = t.flush().expect_err("a 20-byte oid cannot join a 32-byte table");
2198        assert!(format!("{err}").contains("32-byte oids"), "{err}");
2199    }
2200
2201    /// A torn tail is truncated away rather than left mid-stream, where every
2202    /// later batch would be unreachable.
2203    #[test]
2204    fn a_torn_tail_is_dropped_and_the_table_keeps_working() {
2205        let p = tmp("torn");
2206        {
2207            let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2208            t.explode(&oid(1), GitObjectKind::Blob, b"complete").unwrap();
2209            t.flush().unwrap();
2210        }
2211        // Half a message, as a killed process leaves.
2212        {
2213            let mut f = OpenOptions::new().append(true).open(&p).unwrap();
2214            f.write_all(&[0xFF, 0xFF, 0xFF, 0xFF, 0x40, 0x00, 0x00, 0x00, 1, 2, 3])
2215                .unwrap();
2216        }
2217        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2218        assert_eq!(t.rows().unwrap(), 1, "the torn message contributes nothing");
2219        t.explode(&oid(2), GitObjectKind::Blob, b"after").unwrap();
2220        t.flush().unwrap();
2221        assert_eq!(t.rows().unwrap(), 2);
2222        assert_eq!(
2223            t.content(&oid(2)).unwrap().unwrap().1,
2224            b"after".to_vec(),
2225            "the append landed at the last good boundary, not after the garbage"
2226        );
2227    }
2228
2229    #[test]
2230    fn graph_keeps_commits_and_trees_and_drops_blobs() {
2231        let p = ExplodePolicy::Graph;
2232        assert!(p.wants(GitObjectKind::Commit));
2233        assert!(p.wants(GitObjectKind::Tree));
2234        assert!(!p.wants(GitObjectKind::Blob));
2235        assert!(!ExplodePolicy::Off.wants(GitObjectKind::Commit));
2236        assert!(ExplodePolicy::Full.wants(GitObjectKind::Blob));
2237        assert!(ExplodePolicy::Commits.wants(GitObjectKind::Commit));
2238        assert!(!ExplodePolicy::Commits.wants(GitObjectKind::Tree));
2239        assert_eq!(
2240            ExplodePolicy::default(),
2241            ExplodePolicy::Commits,
2242            "the shipped default is the lane Q one: commits and tags eager, trees \
2243             and blobs lazy and budgeted. `Full` is what put 22.4x the pack into \
2244             /data on the rig; `Graph` is what put 28.24 GB of tree payloads there \
2245             and 50.9 GiB of them in RAM."
2246        );
2247    }
2248
2249    /// ★ THE SETTING: a declined object is not stored, and is COUNTED — so a
2250    /// table short by policy and one short by fault do not look alike.
2251    #[test]
2252    fn a_declined_object_is_not_stored_and_is_counted() {
2253        let t = ExplodedArchive::open_with_policy(&tmp("policy"), ExplodePolicy::Graph).unwrap();
2254        t.explode(&oid(1), GitObjectKind::Commit, b"c").unwrap();
2255        t.explode(&oid(2), GitObjectKind::Blob, b"bbbb").unwrap();
2256        t.flush().unwrap();
2257        assert!(t.has(&oid(1)).unwrap());
2258        assert!(!t.has(&oid(2)).unwrap());
2259        assert_eq!(t.skipped(), 1);
2260        assert_eq!(t.rows().unwrap(), 1);
2261    }
2262
2263    /// `Off` writes no file at all — the whole point of the free tier.
2264    #[test]
2265    fn off_stores_nothing_and_leaves_no_file() {
2266        let p = tmp("off");
2267        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Off).unwrap();
2268        t.explode(&oid(1), GitObjectKind::Commit, b"c").unwrap();
2269        t.explode(&oid(2), GitObjectKind::Blob, b"b").unwrap();
2270        assert_eq!(t.flush().unwrap(), 0);
2271        assert_eq!(t.rows().unwrap(), 0);
2272        assert_eq!(t.skipped(), 2);
2273        assert!(!p.exists(), "a disabled table must not create a file");
2274    }
2275
2276    // ════════════════════════════════════════════════════════════════════════
2277    // The stree in front of the lookup
2278    // ════════════════════════════════════════════════════════════════════════
2279
2280    /// `n` oids in `groups` of `per` that share their first EIGHT bytes and
2281    /// differ only after them — the collision is CONSTRUCTED, never hoped for.
2282    ///
2283    /// The prefixes deliberately straddle `0x80`: half the oids in a real repo
2284    /// start there, and they are the half a non-order-preserving key would sort
2285    /// onto the wrong side of the keyspace.
2286    fn colliding_oids(groups: usize, per: usize) -> Vec<[u8; 20]> {
2287        let mut out = Vec::with_capacity(groups * per);
2288        for g in 0..groups {
2289            let mut z = (g as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15);
2290            z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
2291            z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
2292            let prefix = (z ^ (z >> 31)).to_be_bytes();
2293            for k in 0..per {
2294                let mut o = [0u8; 20];
2295                o[..8].copy_from_slice(&prefix);
2296                o[8] = k as u8;
2297                o[9..13].copy_from_slice(&(g as u32).to_be_bytes());
2298                out.push(o);
2299            }
2300        }
2301        out
2302    }
2303
2304    /// ★ THE STREE RETURNS THE ROW THE BINARY SEARCH RETURNED — including on
2305    /// 8-byte prefix collisions, of which this table has 1000.
2306    ///
2307    /// Three answers are compared for every oid: the `stree` probe, the
2308    /// `binary_search` over the identical oid array that this file shipped with,
2309    /// and the **applied output** — the payload `content` hands back, which is
2310    /// unique per oid, so a wrong row is a wrong payload and not a passing
2311    /// assertion about a `Located` nobody read.
2312    ///
2313    /// Seen RED by deleting the full-oid comparison from `OidTree::verify` (i.e.
2314    /// `.find(|&i| self.oid_at(i) == oid)` → `.next()`, the "stree already found
2315    /// it" mistake): **"the stree and the binary search disagree at
2316    /// 0000000000000000010000000000000000000000: Some(Located { batch: 0, row: 0,
2317    /// kind: Blob }) vs Some(Located { batch: 0, row: 1, kind: Blob })"** — the
2318    /// probe was handed the row of the prefix-mate stored one slot earlier.
2319    #[test]
2320    fn the_stree_agrees_with_the_binary_search_it_replaced_on_colliding_prefixes() {
2321        let t = ExplodedArchive::open_with_policy(&tmp("stree-agree"), ExplodePolicy::Full).unwrap();
2322        let oids = colliding_oids(1000, 4);
2323        for (i, o) in oids.iter().enumerate() {
2324            t.explode(o, GitObjectKind::Blob, format!("payload-{i}").as_bytes())
2325                .unwrap();
2326        }
2327        t.flush().unwrap();
2328        // Force the index, then reach into it: the two finders must sit over the
2329        // SAME arrays, or this compares two indexes rather than two finders.
2330        t.index_now().unwrap();
2331        let idx = t.index.lock().unwrap();
2332        let tree = idx.as_ref().expect("index_now built it");
2333        assert_eq!(tree.len(), oids.len());
2334
2335        // The premise: the collisions are real in the built tree.
2336        let mut colliding_runs = 0usize;
2337        for o in &oids {
2338            let run = tree.candidate_run(key_for_oid(o));
2339            assert_eq!(
2340                run.len(),
2341                4,
2342                "expected a 4-entry candidate run for {}, got {run:?} — the collision \
2343                 the rest of this test rests on is not there",
2344                hex::encode(o)
2345            );
2346            colliding_runs += 1;
2347        }
2348        assert_eq!(colliding_runs, oids.len());
2349        // And the sign boundary is really crossed, or the key's top-bit flip is
2350        // untested by this corpus.
2351        assert!(
2352            oids.iter().any(|o| o[0] >= 0x80) && oids.iter().any(|o| o[0] < 0x80),
2353            "premise: the prefixes must straddle 0x80"
2354        );
2355
2356        for o in &oids {
2357            assert_eq!(
2358                tree.find(o),
2359                tree.find_by_binary_search(o),
2360                "the stree and the binary search disagree at {}: {:?} vs {:?}",
2361                hex::encode(o),
2362                tree.find(o),
2363                tree.find_by_binary_search(o)
2364            );
2365            assert!(tree.find(o).is_some(), "{} vanished", hex::encode(o));
2366        }
2367        // An oid on a colliding prefix that was never stored must MISS, in both
2368        // finders — this is the query a verify-less probe answers with somebody
2369        // else's row.
2370        for o in oids.iter().step_by(4) {
2371            let mut absent = *o;
2372            absent[8] = 0xff;
2373            assert_eq!(tree.find(&absent), None, "{} was never stored", hex::encode(absent));
2374            assert_eq!(tree.find_by_binary_search(&absent), None);
2375        }
2376        drop(idx);
2377
2378        // Applied output: each oid's own payload, not its prefix-mate's.
2379        for (i, o) in oids.iter().enumerate() {
2380            assert_eq!(
2381                t.content(o).unwrap().unwrap().1,
2382                format!("payload-{i}").into_bytes(),
2383                "{} came back with another object's bytes",
2384                hex::encode(o)
2385            );
2386        }
2387    }
2388
2389    /// ★ A TOMBSTONE STILL RETIRES AN OID WITH THE STREE IN FRONT — and its
2390    /// prefix-mates do not go with it, nor stand in for it.
2391    ///
2392    /// Two failure modes live here and both return a wrong answer rather than an
2393    /// error: a fold that keeps the *first* sighting rather than the last leaves
2394    /// the retired oid resolvable, and a probe that trusts the 8-byte key answers
2395    /// the retired oid with the row of the live oid it collides with.
2396    ///
2397    /// Seen RED three times:
2398    ///
2399    /// * `Keep::Last` → `Keep::First` in `index_now`: **"a reopen resurrected a
2400    ///   retired oid: 0000000000000000000000000000000000000000"**. The same break
2401    ///   also reds the older, coarser
2402    ///   `retain_drops_only_what_is_dead_and_the_drop_survives_a_reopen`.
2403    /// * the full-oid comparison dropped from `OidTree::verify`:
2404    ///   **"0000000000000000000000000000000000000000 was retired"** — the dead
2405    ///   oid answered `has` with its live prefix-mate's row, in-process, before
2406    ///   any reopen.
2407    /// * the incremental fold in `flush` no longer filtering the retired set
2408    ///   (`if !retired.contains(oid)` → `if true`): the same message, from the
2409    ///   live process rather than from a reopen.
2410    #[test]
2411    fn a_tombstoned_oid_stays_absent_when_a_live_oid_shares_its_prefix() {
2412        let p = tmp("stree-tomb");
2413        // Four oids per prefix: one is retired, three stay.
2414        let oids = colliding_oids(64, 4);
2415        let dead: Vec<[u8; 20]> = oids.iter().copied().step_by(4).collect();
2416        {
2417            let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2418            for (i, o) in oids.iter().enumerate() {
2419                t.explode(o, GitObjectKind::Blob, format!("live-{i}").as_bytes())
2420                    .unwrap();
2421            }
2422            t.flush().unwrap();
2423            let retired = t.retain(&|o: &[u8]| !dead.iter().any(|d| d == o)).unwrap();
2424            assert_eq!(retired, dead.len() as u64);
2425            for d in &dead {
2426                assert!(!t.has(d).unwrap(), "{} was retired", hex::encode(d));
2427                assert!(
2428                    t.content(d).unwrap().is_none(),
2429                    "the retired oid {} was answered with a prefix-mate's row",
2430                    hex::encode(d)
2431                );
2432            }
2433            for (i, o) in oids.iter().enumerate() {
2434                if i % 4 == 0 {
2435                    continue;
2436                }
2437                assert_eq!(
2438                    t.content(o).unwrap().unwrap().1,
2439                    format!("live-{i}").into_bytes(),
2440                    "{} is alive and must still answer with its OWN bytes",
2441                    hex::encode(o)
2442                );
2443            }
2444        }
2445        // ★ And a reopen — which rebuilds the tree from the file, tombstones and
2446        // all — finds every live row and none of the dead.
2447        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2448        for (i, o) in oids.iter().enumerate() {
2449            if i % 4 == 0 {
2450                assert!(
2451                    !t.has(o).unwrap(),
2452                    "a reopen resurrected a retired oid: {}",
2453                    hex::encode(o)
2454                );
2455            } else {
2456                assert_eq!(
2457                    t.content(o).unwrap().unwrap().1,
2458                    format!("live-{i}").into_bytes(),
2459                    "a reopen lost or misrouted the live oid {}",
2460                    hex::encode(o)
2461                );
2462            }
2463        }
2464        assert_eq!(
2465            t.of_kind(GitObjectKind::Blob).unwrap().len(),
2466            oids.len() - dead.len()
2467        );
2468        // Only now — `rows()` answers from batch metadata until an index exists,
2469        // and the file still holds the retired rows and their tombstones. This
2470        // assertion measured 320 against 192 when it was made before the reads
2471        // above, which is the documented behaviour of `rows()` and not the
2472        // tombstone leaking.
2473        assert_eq!(t.rows().unwrap(), (oids.len() - dead.len()) as u64);
2474    }
2475
2476    /// The incremental extension in `flush` and the from-scratch rebuild in
2477    /// `index_now` must land on the same tree — one of them runs while a push is
2478    /// hot and the other after a restart, and a difference between them is a
2479    /// bug that only appears to the second process.
2480    ///
2481    /// Seen RED twice, once from each side, and the two reds are mirror images —
2482    /// which is what tells you the test compares the folds rather than one of
2483    /// them against itself:
2484    ///
2485    /// * incremental path drops the `retired` filter over the entries it carries
2486    ///   forward (`if !retired.contains(oid)` → `if true`): **"the two folds kept
2487    ///   a different number of oids: left: 150, right: 200"** — the live index
2488    ///   kept the 50 it had been told to retire.
2489    /// * `Keep::Last` → `Keep::First` in `index_now`: **"left: 200, right: 150"**
2490    ///   — the rebuild kept them instead.
2491    #[test]
2492    fn the_incremental_index_and_a_rebuilt_one_agree() {
2493        let p = tmp("stree-incr");
2494        let oids = colliding_oids(50, 4);
2495        let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2496        // Build the index FIRST, so every flush below takes the incremental path.
2497        t.index_now().unwrap();
2498        for (i, o) in oids.iter().enumerate() {
2499            t.explode(o, GitObjectKind::Blob, format!("v-{i}").as_bytes())
2500                .unwrap();
2501            if i % 37 == 0 {
2502                t.flush().unwrap();
2503            }
2504        }
2505        t.flush().unwrap();
2506        t.retain(&|o: &[u8]| o[8] != 3).unwrap();
2507        let live: Vec<(Vec<u8>, Vec<u8>)> = {
2508            let idx = t.index.lock().unwrap();
2509            idx.as_ref()
2510                .unwrap()
2511                .iter()
2512                .map(|(o, l)| (o.to_vec(), vec![l.batch as u8, l.row as u8]))
2513                .collect()
2514        };
2515        drop(t);
2516
2517        let again = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
2518        again.index_now().unwrap();
2519        let rebuilt = again.index.lock().unwrap();
2520        let rebuilt = rebuilt.as_ref().unwrap();
2521        assert_eq!(rebuilt.len(), live.len(), "the two folds kept a different number of oids");
2522        for (o, _) in &live {
2523            assert!(
2524                rebuilt.find(o).is_some(),
2525                "the live process and a reopen disagree about {}: Some(..) vs None",
2526                hex::encode(o)
2527            );
2528        }
2529        for o in oids.iter().filter(|o| o[8] == 3) {
2530            assert!(rebuilt.find(o).is_none(), "{} was retired", hex::encode(o));
2531        }
2532    }
2533
2534    /// The measurement behind the change, re-takeable: three finders over the
2535    /// SAME 1 000 000 entries — the `Vec<(Vec<u8>, Located)>` this file shipped
2536    /// with, a binary search over the flat array, and the stree.
2537    ///
2538    /// `--ignored` because it is a measurement, not a guard: it asserts only that
2539    /// the three agree, and prints the times. Run it with
2540    /// `cargo test --release -p znippy-plugin-git --lib exploded_arrow -- --ignored --nocapture`
2541    /// — a debug build measures the borrow checker, not the machine.
2542    ///
2543    /// ## Measured, oden 2026-08-14, `--release`, 1 000 000 entries, 200 000
2544    /// probes (half misses), three runs
2545    ///
2546    /// | arm | ns/probe | vs shipped |
2547    /// |---|---:|---:|
2548    /// | `Vec<(Vec<u8>, Located)>` + `binary_search` (what shipped) | 836 / 862 / 961 | 1.00 |
2549    /// | flat oid array + `binary_search` | 339 / 339 / 348 | 0.40 |
2550    /// | **stree** | **164 / 175 / 185** | **0.21** |
2551    ///
2552    /// So the structure is worth **~2×** on its own and the flattening another
2553    /// **~2.5×**, and the two together take a point lookup to a fifth of what it
2554    /// cost. **The box was NOT quiet** — `/proc/loadavg` 1-min 29.7–31.5 on 32
2555    /// cores, other agents building throughout — so the absolute figures are
2556    /// upper bounds. The ratios are reported instead, and each arm's own spread
2557    /// across the three runs (14% / 2.7% / 12%) is far narrower than the 5×
2558    /// and 2× between them.
2559    #[test]
2560    #[ignore = "measurement; run under --release with --ignored"]
2561    fn lookup_cost_before_and_after() {
2562        use std::time::Instant;
2563        let n = 1_000_000usize;
2564        // 4 oids per prefix, so ~25% of probes land in a run of four and the
2565        // verify step is exercised rather than skipped.
2566        let oids = colliding_oids(n / 4, 4);
2567        let mut b = OidTreeBuilder::with_capacity(oids.len());
2568        for (i, o) in oids.iter().enumerate() {
2569            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, enc: ENC_RAW }))
2570                .unwrap();
2571        }
2572        let build = Instant::now();
2573        let tree = b.finish(Keep::Last);
2574        let build = build.elapsed();
2575
2576        // The shape this replaced: one heap allocation per oid.
2577        let mut old: Vec<(Vec<u8>, Located)> =
2578            tree.iter().map(|(o, l)| (o.to_vec(), l)).collect();
2579        old.sort_by(|a, b| a.0.cmp(&b.0));
2580
2581        // A miss-heavy probe order, and NOT in oid order — a `have` negotiation
2582        // is random and half of it misses.
2583        let mut probes: Vec<[u8; 20]> = Vec::with_capacity(200_000);
2584        for i in (0..oids.len()).step_by(oids.len() / 100_000) {
2585            probes.push(oids[i]);
2586            let mut m = oids[i];
2587            m[8] = 0xfe;
2588            probes.push(m);
2589        }
2590
2591        let mut sink = 0u64;
2592        let t0 = Instant::now();
2593        for p in &probes {
2594            if let Ok(i) = old.binary_search_by(|(o, _)| o.as_slice().cmp(&p[..])) {
2595                sink += old[i].1.row as u64;
2596            }
2597        }
2598        let vec_of_vec = t0.elapsed();
2599        let t0 = Instant::now();
2600        for p in &probes {
2601            if let Some(l) = tree.find_by_binary_search(p) {
2602                sink += l.row as u64;
2603            }
2604        }
2605        let flat_bsearch = t0.elapsed();
2606        let t0 = Instant::now();
2607        for p in &probes {
2608            if let Some(l) = tree.find(p) {
2609                sink += l.row as u64;
2610            }
2611        }
2612        let stree = t0.elapsed();
2613
2614        // The arms must be the same question.
2615        for p in &probes {
2616            let want = tree.find_by_binary_search(p);
2617            assert_eq!(tree.find(p), want, "arms disagree at {}", hex::encode(p));
2618        }
2619        let per = |d: std::time::Duration| d.as_secs_f64() * 1e9 / probes.len() as f64;
2620        println!(
2621            "exploded oid lookup, {} entries ({} groups of 4 colliding prefixes), \
2622             {} probes half of them misses:\n  \
2623             Vec<(Vec<u8>,Located)> binary_search  {:>7.1} ns/probe\n  \
2624             flat array binary_search             {:>7.1} ns/probe\n  \
2625             stree                                {:>7.1} ns/probe\n  \
2626             tree build (sort + fold + stree)      {:.3} s   sink={sink}",
2627            tree.len(),
2628            oids.len() / 4,
2629            probes.len(),
2630            per(vec_of_vec),
2631            per(flat_bsearch),
2632            per(stree),
2633            build.as_secs_f64(),
2634        );
2635    }
2636
2637    #[test]
2638    fn an_unknown_policy_is_refused() {
2639        assert_eq!(ExplodePolicy::parse("full").unwrap(), ExplodePolicy::Full);
2640        let err = ExplodePolicy::parse("ful").expect_err("a typo must be refused");
2641        assert!(format!("{err}").contains("unknown explode policy"), "{err}");
2642    }
2643}
2644