plugins/native/znippy-plugin-git/src/exploded_arrow.rs
//! **The exploded objects table: ONE Arrow IPC file, one row per object.**
//!
//! # What this replaces, and what it measured
//!
//! [`crate::exploded`] holds `oid → [kind][payload…]` in **redb**. It works, and
//! it is the wrong medium — MEASURED on `linux.git`, oden 2026-08-13,
//! 11 697 976 objects:
//!
//! | | |
//! |---|---:|
//! | pack as pushed (verbatim, deflated + delta-encoded) | 6.4 GB |
//! | its objects fully resolved (`git verify-pack -v`, summed) | **16.8 GB** |
//! | `objects.exploded` on disk | **204 GB** |
//!
//! So the 2026-08-08 eager decision was RIGHT — resolved content is **2.6×** the
//! pack, comfortably inside the 3× it budgeted — and the medium was wrong by
//! **12×**. redb is a copy-on-write B-tree: 11.7 million small keys committed in
//! [`FLUSH_BYTES`] batches rewrite interior pages over and over, and 187 of
//! those 204 GB are page churn rather than anybody's data.
//!
//! It cost throughput for the same reason. [`crate::indexer`] fans a drained
//! batch across N workers with `gatling_for_each` (LAW 3, never rayon) and every
//! one of them then queued behind redb's single write transaction — MEASURED:
//! **96.8% of ONE core** with 31 idle while the exploder ran.
//!
//! # The shape
//!
//! One table, one file, the payload IN the row — which is what this engine
//! already does everywhere else ([`crate::reach`] stores a bitmap as
//! `DataType::Binary`, [`crate::secrets`] a ciphertext):
//!
//! ```text
//! oid | object_type | mode | path | payload
//! ```
//!
//! An Arrow IPC stream is `[schema message][batch message]…` — append-only, no
//! pages to rewrite, so the file is the payload total plus framing. Each flush
//! encodes one batch with [`IpcDataGenerator`] and appends it with
//! [`write_message`]; nothing earlier in the file is ever touched again.
//!
//! `payload` is `LargeBinary` and not `Binary` deliberately: `Binary` offsets are
//! `i32`, so one batch would cap at 2 GiB of payload and overflow *silently* on a
//! corpus nobody tested. The wider offset costs four bytes a row.
//!
//! ## `mode` and `path` — the row IS a file
//!
//! A git object on its own is content without a name. The two nullable columns
//! carry the name when the caller knows it (a tree walk does; a bare pack
//! resolve does not), which is what lets this table be read as a filesystem
//! rather than as a content-addressed bag. They are **nullable and best-effort**
//! for a reason that is inherent and not a shortcut: one blob is reachable at
//! many paths in many commits, so a path column can only ever record *a* path —
//! the one it was first exploded at. Null means "not known here", never "at the
//! root".
//!
//! # Reopening is cheap, and the index is lazy
//!
//! A batch's *metadata* carries its row count and its body length, so
//! [`ExplodedArchive::open`] walks the message framing — seeking over the bodies,
//! reading kilobytes — and knows how many rows the table holds and where every
//! batch begins without touching a gigabyte. That is what `adopt_journal` needs
//! on startup.
//!
//! The `oid → row` index costs one sequential pass and is therefore built on the
//! **first lookup**, not on open. A push-heavy process never pays for it; a
//! process that reads pays once. After that a point read is two `pread`s — the
//! schema message and the one batch — decoded through [`StreamDecoder`], so it
//! never scans the batches in front of the one it wants.
//!
//! # The lookup is ragnar's static tree — see [`OidTree`]
//!
//! That index used to be a `Vec<(Vec<u8>, Located)>` walked with
//! `binary_search_by`: one heap allocation per oid — **11.7 million** of them on
//! `linux.git` — and a `log₂ n` chain of dependent loads over scattered `Vec`
//! headers. It is now the same `stree` (`znippy-zoomies`) this crate already puts
//! over oids in [`crate::oid_index`], over flat parallel arrays.
//! `nornir-workspace.toml`'s performance law names that structure by name; this
//! is the second place in this crate that obeys it, and it obeys it by reusing
//! [`crate::oid_index::key_for_oid`] rather than deriving a second key.
//!
//! # Droppable, and it is one `rm`
//!
//! Every row is re-derivable from the verbatim pack bytes, so the file can be
//! deleted without consulting a client:
//! [`crate::git_ops::Absorber::adopt_journal`] compares this table's row count
//! against the `objects` table's and re-queues every pack if it is short. Absent
//! means fall back and rebuild; it never means wrong. One file, one delete —
//! which is the property the engine's own tests assert.
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use anyhow::{Context, Result};
use znippy_common::arrow::array::{
Array, FixedSizeBinaryArray, FixedSizeBinaryBuilder, LargeBinaryArray, LargeBinaryBuilder,
StringArray, StringBuilder, UInt8Array, UInt8Builder, UInt32Array, UInt32Builder,
};
use znippy_common::arrow::buffer::Buffer;
use znippy_common::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use znippy_common::arrow::ipc::reader::StreamDecoder;
use znippy_common::arrow::ipc::writer::{
DictionaryTracker, IpcDataGenerator, IpcWriteOptions, write_message,
};
use znippy_common::arrow::ipc::root_as_message;
use znippy_common::arrow::record_batch::RecordBatch;
use znippy_zoomies::stree::STree64Mmap;
use crate::object::GitObjectKind;
use crate::oid_index::key_for_oid;
pub const COL_OID: &str = "oid";
pub const COL_TYPE: &str = "object_type";
pub const COL_MODE: &str = "mode";
pub const COL_PATH: &str = "path";
pub const COL_PAYLOAD: &str = "payload";
/// Buffered payload bytes that force a flush.
///
/// Not a memory *limit* — one object can exceed it on its own and is written
/// anyway. It is what keeps a 2 GiB push from holding its whole resolved content
/// in RAM before the first batch reaches the file.
pub const FLUSH_BYTES: usize = 64 << 20;
/// `oid | object_type | mode | path | payload`.
///
/// `oid_len` is a parameter because this engine serves both sha1 (20 bytes) and
/// sha256 (32); a hardcoded 20 would silently truncate every sha256 key.
/// **How each row's payload is stored** — `ENC_RAW` or `ENC_ZLIB`. Lever 3
/// (2026-09-04): the eager pool was 1.67 GB of inflated commits and trees, 61 %
/// of everything the store kept above the pack. Arrow IPC's own body
/// compression is per BUFFER, so a point read could not lift one row out of a
/// compressed batch and the `pread` extent fast path — the whole reason the
/// pool is fast — would have died with it (`payload_data_offset` refuses
/// compressed batches for exactly that reason; MEASURED 200 000× read
/// amplification without the extent). So the compression is per ROW: the
/// extent pread returns one compressed payload and inflates just that one.
/// zlib, because the crate already has it (`flate2`, zlib-rs — pure Rust), and
/// because it is the codec git itself uses per object, so the ratios are known
/// ground.
pub const COL_ENC: &str = "enc";
pub const ENC_RAW: u8 = 0;
pub const ENC_ZLIB: u8 = 1;
/// Rows shorter than this are never compressed: a zlib header and trailer are
/// 8 bytes, and a tiny tree does not earn them back. Set from
/// `examples/explode_compress_ratio.rs` over the kernel mirror's pool.
pub const MIN_ROW_COMPRESS: usize = 128;
/// A compressed row must be at least this much smaller than the raw one to be
/// kept compressed, or the inflate on every read buys nothing.
const COMPRESS_KEEP_IF_UNDER_PERCENT: usize = 90;
/// Schema v2 (2026-09-04): `enc` sits BEFORE `payload`, so the payload's data
/// buffer stays the LAST buffer of the batch — which is what
/// [`payload_data_offset`] reads the extent off. A v1 file has no `enc` column;
/// every reader treats a missing column as [`ENC_RAW`], so pools written before
/// this open unchanged and are never rewritten for it.
pub fn exploded_schema(oid_len: usize) -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new(COL_OID, DataType::FixedSizeBinary(oid_len as i32), false),
Field::new(COL_TYPE, DataType::UInt8, false),
Field::new(COL_MODE, DataType::UInt32, true),
Field::new(COL_PATH, DataType::Utf8, true),
Field::new(COL_ENC, DataType::UInt8, false),
Field::new(COL_PAYLOAD, DataType::LargeBinary, false),
]))
}
/// Schema v1 — what every pool written before 2026-09-04 carries, and what this
/// build keeps appending to such a pool (see `ExplodedArchive::legacy_v1`).
pub fn exploded_schema_v1(oid_len: usize) -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new(COL_OID, DataType::FixedSizeBinary(oid_len as i32), false),
Field::new(COL_TYPE, DataType::UInt8, false),
Field::new(COL_MODE, DataType::UInt32, true),
Field::new(COL_PATH, DataType::Utf8, true),
Field::new(COL_PAYLOAD, DataType::LargeBinary, false),
]))
}
/// Compress one row's payload for storage, or `None` when it is not worth it
/// (too short, or zlib did not get it under the keep threshold).
///
/// **`default` (level 6), not `fast`, and that is a measurement**
/// (`examples/explode_compress_ratio.rs`, the kernel mirror's eager pool,
/// 1 722 249 rows / 1 609 688 036 B of commits and tags, 2026-09-04):
///
/// | level | pool after | ratio | saved |
/// |---|---:|---:|---:|
/// | fast | 1 138 092 314 B | 1.41× | 471 595 722 B |
/// | **default** | **889 639 690 B** | **1.81×** | **720 048 346 B** |
///
/// Inflate cost is the SAME at both — 3.07 µs/row mean over 20 000 sampled
/// rows — because zlib decode speed does not depend on the level. So the read
/// path pays +3 µs on a point read (first-read p50 was 3.9 µs) whichever level
/// is chosen, and the only thing the level decides is how much disk that buys:
/// 44.7 % of the pool at `default` against 29.3 % at `fast`. The write pays
/// more CPU at `default`, on the indexer's threads, once per object.
fn compress_row(raw: &[u8]) -> Option<Vec<u8>> {
use std::io::Write as _;
if raw.len() < MIN_ROW_COMPRESS {
return None;
}
let mut e = flate2::write::ZlibEncoder::new(
Vec::with_capacity(raw.len() / 2 + 16),
flate2::Compression::default(),
);
e.write_all(raw).ok()?;
let z = e.finish().ok()?;
(z.len() * 100 < raw.len() * COMPRESS_KEEP_IF_UNDER_PERCENT).then_some(z)
}
/// The row's payload as the caller wants it — inflated if it was stored
/// compressed. `bytes` is exactly what the column (or the extent pread) held.
fn row_payload(enc: u8, bytes: &[u8]) -> Result<Vec<u8>> {
use std::io::Read as _;
match enc {
ENC_RAW => Ok(bytes.to_vec()),
ENC_ZLIB => {
let mut out = Vec::with_capacity(bytes.len() * 3);
flate2::read::ZlibDecoder::new(bytes)
.read_to_end(&mut out)
.context("inflating a compressed exploded row")?;
Ok(out)
}
other => anyhow::bail!("exploded row carries unknown encoding {other}"),
}
}
/// The `enc` column of a decoded batch, or `None` on a v1 batch (every row raw).
fn enc_column(batch: &RecordBatch) -> Option<&UInt8Array> {
batch
.column_by_name(COL_ENC)
.and_then(|c| c.as_any().downcast_ref::<UInt8Array>())
}
#[inline]
fn enc_at(enc: Option<&UInt8Array>, row: usize) -> u8 {
enc.map_or(ENC_RAW, |e| e.value(row))
}
/// The pack type codes, the same mapping [`crate::exploded`] uses, so the two
/// media cannot disagree about what a `2` means.
pub fn kind_code(k: GitObjectKind) -> u8 {
match k {
GitObjectKind::Commit => 1,
GitObjectKind::Tree => 2,
GitObjectKind::Blob => 3,
GitObjectKind::Tag => 4,
}
}
pub fn kind_of(code: u8) -> Option<GitObjectKind> {
match code {
1 => Some(GitObjectKind::Commit),
2 => Some(GitObjectKind::Tree),
3 => Some(GitObjectKind::Blob),
4 => Some(GitObjectKind::Tag),
_ => None,
}
}
/// **A row that retires an oid.**
///
/// The file is append-only, so `retain` cannot reach back and delete: it appends
/// rows that say *this oid is gone*, and the index fold applies them in file
/// order, so the last word about an oid wins. Dropping only from the in-memory
/// index is what a first cut did, and it was WRONG in the way that matters —
/// `a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened` came back
/// with **551 commits instead of 1**, because a reopen rebuilds the index from
/// the file and resurrected every row `gc` had retired.
///
/// `0` and not a sixth column: git's pack type codes are 1–7 and 0 is not one of
/// them, so [`kind_of`] already answers `None` for it. A tombstone costs an oid
/// and an empty payload, not a byte on every live row.
pub const TOMBSTONE: u8 = 0;
/// One buffered row, waiting for the next batch. `code` rather than a
/// [`GitObjectKind`] because a tombstone is a row with no kind.
#[derive(Debug, Clone)]
struct Pending {
oid: Vec<u8>,
code: u8,
mode: Option<u32>,
path: Option<String>,
/// Where this row's bytes are in [`PendingBatch::arena`]. **Not a
/// `Vec<u8>`** — see [`PendingBatch`].
pay: std::ops::Range<usize>,
}
/// The rows buffered for the next batch, and **one arena holding their bytes**.
///
/// # Why the bytes are not in the row
///
/// `payload: Vec<u8>` cost one heap allocation and one `memcpy` of the whole
/// object **per buffered row**, and on the lazy write-back path that allocation
/// is a second copy of bytes the reader is holding anyway. MEASURED by lane AA,
/// 2026-08-30: reading one 3 145 728 B object through `gunnar-cli` allocated
/// 6 340 693 B — *exactly two copies* — and the second was this one.
///
/// An arena makes the write an `extend_from_slice` into a buffer that grows
/// geometrically and is reused after every flush, so the bytes attributable to
/// one admitted read amortise to zero and a **declined** read costs nothing at
/// all (the admission gate returns before this is reached). LAW: zero copy.
#[derive(Default)]
struct PendingBatch {
rows: Vec<Pending>,
arena: Vec<u8>,
}
impl PendingBatch {
fn is_empty(&self) -> bool {
self.rows.is_empty()
}
/// Buffer one row's bytes and return the range they landed in.
fn stash(&mut self, payload: &[u8]) -> std::ops::Range<usize> {
let start = self.arena.len();
self.arena.extend_from_slice(payload);
start..self.arena.len()
}
fn bytes(&self, r: &Pending) -> &[u8] {
&self.arena[r.pay.clone()]
}
/// Hand the batch over for encoding and keep the arena's capacity: the next
/// batch reuses the same allocation, which is what makes the per-read cost
/// amortise.
fn take(&mut self) -> (Vec<Pending>, Vec<u8>) {
let rows = std::mem::take(&mut self.rows);
let arena = std::mem::replace(&mut self.arena, Vec::with_capacity(0));
(rows, arena)
}
}
/// Where one batch message sits in the file. Recovered from the framing walk,
/// which reads metadata and seeks over bodies.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Batch {
/// Byte offset of the message's continuation marker.
at: u64,
/// Bytes the whole message occupies: `8 + metadata + body`.
len: u64,
/// Rows in it — read from the metadata, so counting them costs no body read.
rows: u32,
/// Absolute byte offset of the message BODY — where the buffer offsets the
/// metadata declares are measured from.
body: u64,
/// The payload DATA buffer's offset within the body, from the metadata's
/// own buffer table. [`EXTENT_UNAVAILABLE`] when the body is compressed or
/// the buffer layout is not the one [`exploded_schema`] writes — a point
/// read then falls back to decoding the batch, which is slow and correct.
pay_data: u64,
}
/// Sentinel for "no byte extent could be computed for this row/batch" — the
/// fall-back-and-decode marker, never an offset a file could really have.
const EXTENT_UNAVAILABLE: u64 = u64::MAX;
/// The payload DATA buffer's offset within the message body, read from the
/// batch metadata alone.
///
/// The payload is [`exploded_schema`]'s LAST field and a `LargeBinary`'s data
/// buffer is its last buffer, so the table's final buffer entry IS the payload
/// bytes — asserted by also counting the buffers (12: validity+data for oid and
/// object_type, validity+data for mode, validity+offsets+data for path and for
/// payload). `None` when the count disagrees or the body is compressed;
/// [`ExplodedArchive::content`] then decodes the batch instead of guessing.
fn payload_data_offset(meta: &[u8]) -> Option<u64> {
let msg = root_as_message(meta).ok()?;
let rb = msg.header_as_record_batch()?;
if rb.compression().is_some() {
return None;
}
let bufs = rb.buffers()?;
// 12 buffers is schema v1 (oid 2, type 2, mode 2, path 3, payload 3); 14 is
// v2 with the `enc` column's two. In both the payload's data buffer is the
// last one, which is the only fact used below.
if bufs.len() != 12 && bufs.len() != 14 {
return None;
}
let last = bufs.get(bufs.len() - 1);
(last.offset() >= 0).then(|| last.offset() as u64)
}
/// Where one object's row is — and where its payload BYTES are.
///
/// The extent is what makes a point read one small `pread` instead of a whole
/// 64 MiB batch decode. MEASURED without it, rust-lang/rust push, 2026-08-19:
/// the pre-ack connectivity walk visited commits and trees in graph order —
/// effectively random across ~574 batches against a 4-slot cache — and decoded
/// 6.7 GB/s of page cache to serve ~98 objects/s, one core pinned for hours, a
/// ~200 000× read amplification. The extent costs 16 bytes per live oid.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Located {
batch: u32,
row: u32,
kind: GitObjectKind,
/// Absolute file offset of this row's payload bytes, or
/// [`EXTENT_UNAVAILABLE`] — then the read falls back to the batch decode.
pay_at: u64,
/// Payload length in bytes. Meaningless when `pay_at` is the sentinel.
pay_len: u64,
/// [`ENC_RAW`] or [`ENC_ZLIB`] — carried in the index so the extent fast
/// path knows whether to inflate without decoding the batch.
enc: u8,
}
/// **`oid → Located`, as ragnar's static tree over flat arrays.**
///
/// Three parallel arrays in one oid-ascending order, plus an [`STree64Mmap`]
/// built over the first of them:
///
/// ```text
/// keys count * 8 i64 little-endian, ascending — the stree keyspace
/// oids count * oid_len the FULL oid, for the verify step
/// locs count batch, row, kind
/// ```
///
/// ## Why this and not the `Vec<(Vec<u8>, Located)>` it replaces
///
/// That vector cost one heap allocation *per oid* — 11.7 M of them for
/// `linux.git` — and its `binary_search_by` chased `log₂ n` dependent loads over
/// `Vec` headers scattered across the heap before it could compare a byte. Here
/// the oid bytes are contiguous, and `stree` routes through 64-byte nodes with a
/// branchless AVX2 compare. `nornir-workspace.toml` names this structure as the
/// one to use, and [`crate::oid_index`] already uses it over exactly these keys —
/// so the key function is **imported from there**, not written again.
///
/// ## The 8-byte key is a FILTER, not a key — and this is where a wrong answer
/// would come from
///
/// The key is [`key_for_oid`]: the first eight bytes of the oid, big-endian, top
/// bit flipped so unsigned byte order becomes signed `i64` order (without the
/// flip every oid starting `0x80..0xff` — half of them — sorts below every oid
/// starting `0x00..0x7f`, and the keyspace `stree` requires to be ascending is
/// not). **Twelve** of a sha1 oid's twenty bytes are not in the key at all (24 of
/// 32 for sha256), so two distinct objects **can** share one key.
///
/// How likely, honestly: for `linux.git`'s 11.7 M objects, birthday arithmetic
/// gives n²/2 · 2⁻⁶⁴ ≈ **3.7 × 10⁻⁶** — one table in ~270 000. So it will not
/// happen by accident on one repository, and that is exactly why a verify-less
/// probe would ship and pass every test anyone bothered to write. It does not
/// stay accidental: finding *some* pair of contents whose oids share eight bytes
/// is a 2³² birthday search — hours of commodity hashing, not a research result —
/// and both objects are then perfectly ordinary blobs a client may push. (Hitting
/// a *specific* stored oid's prefix is 2⁶⁴ and is not a threat.) A hosted forge
/// serves oids that other people chose, so this is a property to hold by
/// construction rather than a probability to accept.
///
/// So `stree` narrows and never decides. It routes to *a* member of the run of
/// equal keys — which member is its own business, because its leaf scan starts at
/// a block boundary that can fall in the middle of the run — so
/// [`expand_run`](Self::expand_run) walks **both** directions to the run's ends,
/// and [`find`](Self::find) then compares the **whole oid** against every
/// candidate. Skipping that comparison would not merely be sloppy: a `have`
/// negotiation for an oid this table has never seen, whose first eight bytes
/// happen to match a stored one, would be answered with **the other object's
/// bytes** — a silently wrong object, which is far worse than a slow one.
struct OidTree {
/// Width of one oid. `0` only when the tree is empty.
oid_len: usize,
count: usize,
keys: Vec<u8>,
oids: Vec<u8>,
locs: Vec<Located>,
/// `None` for an empty table — `STree64Mmap::new_with_stride` asserts
/// `count > 0`.
tree: Option<STree64Mmap>,
}
/// Which member of a run of equal oids survives the fold — the two callers
/// genuinely differ, and the difference is not cosmetic.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Keep {
/// [`ExplodedArchive::flush`]'s incremental extension: an oid already in the
/// index keeps the [`Located`] it already had. That is what
/// [`ExplodedArchive::explode_at`] documents — a blob is reachable at many
/// paths and the table records the **first** name it was seen at.
First,
/// [`ExplodedArchive::index_now`]'s rebuild from the file: batches are walked
/// in file order and rows in row order, so the **last** word about an oid
/// wins. That is what makes a [`TOMBSTONE`] durable rather than advisory.
Last,
}
/// Accumulates sightings in the order they happen, then folds them.
///
/// Sightings, not entries: the same oid may arrive many times, and a `None`
/// [`Located`] is a [`TOMBSTONE`] row saying the oid is gone. Nothing is
/// deduplicated until [`finish`](Self::finish), because which sighting wins
/// depends on their order and on [`Keep`].
struct OidTreeBuilder {
/// Taken from the first oid pushed; every later one must match it.
oid_len: usize,
oids: Vec<u8>,
locs: Vec<Option<Located>>,
}
impl OidTreeBuilder {
fn with_capacity(rows: usize) -> Self {
Self { oid_len: 0, oids: Vec::new(), locs: Vec::with_capacity(rows) }
}
/// Record one sighting. `loc` is `None` for a tombstone row.
///
/// A width that disagrees with the rows already pushed is refused rather
/// than padded: a truncated or extended oid would sort and compare as a
/// different object.
fn push(&mut self, oid: &[u8], loc: Option<Located>) -> Result<()> {
if self.locs.is_empty() && self.oids.is_empty() {
self.oid_len = oid.len();
self.oids.reserve(self.locs.capacity() * self.oid_len);
}
anyhow::ensure!(
oid.len() == self.oid_len,
"the exploded index holds {}-byte oids and a {}-byte one arrived",
self.oid_len,
oid.len()
);
self.oids.extend_from_slice(oid);
self.locs.push(loc);
Ok(())
}
fn oid_at(&self, i: usize) -> &[u8] {
&self.oids[i * self.oid_len..(i + 1) * self.oid_len]
}
/// Sort, fold each run of equal oids down to its winner, drop the ones whose
/// winner is a tombstone, and build the tree.
fn finish(self, keep: Keep) -> OidTree {
let n = self.locs.len();
let mut order: Vec<u32> = (0..n as u32).collect();
// STABLE, and load-bearing: `Keep` is about *insertion* order within a
// run of equal oids, so an unstable sort would make "first sighting"
// and "last word wins" both mean "whichever one the sort happened to
// leave there".
order.sort_by(|&a, &b| self.oid_at(a as usize).cmp(self.oid_at(b as usize)));
let mut keys: Vec<u8> = Vec::with_capacity(n * 8);
let mut oids: Vec<u8> = Vec::with_capacity(n * self.oid_len);
let mut locs: Vec<Located> = Vec::with_capacity(n);
let mut i = 0usize;
while i < n {
let oid = self.oid_at(order[i] as usize);
let mut j = i + 1;
while j < n && self.oid_at(order[j] as usize) == oid {
j += 1;
}
let pick = match keep {
Keep::First => order[i],
Keep::Last => order[j - 1],
} as usize;
if let Some(loc) = self.locs[pick] {
// Ascending oid order gives ascending keys, because
// `key_for_oid` is order-preserving over the oid's leading
// bytes — the property `stree` needs and the reason for the
// sign-bit flip.
keys.extend_from_slice(&key_for_oid(oid).to_le_bytes());
oids.extend_from_slice(oid);
locs.push(loc);
}
i = j;
}
let count = locs.len();
let tree = (count > 0).then(|| STree64Mmap::new_with_stride(&keys, count, 8));
OidTree { oid_len: self.oid_len, count, keys, oids, locs, tree }
}
}
impl OidTree {
fn len(&self) -> usize {
self.count
}
fn oid_at(&self, i: usize) -> &[u8] {
&self.oids[i * self.oid_len..(i + 1) * self.oid_len]
}
fn key_at(&self, i: usize) -> i64 {
i64::from_le_bytes(self.keys[i * 8..i * 8 + 8].try_into().unwrap())
}
/// Widen a hit to the whole run of equal keys. `stree` routes to *a* member
/// of the run — its leaf scan starts at a block boundary, which can fall
/// anywhere inside one — so both directions are walked rather than assumed.
fn expand_run(&self, pos: usize, key: i64) -> std::ops::Range<usize> {
let mut lo = pos;
while lo > 0 && self.key_at(lo - 1) == key {
lo -= 1;
}
let mut hi = pos + 1;
while hi < self.count && self.key_at(hi) == key {
hi += 1;
}
lo..hi
}
/// Resolve an oid. `None` when absent — never another object's row.
fn find(&self, oid: &[u8]) -> Option<Located> {
if oid.len() != self.oid_len {
return None;
}
let tree = self.tree.as_ref()?;
let key = key_for_oid(oid);
let pos = tree.find_exact(key, &self.keys)?;
self.verify(pos, key, oid)
}
/// The full-oid comparison the whole structure's correctness rests on.
fn verify(&self, pos: usize, key: i64, oid: &[u8]) -> Option<Located> {
self.expand_run(pos, key)
.find(|&i| self.oid_at(i) == oid)
.map(|i| self.locs[i])
}
/// Every live entry, in ascending oid order — what `of_kind` and `retain`
/// fold over.
fn iter(&self) -> impl Iterator<Item = (&[u8], Located)> + '_ {
(0..self.count).map(move |i| (self.oid_at(i), self.locs[i]))
}
/// The **baseline** the tree has to agree with: `binary_search` over the very
/// same oid array, followed by the very same run-and-verify. It is the
/// finder this file shipped with, and it exists so "the stree returns the
/// same row" is a measurement rather than an assertion.
#[cfg(test)]
fn find_by_binary_search(&self, oid: &[u8]) -> Option<Located> {
if oid.len() != self.oid_len || self.count == 0 {
return None;
}
let mut lo = 0usize;
let mut hi = self.count;
while lo < hi {
let mid = lo + (hi - lo) / 2;
if self.oid_at(mid) < oid { lo = mid + 1 } else { hi = mid }
}
(lo < self.count && self.oid_at(lo) == oid).then(|| self.locs[lo])
}
/// The **unverified** candidate run for a key: every entry sharing that
/// 8-byte prefix. Normally length 1; length > 1 is a real prefix collision.
/// Exposed so a test can prove the collision it constructed is real and that
/// the verify step is what tells the candidates apart.
#[cfg(test)]
fn candidate_run(&self, key: i64) -> std::ops::Range<usize> {
let Some(tree) = self.tree.as_ref() else { return 0..0 };
let Some(pos) = tree.find_exact(key, &self.keys) else { return 0..0 };
self.expand_run(pos, key)
}
}
/// Counters, all of them applied output.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct Stats {
pub rows: u64,
pub written: u64,
pub served: u64,
pub rederived: u64,
pub absent: u64,
}
/// §14's derived table for one repository.
pub struct ExplodedArchive {
path: PathBuf,
/// The schema message: the file's first message, kept so a point read can
/// feed a [`StreamDecoder`] without re-reading it from disk each time.
schema_msg: Mutex<Option<Vec<u8>>>,
/// Every batch message in the file, in order. Cheap — one entry per 64 MiB.
batches: Mutex<Vec<Batch>>,
/// `oid → Located` as an [`OidTree`]. `None` until the first lookup forces
/// the pass that builds it, so a push-only process never pays for it.
index: Mutex<Option<OidTree>>,
/// The last few decoded batches, most recent first.
///
/// Not an optimisation looking for a problem: without it every point read
/// decodes its whole batch, and the 2687-object push guard went from 4 s to
/// **over 100 s** — 2687 lookups × one 106 MiB batch each. A batch is
/// `Arc`-backed, so a hit is a clone of pointers.
cache: Mutex<Vec<(u32, RecordBatch)>>,
pending: Mutex<PendingBatch>,
pending_bytes: AtomicU64,
/// Bytes in the file. Kept here rather than stat'ed so the append offset and
/// the batch table cannot drift apart.
end: Mutex<u64>,
oid_len: Mutex<Option<usize>>,
policy: ExplodePolicy,
skipped: AtomicU64,
written: AtomicU64,
served: AtomicU64,
/// The subset of `served` answered by the extent `pread` fast path.
pread_served: AtomicU64,
rederived: AtomicU64,
absent: AtomicU64,
/// The file was written with schema v1 (no `enc` column). Every batch this
/// process appends to it is then written in v1 shape too — raw, no `enc` —
/// because an IPC stream has ONE schema message and a six-column batch
/// under a five-column schema does not decode. The pool is a cache; it moves
/// to v2 the day it is dropped and rebuilt, never in place.
legacy_v1: std::sync::atomic::AtomicBool,
/// The file has been evicted ([`Self::retire`]): it is unlinked or about to
/// be. A flush is then a no-op instead of an `O_CREAT` that brings the file
/// back as an orphan nobody lists, and a failed read of it is a miss.
retired: std::sync::atomic::AtomicBool,
}
impl ExplodedArchive {
/// Open or create the table, with the policy `ZNIPPY_GIT_EXPLODE` names.
/// Reads kilobytes, never the payloads.
pub fn open(path: &Path) -> Result<Self> {
Self::open_with_policy(path, ExplodePolicy::from_env()?)
}
/// Open with an explicit policy — what a test or a per-tenant configuration
/// uses, so the setting is not reachable only through the environment.
pub fn open_with_policy(path: &Path, policy: ExplodePolicy) -> Result<Self> {
if let Some(parent) = path.parent()
&& !parent.as_os_str().is_empty()
{
std::fs::create_dir_all(parent)
.with_context(|| format!("creating {}", parent.display()))?;
}
let (schema_msg, batches, end, oid_len) = Self::walk(path)?;
let legacy_v1 = schema_msg.as_ref().is_some_and(|m| !Self::schema_has_enc(m));
Ok(Self {
legacy_v1: std::sync::atomic::AtomicBool::new(legacy_v1),
retired: std::sync::atomic::AtomicBool::new(false),
path: path.to_path_buf(),
schema_msg: Mutex::new(schema_msg),
batches: Mutex::new(batches),
index: Mutex::new(None),
cache: Mutex::new(Vec::new()),
pending: Mutex::new(PendingBatch::default()),
pending_bytes: AtomicU64::new(0),
end: Mutex::new(end),
oid_len: Mutex::new(oid_len),
policy,
skipped: AtomicU64::new(0),
written: AtomicU64::new(0),
served: AtomicU64::new(0),
pread_served: AtomicU64::new(0),
rederived: AtomicU64::new(0),
absent: AtomicU64::new(0),
})
}
pub fn path(&self) -> &Path {
&self.path
}
/// **This file is about to be unlinked by an eviction.** Called with the
/// segment already out of the pool's list and BEFORE the unlink, under the
/// append lock, so a flush that is already writing finishes first and every
/// later one is a no-op. Without it, a holder of an `Arc` taken before the
/// eviction (a reader's snapshot, the indexer's `flush`/`of_kind`) re-created
/// the path with `O_CREAT` after the unlink: an orphan file no pool lists,
/// counted by no budget, adopted as a segment on the next open. Buffered
/// rows are dropped: this is a cache, every row is re-derivable.
pub fn retire(&self) {
let end = self.end.lock().unwrap_or_else(|e| e.into_inner());
self.retired.store(true, Ordering::Release);
drop(end);
if let Ok(mut p) = self.pending.lock() {
let _ = p.take();
}
self.pending_bytes.store(0, Ordering::Release);
}
/// See [`Self::retire`].
pub fn is_retired(&self) -> bool {
self.retired.load(Ordering::Acquire)
}
/// **Drop every derived, rebuildable structure**: the `oid → Located`
/// tree (~50 B per row — 700 MB on linux, and re-extended on every flush
/// while a reader keeps it alive) and the decoded-batch cache. What the
/// memory watchdog's flush reaches for ([`crate::memguard`]); the next
/// lookup rebuilds the tree from the file in one pass.
pub fn drop_caches(&self) {
if let Ok(mut i) = self.index.lock() {
*i = None;
}
if let Ok(mut c) = self.cache.lock() {
c.clear();
}
}
/// Walk the message framing: `[0xFFFFFFFF][len u32][metadata][body]`.
///
/// The metadata flatbuffer carries `bodyLength`, so this SEEKS over every
/// payload — opening a 17 GB table touches kilobytes. It also carries a
/// record batch's row `length`, which is how `rows()` answers without
/// decoding anything.
///
/// A torn tail — a half-written message at EOF — stops the walk and keeps
/// what came before it, and the file is truncated to that boundary on the
/// next append. The table is derived, so a partial batch is re-exploded
/// rather than being an error a client ever sees.
#[allow(clippy::type_complexity)]
fn walk(path: &Path) -> Result<(Option<Vec<u8>>, Vec<Batch>, u64, Option<usize>)> {
let Ok(mut f) = File::open(path) else {
return Ok((None, Vec::new(), 0, None));
};
let total = f.metadata()?.len();
let mut schema_msg: Option<Vec<u8>> = None;
let mut oid_len: Option<usize> = None;
let mut batches = Vec::new();
let mut at = 0u64;
while at + 8 <= total {
f.seek(SeekFrom::Start(at))?;
let mut hdr = [0u8; 8];
if f.read_exact(&mut hdr).is_err() {
break;
}
if u32::from_le_bytes(hdr[0..4].try_into().unwrap()) != 0xFFFF_FFFF {
break;
}
let meta_len = u32::from_le_bytes(hdr[4..8].try_into().unwrap()) as u64;
if meta_len == 0 {
break; // end-of-stream marker
}
if at + 8 + meta_len > total {
break; // torn metadata
}
let mut meta = vec![0u8; meta_len as usize];
if f.read_exact(&mut meta).is_err() {
break;
}
let Ok(msg) = root_as_message(&meta) else {
break;
};
let body = msg.bodyLength().max(0) as u64;
let whole = 8 + meta_len + body;
if at + whole > total {
break; // torn body
}
if let Some(rb) = msg.header_as_record_batch() {
batches.push(Batch {
at,
len: whole,
rows: rb.length().max(0) as u32,
body: at + 8 + meta_len,
pay_data: payload_data_offset(&meta).unwrap_or(EXTENT_UNAVAILABLE),
});
} else if msg.header_as_schema().is_some() {
let mut whole_msg = Vec::with_capacity((8 + meta_len) as usize);
whole_msg.extend_from_slice(&hdr);
whole_msg.extend_from_slice(&meta);
// A schema message the IPC guard refuses is where the good part
// of the file ends, like a torn tail: the table is derived and
// is re-exploded from there.
if znippy_common::ipc_guard::check_stream(&whole_msg).is_err() {
break;
}
schema_msg = Some(whole_msg);
oid_len = Self::oid_len_of_schema_msg(schema_msg.as_ref().unwrap());
}
at += whole;
}
Ok((schema_msg, batches, at, oid_len))
}
/// The declared oid width, so a reopened table refuses a hash it cannot
/// hold instead of writing a batch no reader can line up.
/// The schema inside a stream schema message (`msg` = the 8-byte
/// continuation header + the flatbuffer), read off the flatbuffer directly.
///
/// Not through `StreamDecoder`: fed a schema message alone it returns
/// `Ok(None)` and its `schema()` stays `None` — MEASURED 2026-09-05 with a
/// probe on a file this writer had just produced (`decode=Ok(false)
/// schema=None`). `oid_len_of_schema_msg` had used exactly that path since
/// it was written and therefore always answered `None`; nothing depended on
/// it, so nothing noticed. `legacy_v1` does depend on it, which is how it
/// surfaced: a fresh v2 file was read as v1 and the next append was written
/// five columns wide under a six-column schema.
fn schema_of_msg(msg: &[u8]) -> Option<Schema> {
// `fb_to_schema` panics on a schema with no field vector or a type it
// half-knows: the IPC guard checks the message first.
znippy_common::ipc_guard::check_stream(msg).ok()?;
let meta = msg.get(8..)?;
let m = root_as_message(meta).ok()?;
let fb = m.header_as_schema()?;
Some(znippy_common::arrow::ipc::convert::fb_to_schema(fb))
}
/// Whether the file's schema message carries the v2 `enc` column.
fn schema_has_enc(msg: &[u8]) -> bool {
Self::schema_of_msg(msg).is_some_and(|s| s.field_with_name(COL_ENC).is_ok())
}
fn oid_len_of_schema_msg(msg: &[u8]) -> Option<usize> {
let schema = Self::schema_of_msg(msg)?;
match schema.field_with_name(COL_OID).ok()?.data_type() {
DataType::FixedSizeBinary(n) => Some(*n as usize),
_ => None,
}
}
/// One resolved object with no name attached — what [`crate::exploded::PayloadSink`]
/// can supply.
pub fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
self.explode_at(oid, kind, None, None, payload)
}
/// One resolved object **with its name**: the mode and path a tree walk knows.
///
/// A blob is reachable at many paths, so this records the first one seen and
/// never rewrites it — a row that already exists is not re-keyed by a second
/// sighting, because "which of its names" is not a question this table
/// pretends to answer.
pub fn explode_at(
&self,
oid: &[u8],
kind: GitObjectKind,
mode: Option<u32>,
path: Option<&str>,
payload: &[u8],
) -> Result<()> {
if !self.policy.wants(kind) {
self.skipped.fetch_add(1, Ordering::Relaxed);
return Ok(());
}
{
let mut p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
// **One `extend_from_slice` into the arena, no allocation of its
// own.** This line was `payload.to_vec()` — a whole second copy of
// the object the caller is holding, which is the byte lane AA
// measured on 2026-08-30.
let pay = p.stash(payload);
p.rows.push(Pending {
oid: oid.to_vec(),
code: kind_code(kind),
mode,
path: path.map(str::to_owned),
pay,
});
}
let n = self
.pending_bytes
.fetch_add(payload.len() as u64, Ordering::AcqRel)
+ payload.len() as u64;
if n >= FLUSH_BYTES as u64 {
self.flush()?;
}
Ok(())
}
/// `v1`: write the batch in schema v1 — no `enc` column, every row raw —
/// because the file it is appended to was opened with a v1 schema message.
fn build_batch(rows: &[Pending], arena: &[u8], oid_len: usize, v1: bool) -> Result<RecordBatch> {
if v1 {
let mut oid_b = FixedSizeBinaryBuilder::with_capacity(rows.len(), oid_len as i32);
let mut kind_b = UInt8Builder::with_capacity(rows.len());
let mut mode_b = UInt32Builder::with_capacity(rows.len());
let mut path_b = StringBuilder::new();
let mut pay_b = LargeBinaryBuilder::new();
for r in rows {
oid_b
.append_value(&r.oid)
.map_err(|e| anyhow::anyhow!("exploded oid column: {e}"))?;
kind_b.append_value(r.code);
mode_b.append_option(r.mode);
path_b.append_option(r.path.as_deref());
pay_b.append_value(&arena[r.pay.clone()]);
}
return RecordBatch::try_new(
exploded_schema_v1(oid_len),
vec![
Arc::new(oid_b.finish()),
Arc::new(kind_b.finish()),
Arc::new(mode_b.finish()),
Arc::new(path_b.finish()),
Arc::new(pay_b.finish()),
],
)
.context("building a v1 exploded batch");
}
let mut oid_b = FixedSizeBinaryBuilder::with_capacity(rows.len(), oid_len as i32);
let mut kind_b = UInt8Builder::with_capacity(rows.len());
let mut mode_b = UInt32Builder::with_capacity(rows.len());
let mut path_b = StringBuilder::new();
let mut enc_b = UInt8Builder::with_capacity(rows.len());
let mut pay_b = LargeBinaryBuilder::new();
for r in rows {
oid_b
.append_value(&r.oid)
.map_err(|e| anyhow::anyhow!("exploded oid column: {e}"))?;
kind_b.append_value(r.code);
mode_b.append_option(r.mode);
path_b.append_option(r.path.as_deref());
// Per row, at build time: the pending arena keeps rows raw (so the
// pending-serve path hands them out untouched), and what lands in
// the file is whatever is smaller.
let raw = &arena[r.pay.clone()];
match compress_row(raw) {
Some(z) => {
enc_b.append_value(ENC_ZLIB);
pay_b.append_value(&z);
}
None => {
enc_b.append_value(ENC_RAW);
pay_b.append_value(raw);
}
}
}
RecordBatch::try_new(
exploded_schema(oid_len),
vec![
Arc::new(oid_b.finish()),
Arc::new(kind_b.finish()),
Arc::new(mode_b.finish()),
Arc::new(path_b.finish()),
Arc::new(enc_b.finish()),
Arc::new(pay_b.finish()),
],
)
.context("building the exploded batch")
}
/// Append the buffered rows as ONE batch message. Nothing already in the
/// file is read or rewritten.
pub fn flush(&self) -> Result<u64> {
let (rows, arena) = {
let mut p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
if p.is_empty() {
return Ok(0);
}
self.pending_bytes.store(0, Ordering::Release);
p.take()
};
let n = rows.len() as u64;
let width = rows[0].oid.len();
anyhow::ensure!(
rows.iter().all(|r| r.oid.len() == width),
"one exploded batch cannot hold two hash widths"
);
{
let mut w = self
.oid_len
.lock()
.map_err(|_| anyhow::anyhow!("exploded oid width poisoned"))?;
match *w {
Some(have) => anyhow::ensure!(
have == width,
"this table holds {have}-byte oids and a {width}-byte one arrived"
),
None => *w = Some(width),
}
}
let batch = Self::build_batch(&rows, &arena, width, self.legacy_v1.load(Ordering::Relaxed))?;
let opts = IpcWriteOptions::default();
let ipc = IpcDataGenerator::default();
let mut tracker = DictionaryTracker::new(false);
// The file is opened per flush rather than held: an append writes at a
// known offset and closing between batches is what makes a crashed
// process leave a file whose last complete message is intact.
let mut end = self
.end
.lock()
.map_err(|_| anyhow::anyhow!("exploded end poisoned"))?;
if self.retired.load(Ordering::Acquire) {
// Evicted: its path is gone, and opening it with `create` would
// bring it back as an orphan. The rows were a cache.
return Ok(0);
}
let mut f = OpenOptions::new()
.create(true)
.read(true)
.write(true)
.truncate(false)
.open(&self.path)
.with_context(|| format!("opening {}", self.path.display()))?;
// Drop anything after the last complete message before appending — a
// torn tail from a previous crash must not sit in the middle of the
// stream, where every later batch would be unreachable.
if f.metadata()?.len() != *end {
f.set_len(*end)
.with_context(|| format!("truncating {} to {}", self.path.display(), *end))?;
}
f.seek(SeekFrom::Start(*end))?;
let mut schema_guard = self
.schema_msg
.lock()
.map_err(|_| anyhow::anyhow!("exploded schema poisoned"))?;
if schema_guard.is_none() {
let enc =
ipc.schema_to_bytes_with_dictionary_tracker(&exploded_schema(width), &mut tracker, &opts);
let mut msg = Vec::new();
let (meta, body) = write_message(&mut msg, enc, &opts)
.map_err(|e| anyhow::anyhow!("encoding the exploded schema: {e}"))?;
debug_assert_eq!(meta + body, msg.len());
f.write_all(&msg).context("writing the exploded schema")?;
*end += msg.len() as u64;
*schema_guard = Some(msg);
}
drop(schema_guard);
let (dicts, enc) = ipc
.encode(&batch, &mut tracker, &opts, &mut Default::default())
.map_err(|e| anyhow::anyhow!("encoding an exploded batch: {e}"))?;
anyhow::ensure!(
dicts.is_empty(),
"the exploded schema has no dictionary columns, yet {} arrived",
dicts.len()
);
let mut msg = Vec::new();
let (meta, body) = write_message(&mut msg, enc, &opts)
.map_err(|e| anyhow::anyhow!("encoding an exploded batch: {e}"))?;
debug_assert_eq!(meta + body, msg.len());
f.write_all(&msg).context("appending an exploded batch")?;
let placed = Batch {
at: *end,
len: msg.len() as u64,
rows: rows.len() as u32,
// `meta` from `write_message` is the 8-byte prefix plus the padded
// metadata, so the body begins exactly `meta` bytes in — and the
// metadata slice between prefix and body is what the offset helper
// parses, the same bytes `walk` reads back after a reopen.
body: *end + meta as u64,
pay_data: payload_data_offset(&msg[8..meta]).unwrap_or(EXTENT_UNAVAILABLE),
};
*end += msg.len() as u64;
// Recorded while the append lock is still held, so the batch table stays
// in file order under N concurrent exploders. Correctness would survive
// it being out of order — a lookup addresses a batch by its slot, not by
// its offset — but a table that reads back in file order is one a human
// can check against the file with `od`.
let batch_no = {
let mut b = self
.batches
.lock()
.map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
b.push(placed);
(b.len() - 1) as u32
};
drop(end);
// Only extend an index that already exists. Building one here would make
// the first push pay for a structure no reader has asked for.
let mut idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
if let Some(old) = idx.take() {
// Retired oids are collected and applied in ONE pass. Removing them
// one at a time is quadratic, and a `gc` retires millions.
let mut retired: std::collections::HashSet<&[u8]> = std::collections::HashSet::new();
for r in rows.iter() {
if kind_of(r.code).is_none() {
retired.insert(r.oid.as_slice());
}
}
let mut b = OidTreeBuilder::with_capacity(old.len() + rows.len());
for (oid, loc) in old.iter() {
if !retired.contains(oid) {
b.push(oid, Some(loc))?;
}
}
let pay = batch
.column_by_name(COL_PAYLOAD)
.and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
.context("the exploded table has no payload column")?;
// Freshly built, never sliced, so these offsets are byte-for-byte
// what the encoder just wrote into the file — the same equivalence
// `index_now` relies on after a reopen.
let off = pay.value_offsets();
let enc = enc_column(&batch);
for (i, r) in rows.iter().enumerate() {
if let Some(kind) = kind_of(r.code)
&& !retired.contains(r.oid.as_slice())
{
let (pay_at, pay_len) = if placed.pay_data == EXTENT_UNAVAILABLE {
(EXTENT_UNAVAILABLE, 0)
} else {
(
placed.body + placed.pay_data + off[i] as u64,
(off[i + 1] - off[i]) as u64,
)
};
b.push(
&r.oid,
Some(Located {
batch: batch_no,
row: i as u32,
kind,
pay_at,
pay_len,
enc: enc_at(enc, i),
}),
)?;
}
}
// `Keep::First`, because the surviving entries were pushed before
// this batch's rows: an oid already in the index keeps the row it
// already had, which is what `explode_at` documents about names.
*idx = Some(b.finish(Keep::First));
}
self.written.fetch_add(n, Ordering::Relaxed);
Ok(n)
}
/// How many decoded batches stay resident. Four × [`FLUSH_BYTES`] is the
/// worst case, and a batch is `Arc`-backed so a hit copies pointers.
const CACHE: usize = 4;
/// Read one batch message off disk and decode it — or hand back the decoded
/// copy if it is one of the last [`Self::CACHE`].
///
/// Two `pread`s on a miss: the schema message and the batch. It never reads
/// the batches in front of the one it wants — the property a stream reader
/// iterating from byte zero does not have.
fn read_batch(&self, no: u32) -> Result<RecordBatch> {
{
let mut c = self
.cache
.lock()
.map_err(|_| anyhow::anyhow!("exploded batch cache poisoned"))?;
if let Some(at) = c.iter().position(|(n, _)| *n == no) {
let hit = c.remove(at);
let batch = hit.1.clone();
c.insert(0, hit);
return Ok(batch);
}
}
let placed = {
let b = self
.batches
.lock()
.map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
*b.get(no as usize)
.with_context(|| format!("exploded batch {no} is not in this table"))?
};
let schema_msg = {
let s = self
.schema_msg
.lock()
.map_err(|_| anyhow::anyhow!("exploded schema poisoned"))?;
s.clone()
.context("the exploded table has batches but no schema message")?
};
let f = File::open(&self.path)
.with_context(|| format!("opening {}", self.path.display()))?;
let mut raw = vec![0u8; placed.len as usize];
f.read_exact_at(&mut raw, placed.at)
.with_context(|| format!("reading exploded batch {no}"))?;
// The batch is read off disk under the schema message held apart: the
// IPC guard checks both before arrow sees them, so a hostile file is an
// `Err` and never an arrow-ipc panic or abort. Metadata only, no copy.
znippy_common::ipc_guard::check_batches_after_schema(&schema_msg, &raw)
.map_err(|e| anyhow::anyhow!("exploded batch {no}: {e}"))?;
let mut dec = StreamDecoder::new();
let mut head = Buffer::from_vec(schema_msg);
dec.decode(&mut head)
.map_err(|e| anyhow::anyhow!("decoding the exploded schema: {e}"))?;
let mut body = Buffer::from_vec(raw);
let batch = dec
.decode(&mut body)
.map_err(|e| anyhow::anyhow!("decoding exploded batch {no}: {e}"))?
.with_context(|| format!("exploded batch {no} decoded to no rows"))?;
{
let mut c = self
.cache
.lock()
.map_err(|_| anyhow::anyhow!("exploded batch cache poisoned"))?;
c.retain(|(n, _)| *n != no);
c.insert(0, (no, batch.clone()));
c.truncate(Self::CACHE);
}
Ok(batch)
}
/// Build `oid → Located` with one sequential pass, applying every
/// [`TOMBSTONE`] in file order so a retired oid stays retired across a
/// reopen. Called on the first lookup, never on open.
fn index_now(&self) -> Result<()> {
{
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
if idx.is_some() {
return Ok(());
}
}
let (count, rows, placed) = {
let b = self
.batches
.lock()
.map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
// The row total is in the batch metadata the framing walk already
// read, so the builder's arrays are sized once instead of doubling
// their way to 11.7 million. The count is only a size HINT, and
// a file's metadata can claim any count: each batch's is capped by
// its own length in bytes (every row carries at least its oid), so
// a hostile file cannot size the index past the file itself.
let hint = |x: &Batch| (x.rows as u64).min(x.len) as usize;
(b.len() as u32, b.iter().map(hint).sum::<usize>(), b.clone())
};
// Every row is recorded as a sighting and the fold happens once, at
// `finish`. The `HashMap<Vec<u8>, Located>` this replaced allocated —
// and then dropped — one `Vec<u8>` per oid on a corpus with 11.7 M of
// them, to compute an answer a single stable sort gives.
let mut fold = OidTreeBuilder::with_capacity(rows);
for no in 0..count {
let batch = self.read_batch(no)?;
let oid = batch
.column_by_name(COL_OID)
.and_then(|c| c.as_any().downcast_ref::<FixedSizeBinaryArray>())
.context("the exploded table has no oid column")?;
let kinds = batch
.column_by_name(COL_TYPE)
.and_then(|c| c.as_any().downcast_ref::<UInt8Array>())
.context("the exploded table has no object_type column")?;
let pay = batch
.column_by_name(COL_PAYLOAD)
.and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
.context("the exploded table has no payload column")?;
// The decoded offsets carry the SAME values the file's offsets
// buffer holds — a StreamDecoder copies buffers verbatim — so the
// extent needs no second parse of the body.
let off = pay.value_offsets();
let enc = enc_column(&batch);
let b = placed[no as usize];
for i in 0..batch.num_rows() {
// A tombstone is pushed as a sighting with no location, not
// skipped: batches are walked in file order and rows in row
// order, so `Keep::Last` gives the last word about an oid — and
// when that word is a tombstone the oid is dropped, which is
// what makes the tombstone durable rather than advisory.
fold.push(
oid.value(i),
kind_of(kinds.value(i)).map(|kind| {
let (pay_at, pay_len) = if b.pay_data == EXTENT_UNAVAILABLE {
(EXTENT_UNAVAILABLE, 0)
} else {
(
b.body + b.pay_data + off[i] as u64,
(off[i + 1] - off[i]) as u64,
)
};
Located { batch: no, row: i as u32, kind, pay_at, pay_len, enc: enc_at(enc, i) }
}),
)?;
}
}
let out = fold.finish(Keep::Last);
let mut idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
if idx.is_none() {
*idx = Some(out);
}
Ok(())
}
fn find(&self, oid: &[u8]) -> Result<Option<Located>> {
self.index_now()?;
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
Ok(idx.as_ref().expect("index_now built it").find(oid))
}
/// The object's kind and bytes, or `None` — *fall back and rebuild*, never
/// *wrong*.
pub fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
// A buffered row has not been written yet, and a read must not miss it:
// an "absent" that is really "not flushed" sends the caller off to
// re-resolve a whole pack for an object we are holding in RAM.
{
let p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
if let Some(r) = p.rows.iter().rev().find(|r| r.oid == oid) {
let Some(kind) = kind_of(r.code) else {
// The newest buffered word about this oid retires it.
self.absent.fetch_add(1, Ordering::Relaxed);
return Ok(None);
};
self.served.fetch_add(1, Ordering::Relaxed);
return Ok(Some((kind, p.bytes(r).to_vec())));
}
}
let Some(loc) = self.find(oid)? else {
self.absent.fetch_add(1, Ordering::Relaxed);
return Ok(None);
};
// ── the point read IS a point read ──────────────────────────────────
//
// One pread of exactly the payload's bytes, addressed by the extent the
// index carries. The batch-decode below is the FALLBACK — correct for a
// compressed or foreign-layout batch — not the path a healthy table
// takes: taking it per object is what turned a rust-lang/rust
// connectivity walk into hours of one-core batch decoding (2026-08-19).
if loc.pay_at != EXTENT_UNAVAILABLE {
let f = File::open(&self.path)
.with_context(|| format!("opening {}", self.path.display()))?;
let mut buf = vec![0u8; loc.pay_len as usize];
f.read_exact_at(&mut buf, loc.pay_at)
.with_context(|| format!("preading payload at {}", loc.pay_at))?;
self.served.fetch_add(1, Ordering::Relaxed);
self.pread_served.fetch_add(1, Ordering::Relaxed);
// The extent held the row as stored; a compressed row is inflated
// here, from the index's `enc`, without decoding the batch.
return Ok(Some((loc.kind, row_payload(loc.enc, &buf)?)));
}
let batch = self.read_batch(loc.batch)?;
let pay = batch
.column_by_name(COL_PAYLOAD)
.and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
.context("the exploded table has no payload column")?;
let enc = enc_at(enc_column(&batch), loc.row as usize);
self.served.fetch_add(1, Ordering::Relaxed);
Ok(Some((loc.kind, row_payload(enc, pay.value(loc.row as usize))?)))
}
/// Point reads served by the extent `pread` rather than a batch decode —
/// how a test proves the fast path is the one actually taken, instead of
/// timing something.
pub fn pread_served(&self) -> u64 {
self.pread_served.load(Ordering::Relaxed)
}
/// The name this object was first exploded at, if any caller knew one.
/// `None` is "not recorded", never "at the root".
pub fn name(&self, oid: &[u8]) -> Result<Option<(Option<u32>, Option<String>)>> {
{
let p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
if let Some(r) = p.rows.iter().rev().find(|r| r.oid == oid) {
return match kind_of(r.code) {
Some(_) => Ok(Some((r.mode, r.path.clone()))),
None => Ok(None),
};
}
}
let Some(loc) = self.find(oid)? else {
return Ok(None);
};
let batch = self.read_batch(loc.batch)?;
let mode = batch
.column_by_name(COL_MODE)
.and_then(|c| c.as_any().downcast_ref::<UInt32Array>())
.context("the exploded table has no mode column")?;
let path = batch
.column_by_name(COL_PATH)
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
.context("the exploded table has no path column")?;
let i = loc.row as usize;
Ok(Some((
(!mode.is_null(i)).then(|| mode.value(i)),
(!path.is_null(i)).then(|| path.value(i).to_owned()),
)))
}
pub fn has(&self, oid: &[u8]) -> Result<bool> {
{
let p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
if let Some(r) = p.rows.iter().rev().find(|r| r.oid == oid) {
return Ok(kind_of(r.code).is_some());
}
}
Ok(self.find(oid)?.is_some())
}
/// Every oid of one kind, with its payload. The graph fold's input.
///
/// Driven by the **index**, not by a scan of the file, because the file
/// holds superseded and retired rows too. A scan that read the file directly
/// is exactly how `gc`'d commits came back after a reopen.
pub fn of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
self.flush()?;
self.index_now()?;
// Grouped by batch so each one is decoded once, in file order.
let mut by_batch: std::collections::BTreeMap<u32, Vec<(u32, Vec<u8>)>> =
std::collections::BTreeMap::new();
{
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
for (oid, loc) in idx.as_ref().expect("index_now built it").iter() {
if loc.kind == kind {
by_batch.entry(loc.batch).or_default().push((loc.row, oid.to_vec()));
}
}
}
let mut out = Vec::new();
for (no, mut wanted) in by_batch {
wanted.sort_unstable_by_key(|(r, _)| *r);
let batch = self.read_batch(no)?;
let pay = batch
.column_by_name(COL_PAYLOAD)
.and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
.context("the exploded table has no payload column")?;
let enc = enc_column(&batch);
for (row, oid) in wanted {
out.push((oid, row_payload(enc_at(enc, row as usize), pay.value(row as usize))?));
}
}
Ok(out)
}
/// Retire every oid `live` rejects, **durably**.
///
/// The file is append-only, so this appends a [`TOMBSTONE`] row per retired
/// oid rather than deleting anything. The bytes of the dead rows stay until
/// the table is rebuilt — reclaiming them means rewriting up to 17 GB, which
/// is a job with its own proof obligations and not something to smuggle into
/// `gc` — but the *rows* are gone from every reader, including a reopened
/// one.
///
/// Dropping only from the in-memory index is what a first cut did. Seen RED
/// by `store::tests::a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened`:
/// **"the graph after gc + reopen holds 551 commits, not the one live one"**.
pub fn retain(&self, live: &dyn Fn(&[u8]) -> bool) -> Result<u64> {
self.flush()?;
self.index_now()?;
let dead: Vec<Vec<u8>> = {
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
idx.as_ref()
.expect("index_now built it")
.iter()
.filter(|(oid, _)| !live(oid))
.map(|(oid, _)| oid.to_vec())
.collect()
};
if dead.is_empty() {
return Ok(0);
}
{
let mut p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
for oid in &dead {
p.rows.push(Pending {
oid: oid.clone(),
code: TOMBSTONE,
mode: None,
path: None,
pay: 0..0,
});
}
}
// Written before returning, so a `gc` that says it retired rows has
// already said so on disk.
self.flush()?;
Ok(dead.len() as u64)
}
/// Rows in the table. Answered from batch metadata when no index has been
/// built, so startup does not read a payload.
pub fn rows(&self) -> Result<u64> {
let pending = {
let p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
p.rows.len() as u64
};
{
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
if let Some(idx) = idx.as_ref() {
return Ok(idx.len() as u64 + pending);
}
}
let b = self
.batches
.lock()
.map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
Ok(b.iter().map(|x| x.rows as u64).sum::<u64>() + pending)
}
/// Bytes the table occupies — the number redb could not keep anywhere near
/// its payload total.
pub fn disk_bytes(&self) -> u64 {
std::fs::metadata(&self.path).map(|m| m.len()).unwrap_or(0)
}
pub fn note_rederived(&self) {
self.rederived.fetch_add(1, Ordering::Relaxed);
}
/// What this table is configured to keep.
pub fn policy(&self) -> ExplodePolicy {
self.policy
}
/// Objects the policy declined, COUNTED: a table short because of policy and
/// one short because of a fault must not look alike to `adopt_journal`.
pub fn skipped(&self) -> u64 {
self.skipped.load(Ordering::Relaxed)
}
pub fn stats(&self) -> Stats {
Stats {
rows: self.rows().unwrap_or(0),
written: self.written.load(Ordering::Relaxed),
served: self.served.load(Ordering::Relaxed),
rederived: self.rederived.load(Ordering::Relaxed),
absent: self.absent.load(Ordering::Relaxed),
}
}
/// The engine-wide counter type, so no caller can tell the medium apart by
/// its instruments.
pub fn engine_stats(&self) -> crate::exploded::ExplodedStats {
let s = self.stats();
crate::exploded::ExplodedStats {
rows: s.rows,
written: s.written,
served: s.served,
rederived: s.rederived,
absent: s.absent,
}
}
}
impl crate::exploded::PayloadSink for ExplodedArchive {
/// The policy, asked before the payload exists rather than after — and the
/// `skipped` counter is bumped here, which is the only place the object is
/// ever seen when the answer is no.
fn wants(&self, kind: GitObjectKind) -> bool {
if self.policy.wants(kind) {
return true;
}
self.skipped.fetch_add(1, Ordering::Relaxed);
false
}
fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
ExplodedArchive::explode(self, oid, kind, payload)
}
fn flush(&self) -> Result<()> {
ExplodedArchive::flush(self).map(|_| ())
}
}
// ════════════════════════════════════════════════════════════════════════════
// The policy: what gets exploded at all
// ════════════════════════════════════════════════════════════════════════════
/// **How much of a pack is resolved into the exploded table.**
///
/// Three values and not a boolean, because a flat off-switch reintroduces a
/// correctness bug this table exists to fix. Commit and tree payloads used to
/// live only in RAM, so a **clean** shutdown came back with every pack's
/// `indexed` bit legitimately set and an EMPTY graph — MEASURED on oden
/// 2026-08-08: 2687 rows and **0 of 551 commits**, `reachable()` returning a
/// commit instead of its closure and `gc` seeing an empty live set.
///
/// MEASURED on `linux.git`: resolved content is 16.8 GB against a 6.4 GB pack,
/// and blobs are nearly all of it. So [`Graph`](ExplodePolicy::Graph) buys the
/// correctness fix for a small fraction of the disk, and
/// [`Full`](ExplodePolicy::Full) — one-lookup content reads, thin-pack bases,
/// and the path columns that make the table a filesystem — is the part worth
/// charging for.
///
/// **The default is [`Commits`](ExplodePolicy::Commits) since lane Q
/// (2026-08-30), and the change is a measurement.** §14's `indexed` bit is
/// derived by comparing this table's row count against the index's, so a table
/// that is *deliberately* short is indistinguishable, on disk, from one that was
/// dropped — see [`crate::git_ops::Absorber::adopt_journal`], which weakens that
/// comparison for every deliberately-short policy and cannot make it exact
/// without a durable per-kind count. That weakening is the price of the
/// default, it is stated here rather than hidden, and it costs a store that
/// really did lose its table one extra fall-back read per object rather than a
/// re-ingest.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ExplodePolicy {
/// Store nothing. Reads re-resolve, and every pack is re-queued on reopen —
/// which is what rebuilds the graph, since nothing it is folded from is on
/// disk.
Off,
/// Commits and trees, both eager. The shape lane Q shipped first and
/// **measured**: on the real `torvalds/linux` mirror the tree payloads
/// alone are **28.24 GB — 3.24× the 8.13 GiB pack** — and folding them into
/// RAM took peak RSS to 50.9 GiB. Kept as a setting, because a small
/// repository that is read constantly wants exactly this; not the default,
/// because an appliance ships with 16 GB.
Graph,
/// **Commits and tags only — the SHIPPED DEFAULT since lane Q
/// (2026-08-30).**
///
/// The index-only-at-push shape of
/// `ideas/lazy-explode-index-only-at-push.md`, in full. Commits are what
/// `graph` and `reach` fold and what a visibility or pull check reads; a
/// tag is what a clone peels. Both are small and both stay eager, so
/// neither ever stalls. **Trees and blobs** — 90 % of the objects and
/// essentially all of the bytes — are written into
/// [`crate::exploded_cache`]'s budgeted pool by the FIRST reader that
/// inflates one, and evicted from it when the budget says so.
///
/// Trees became lazy when [`crate::reach::TreeSource`] replaced the
/// resident `HashMap<String, Vec<u8>>` that [`crate::serve`]'s closure walk
/// and the reachability builder both read. That map was the 28.24 GB above.
///
/// MEASURED, `torvalds/linux`, oden 2026-08-30, /data bytes per pack byte:
///
/// | | eager (`Full`) | `Graph` | `Commits` |
/// |---|---:|---:|---:|
/// | at ack | 1.05× | 1.05× | 1.05× |
/// | at index-complete | 22.4× (lane P, 78 % done) | 4.58× | see the lane Q rows |
#[default]
Commits,
/// Every object, eagerly, with no bound. What shipped before lane Q and
/// what the 17× measurement is of; kept so a bench can measure the
/// unbounded shape and so a store written under it still opens.
Full,
}
impl ExplodePolicy {
pub const fn wants(self, kind: GitObjectKind) -> bool {
match self {
ExplodePolicy::Off => false,
ExplodePolicy::Graph => matches!(kind, GitObjectKind::Commit | GitObjectKind::Tree),
ExplodePolicy::Commits => matches!(kind, GitObjectKind::Commit | GitObjectKind::Tag),
ExplodePolicy::Full => true,
}
}
pub const fn as_str(self) -> &'static str {
match self {
ExplodePolicy::Off => "off",
ExplodePolicy::Graph => "graph",
ExplodePolicy::Commits => "commits",
ExplodePolicy::Full => "full",
}
}
/// An unknown value is an ERROR, never a silent fall back to the default: a
/// typo that quietly disables a paid tier is not noticed until a bill is
/// wrong.
pub fn parse(s: &str) -> Result<Self> {
match s.trim().to_ascii_lowercase().as_str() {
"off" | "none" => Ok(ExplodePolicy::Off),
"graph" | "commits-and-trees" => Ok(ExplodePolicy::Graph),
"commits" | "lazy" | "commits-only" => Ok(ExplodePolicy::Commits),
"full" | "all" => Ok(ExplodePolicy::Full),
other => anyhow::bail!(
"unknown explode policy {other:?}; expected off, commits, graph or full"
),
}
}
/// Reads `ZNIPPY_GIT_EXPLODE` ([`crate::arms::ENV_EXPLODE`]) through the
/// crate's one counted door, once per store open.
pub fn from_env() -> Result<Self> {
match crate::arms::read_env(crate::arms::ENV_EXPLODE) {
Some(v) => Self::parse(&v),
None => Ok(explode_default()),
}
}
/// The discriminant, for the atomic behind [`set_explode_default`]. Private
/// because it is storage and not a contract — the wire spelling is
/// [`ExplodePolicy::as_str`].
const fn as_u8(self) -> u8 {
match self {
ExplodePolicy::Off => 0,
ExplodePolicy::Graph => 1,
ExplodePolicy::Commits => 2,
ExplodePolicy::Full => 3,
}
}
const fn from_u8(v: u8) -> Self {
match v {
0 => ExplodePolicy::Off,
1 => ExplodePolicy::Graph,
3 => ExplodePolicy::Full,
_ => ExplodePolicy::Commits,
}
}
}
/// What [`ExplodePolicy::from_env`] answers when [`crate::arms::ENV_EXPLODE`] is unset.
/// [`ExplodePolicy::default`] unless an embedder said otherwise.
static EXPLODE_DEFAULT: std::sync::atomic::AtomicU8 =
std::sync::atomic::AtomicU8::new(ExplodePolicy::Commits.as_u8());
/// The current process-wide default — what an unset environment means right now.
pub fn explode_default() -> ExplodePolicy {
ExplodePolicy::from_u8(EXPLODE_DEFAULT.load(Ordering::Relaxed))
}
/// **Set the process-wide default explode policy** for every store opened AFTER this call
/// (an explicit [`crate::arms::ENV_EXPLODE`] still wins). The twin of
/// [`crate::arms::set_graph_cache_default`], for an embedder that learns its role at runtime
/// and cannot set the environment — edition 2024 makes `set_var` unsafe, and gunnar forbids
/// `unsafe` — such as gunnar's twin, which holds packs, serves no reads, and must store no
/// exploded pool at all ([`ExplodePolicy::Off`]).
pub fn set_explode_default(policy: ExplodePolicy) {
EXPLODE_DEFAULT.store(policy.as_u8(), Ordering::Relaxed);
}
#[cfg(test)]
mod tests {
use super::*;
use crate::exploded::PayloadSink as _;
fn tmp(name: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("exploded-arrow-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
std::fs::create_dir_all(&d).unwrap();
d.join("objects.exploded")
}
fn oid(n: u8) -> [u8; 20] {
[n; 20]
}
/// ★ ONE TABLE, and the payload is IN the row.
///
/// Red-first: drop the payload column and nothing below compiles; widen
/// `payload` to `Binary` and this fails on the type it asserts.
#[test]
fn the_table_holds_the_payload_and_a_name_in_one_row() {
let s = exploded_schema(20);
let names: Vec<&str> = s.fields().iter().map(|f| f.name().as_str()).collect();
// `enc` sits before `payload` so the payload data buffer stays the last
// buffer of the batch — `payload_data_offset` depends on that.
assert_eq!(names, vec![COL_OID, COL_TYPE, COL_MODE, COL_PATH, COL_ENC, COL_PAYLOAD]);
assert_eq!(
s.field_with_name(COL_PAYLOAD).unwrap().data_type(),
&DataType::LargeBinary,
"Binary offsets are i32 and would cap one batch at 2 GiB of payload"
);
assert!(
s.field_with_name(COL_PATH).unwrap().is_nullable(),
"a bare pack resolve knows no path; null must be a legal answer"
);
}
/// ★ THE WHOLE POINT: the file stays near the payload total.
///
/// redb held 16.8 GB of kernel payload in a 204 GB file — 12×. An Arrow IPC
/// stream has no pages to rewrite, so the overhead is framing. Red-first:
/// put this back on a copy-on-write B-tree and it fails.
#[test]
fn the_file_stays_close_to_the_payload_total() {
let t = ExplodedArchive::open_with_policy(&tmp("size"), ExplodePolicy::Full).unwrap();
let payload = vec![7u8; 4096];
let n = 2000u64;
for i in 0..n {
let mut o = [0u8; 20];
o[..8].copy_from_slice(&i.to_le_bytes());
t.explode(&o, GitObjectKind::Blob, &payload).unwrap();
}
t.flush().unwrap();
let payload_total = n * payload.len() as u64;
let on_disk = t.disk_bytes();
assert!(
on_disk < payload_total + payload_total / 4,
"{on_disk} bytes on disk against {payload_total} of payload — the medium is amplifying"
);
assert_eq!(t.rows().unwrap(), n);
}
/// ★ A point read is one small `pread` of the payload's own bytes, and it
/// is the path ACTUALLY taken — counted, not timed.
///
/// Red-first, measured on a rust-lang/rust push (2026-08-19): without the
/// extent, every `content` decoded its whole 64 MiB batch against a 4-slot
/// cache, and the pre-ack connectivity walk — graph order, so effectively
/// random across ~574 batches — decoded 6.7 GB/s of page cache to serve
/// ~98 objects/s, one core pinned for hours. Both index-building paths are
/// exercised: the in-process flush extension AND `index_now` after a
/// reopen, against multi-batch tables with varied payload sizes, so a
/// mis-computed extent returns the WRONG BYTES here rather than in a
/// customer's clone.
#[test]
fn a_point_read_preads_the_exact_payload_without_decoding_the_batch() {
let path = tmp("pread");
let pay = |b: u8, i: u8| vec![b ^ i; 100 + i as usize * 7];
let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
// Batch 0, then force the index to exist so batch 1 goes through the
// flush EXTENSION path rather than a later index_now.
for i in 0..40u8 {
t.explode(&[i; 20], GitObjectKind::Blob, &pay(0, i)).unwrap();
}
t.flush().unwrap();
assert!(t.content(&[0u8; 20]).unwrap().is_some(), "build the index");
for i in 40..80u8 {
t.explode(&[i; 20], GitObjectKind::Blob, &pay(1, i)).unwrap();
}
t.flush().unwrap();
let before = t.pread_served();
for i in 0..80u8 {
let (kind, got) = t.content(&[i; 20]).unwrap().expect("every oid is live");
assert_eq!(kind, GitObjectKind::Blob);
let want = if i < 40 { pay(0, i) } else { pay(1, i) };
assert_eq!(got, want, "oid {i}: the pread returned some OTHER bytes");
}
assert_eq!(
t.pread_served() - before,
80,
"a read fell back to the batch decode — the extent was not computed"
);
// Reopen: the framing walk + index_now must reproduce the same extents
// from the file alone.
drop(t);
let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
for i in 0..80u8 {
let (_, got) = t.content(&[i; 20]).unwrap().expect("survives a reopen");
let want = if i < 40 { pay(0, i) } else { pay(1, i) };
assert_eq!(got, want, "oid {i} after reopen");
}
assert_eq!(t.pread_served(), 80, "the reopened table must pread too");
}
/// **Lever 3: a compressible row is stored compressed and read back whole, on
/// every read path.** Three rows: a 4 KiB compressible one (stored zlib), a
/// row under [`MIN_ROW_COMPRESS`] (stored raw), and a 4 KiB incompressible
/// one (zlib cannot get it under the keep threshold — stored raw). Asserted
/// on the `enc` column read straight off the file, on the file being smaller
/// than the payloads, and on `content` (the pread fast path — `pread_served`
/// moves) and `of_kind` (the batch-decode path) both handing back the
/// original bytes.
///
/// Seen RED by returning `bytes.to_vec()` for `ENC_ZLIB` in `row_payload`:
/// "the pread path returned the stored bytes, not the payload" — 4096 bytes
/// expected, ~60 came back. Restored.
#[test]
fn a_compressible_row_is_stored_compressed_and_read_back_whole() {
let path = tmp("lever3");
let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
let big: Vec<u8> = b"tree 0123456789abcdef0123456789abcdef01234567\nparent x\n"
.iter()
.cycle()
.take(4096)
.copied()
.collect();
let tiny = b"tree x\n".to_vec();
let noise: Vec<u8> = {
let mut x: u32 = 0x1234_5678;
(0..4096)
.map(|_| {
x = x.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
(x >> 24) as u8
})
.collect()
};
assert!(tiny.len() < MIN_ROW_COMPRESS && big.len() >= MIN_ROW_COMPRESS);
t.explode(&oid(1), GitObjectKind::Commit, &big).unwrap();
t.explode(&oid(2), GitObjectKind::Commit, &tiny).unwrap();
t.explode(&oid(3), GitObjectKind::Blob, &noise).unwrap();
t.flush().unwrap();
// The file holds the compressed row: smaller than the payloads it
// carries, which a raw pool never is.
let file = std::fs::metadata(&path).unwrap().len() as usize;
assert!(
file < big.len() + tiny.len() + noise.len(),
"the pool did not compress: {file} B for {} B of payload",
big.len() + tiny.len() + noise.len()
);
// The `enc` column, read off the file as the plain IPC stream it is.
let rd = znippy_common::arrow::ipc::reader::StreamReader::try_new(
std::fs::File::open(&path).unwrap(),
None,
)
.unwrap();
let mut encs = std::collections::BTreeMap::new();
for b in rd {
let b = b.unwrap();
let oids = b
.column_by_name(COL_OID)
.and_then(|c| c.as_any().downcast_ref::<FixedSizeBinaryArray>())
.unwrap();
let enc = enc_column(&b).expect("a v2 pool has the enc column");
for i in 0..b.num_rows() {
encs.insert(oids.value(i)[0], enc.value(i));
}
}
assert_eq!(encs[&1], ENC_ZLIB, "the compressible row is stored compressed");
assert_eq!(encs[&2], ENC_RAW, "a row under the floor stays raw");
assert_eq!(encs[&3], ENC_RAW, "an incompressible row stays raw");
// The pread fast path inflates.
let before = t.pread_served();
let (k, got) = t.content(&oid(1)).unwrap().unwrap();
assert_eq!(k, GitObjectKind::Commit);
assert_eq!(got, big, "the pread path returned the stored bytes, not the payload");
assert_eq!(t.content(&oid(2)).unwrap().unwrap().1, tiny);
assert_eq!(t.content(&oid(3)).unwrap().unwrap().1, noise);
assert_eq!(t.pread_served() - before, 3, "all three went through the extent pread");
// The batch-decode path (`of_kind`) inflates too.
let commits = t.of_kind(GitObjectKind::Commit).unwrap();
let by: std::collections::HashMap<u8, Vec<u8>> =
commits.into_iter().map(|(o, p)| (o[0], p)).collect();
assert_eq!(by[&1], big, "of_kind returned the stored bytes, not the payload");
assert_eq!(by[&2], tiny);
// And across a reopen, where the index is rebuilt from the file.
drop(t);
let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
assert_eq!(t.content(&oid(1)).unwrap().unwrap().1, big, "after reopen");
assert_eq!(t.content(&oid(3)).unwrap().unwrap().1, noise, "after reopen");
}
/// **A v1 pool — no `enc` column — still reads, every row raw.** Written by
/// hand in v1's shape (five columns, the same framing `flush` uses), then
/// opened by this build: the extent is computed (12 buffers is accepted),
/// `enc` is absent and read as [`ENC_RAW`], and the bytes come back as
/// written. This is the promise that lets an existing appliance's pools
/// open unchanged.
///
/// Seen RED by making `payload_data_offset` accept only 14 buffers: the
/// pread count stayed 0 ("a v1 pool must still take the extent fast path").
/// Restored.
#[test]
fn a_v1_pool_without_the_enc_column_reads_raw() {
use znippy_common::arrow::array::{
FixedSizeBinaryBuilder, LargeBinaryBuilder, StringBuilder, UInt32Builder,
};
use znippy_common::arrow::ipc::writer::{write_message, DictionaryTracker, IpcDataGenerator, IpcWriteOptions};
let path = tmp("v1pool");
let v1_schema: SchemaRef = Arc::new(Schema::new(vec![
Field::new(COL_OID, DataType::FixedSizeBinary(20), false),
Field::new(COL_TYPE, DataType::UInt8, false),
Field::new(COL_MODE, DataType::UInt32, true),
Field::new(COL_PATH, DataType::Utf8, true),
Field::new(COL_PAYLOAD, DataType::LargeBinary, false),
]));
let payloads: Vec<Vec<u8>> = (1..=5u8)
.map(|i| b"tree abc\n".iter().cycle().take(300 + i as usize * 100).copied().collect())
.collect();
let mut oid_b = FixedSizeBinaryBuilder::with_capacity(5, 20);
let mut kind_b = UInt8Builder::with_capacity(5);
let mut mode_b = UInt32Builder::with_capacity(5);
let mut path_b = StringBuilder::new();
let mut pay_b = LargeBinaryBuilder::new();
for (i, p) in payloads.iter().enumerate() {
oid_b.append_value(oid(i as u8 + 1)).unwrap();
kind_b.append_value(kind_code(GitObjectKind::Commit));
mode_b.append_null();
path_b.append_null();
pay_b.append_value(p);
}
let batch = RecordBatch::try_new(
v1_schema.clone(),
vec![
Arc::new(oid_b.finish()),
Arc::new(kind_b.finish()),
Arc::new(mode_b.finish()),
Arc::new(path_b.finish()),
Arc::new(pay_b.finish()),
],
)
.unwrap();
let opts = IpcWriteOptions::default();
let ipc = IpcDataGenerator::default();
let mut tracker = DictionaryTracker::new(false);
let mut file = Vec::new();
{
let enc = ipc.schema_to_bytes_with_dictionary_tracker(&v1_schema, &mut tracker, &opts);
let mut msg = Vec::new();
write_message(&mut msg, enc, &opts).unwrap();
file.extend_from_slice(&msg);
}
{
let (_, enc) = ipc.encode(&batch, &mut tracker, &opts, &mut Default::default()).unwrap();
let mut msg = Vec::new();
write_message(&mut msg, enc, &opts).unwrap();
file.extend_from_slice(&msg);
}
std::fs::write(&path, &file).unwrap();
let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
for (i, p) in payloads.iter().enumerate() {
let (k, got) = t.content(&oid(i as u8 + 1)).unwrap().expect("a v1 row is live");
assert_eq!(k, GitObjectKind::Commit);
assert_eq!(&got, p, "v1 row {} must read back raw", i + 1);
}
assert_eq!(t.pread_served(), 5, "a v1 pool must still take the extent fast path");
// And it keeps working when this build appends to it: the appended
// batch must be v1-shaped, because the stream's one schema message is
// v1 — a six-column batch under it would not decode. Proven by reading
// every batch back through Arrow's own stream reader afterwards.
t.explode(&oid(9), GitObjectKind::Commit, &payloads[0]).unwrap();
t.flush().unwrap();
assert_eq!(t.content(&oid(9)).unwrap().unwrap().1, payloads[0]);
assert_eq!(t.content(&oid(1)).unwrap().unwrap().1, payloads[0]);
let rd = znippy_common::arrow::ipc::reader::StreamReader::try_new(
std::fs::File::open(&path).unwrap(),
None,
)
.unwrap();
let mut rows = 0;
for b in rd {
let b = b.expect("every batch in a v1 file decodes under the v1 schema");
assert!(enc_column(&b).is_none(), "a batch appended to a v1 pool must be v1-shaped");
rows += b.num_rows();
}
assert_eq!(rows, 6, "five v1 rows and the one appended");
}
#[test]
fn a_payload_round_trips_through_the_table() {
let t = ExplodedArchive::open_with_policy(&tmp("rt"), ExplodePolicy::Full).unwrap();
t.explode(&oid(1), GitObjectKind::Commit, b"tree deadbeef\n")
.unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"hello world")
.unwrap();
t.flush().unwrap();
assert_eq!(
t.content(&oid(2)).unwrap().unwrap(),
(GitObjectKind::Blob, b"hello world".to_vec())
);
assert_eq!(
t.content(&oid(1)).unwrap().unwrap(),
(GitObjectKind::Commit, b"tree deadbeef\n".to_vec())
);
assert!(t.has(&oid(1)).unwrap());
assert!(!t.has(&oid(9)).unwrap());
}
/// The row is a FILE: mode and path survive the round trip.
#[test]
fn a_row_can_carry_the_name_it_was_seen_at() {
let t = ExplodedArchive::open_with_policy(&tmp("named"), ExplodePolicy::Full).unwrap();
t.explode_at(
&oid(1),
GitObjectKind::Blob,
Some(0o100755),
Some("scripts/build.sh"),
b"#!/bin/sh\n",
)
.unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"anonymous").unwrap();
t.flush().unwrap();
assert_eq!(
t.name(&oid(1)).unwrap().unwrap(),
(Some(0o100755), Some("scripts/build.sh".to_owned()))
);
assert_eq!(
t.name(&oid(2)).unwrap().unwrap(),
(None, None),
"null is 'not recorded', and must not read as a real name"
);
}
/// Reopening finds every row without reading a payload, and appends continue
/// the SAME stream rather than starting a second one.
#[test]
fn the_table_survives_a_reopen_and_keeps_appending() {
let p = tmp("reopen");
{
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
t.explode(&oid(5), GitObjectKind::Tree, b"100644 f\0").unwrap();
t.flush().unwrap();
}
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
assert_eq!(t.rows().unwrap(), 1, "row count comes from batch metadata");
t.explode(&oid(6), GitObjectKind::Blob, b"second session").unwrap();
t.flush().unwrap();
assert_eq!(t.rows().unwrap(), 2);
assert_eq!(
t.content(&oid(5)).unwrap().unwrap(),
(GitObjectKind::Tree, b"100644 f\0".to_vec())
);
assert_eq!(
t.content(&oid(6)).unwrap().unwrap(),
(GitObjectKind::Blob, b"second session".to_vec())
);
// And a third open sees both — one schema message, two batches.
let again = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
assert_eq!(again.rows().unwrap(), 2);
assert_eq!(again.batches.lock().unwrap().len(), 2);
}
/// A read must not miss a row that is buffered but not yet written — an
/// "absent" that is really "not flushed" sends the caller off to re-resolve
/// a whole pack for an object we are holding.
#[test]
fn an_unflushed_row_is_still_served() {
let t = ExplodedArchive::open_with_policy(&tmp("unflushed"), ExplodePolicy::Full).unwrap();
t.explode(&oid(4), GitObjectKind::Commit, b"buffered").unwrap();
assert_eq!(
t.content(&oid(4)).unwrap().unwrap(),
(GitObjectKind::Commit, b"buffered".to_vec())
);
assert!(t.has(&oid(4)).unwrap());
}
#[test]
fn an_absent_oid_is_none_and_counted() {
let t = ExplodedArchive::open_with_policy(&tmp("absent"), ExplodePolicy::Full).unwrap();
assert!(t.content(&oid(3)).unwrap().is_none());
assert_eq!(t.stats().absent, 1);
assert_eq!(t.stats().served, 0);
}
#[test]
fn of_kind_selects_by_kind() {
let t = ExplodedArchive::open_with_policy(&tmp("kind"), ExplodePolicy::Full).unwrap();
t.explode(&oid(1), GitObjectKind::Commit, b"c1").unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"bb").unwrap();
t.explode(&oid(3), GitObjectKind::Commit, b"c2").unwrap();
let commits = t.of_kind(GitObjectKind::Commit).unwrap();
assert_eq!(commits.len(), 2);
assert!(commits.iter().all(|(_, p)| p[0] == b'c'));
}
/// ★ A RETIRED ROW STAYS RETIRED ACROSS A REOPEN.
///
/// The file is append-only, so `retain` writes a [`TOMBSTONE`] rather than
/// deleting. Seen RED with an index-only `retain`: the reopen below rebuilds
/// from the file and brought the dead row back — which is
/// `store::tests::a_wholly_dead_pack_does_not_come_back_when_the_store_is_reopened`
/// failing with "551 commits, not the one live one", reached from here in
/// three lines instead of a whole gc.
#[test]
fn retain_drops_only_what_is_dead_and_the_drop_survives_a_reopen() {
let p = tmp("retain");
{
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
t.explode(&oid(1), GitObjectKind::Blob, b"a").unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"b").unwrap();
t.flush().unwrap();
assert_eq!(t.retain(&|o: &[u8]| o[0] == 1).unwrap(), 1);
assert!(t.has(&oid(1)).unwrap());
assert!(!t.has(&oid(2)).unwrap());
assert_eq!(
t.of_kind(GitObjectKind::Blob).unwrap().len(),
1,
"a scan must not resurrect what retain dropped"
);
}
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
assert!(t.has(&oid(1)).unwrap(), "the live row did not survive");
assert!(
!t.has(&oid(2)).unwrap(),
"a reopen resurrected a retired row — the tombstone is not durable"
);
assert!(t.content(&oid(2)).unwrap().is_none());
assert_eq!(t.of_kind(GitObjectKind::Blob).unwrap().len(), 1);
}
/// sha256 keys are not truncated to sha1 width, and a table refuses a width
/// it cannot hold.
#[test]
fn a_sha256_oid_keeps_all_thirty_two_bytes() {
let p = tmp("sha256");
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
let o = [3u8; 32];
t.explode(&o, GitObjectKind::Tree, b"t").unwrap();
t.flush().unwrap();
assert!(t.has(&o).unwrap());
assert_eq!(t.content(&o).unwrap().unwrap().1, b"t");
t.explode(&oid(1), GitObjectKind::Blob, b"narrow").unwrap();
let err = t.flush().expect_err("a 20-byte oid cannot join a 32-byte table");
assert!(format!("{err}").contains("32-byte oids"), "{err}");
}
/// A torn tail is truncated away rather than left mid-stream, where every
/// later batch would be unreachable.
#[test]
fn a_torn_tail_is_dropped_and_the_table_keeps_working() {
let p = tmp("torn");
{
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
t.explode(&oid(1), GitObjectKind::Blob, b"complete").unwrap();
t.flush().unwrap();
}
// Half a message, as a killed process leaves.
{
let mut f = OpenOptions::new().append(true).open(&p).unwrap();
f.write_all(&[0xFF, 0xFF, 0xFF, 0xFF, 0x40, 0x00, 0x00, 0x00, 1, 2, 3])
.unwrap();
}
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
assert_eq!(t.rows().unwrap(), 1, "the torn message contributes nothing");
t.explode(&oid(2), GitObjectKind::Blob, b"after").unwrap();
t.flush().unwrap();
assert_eq!(t.rows().unwrap(), 2);
assert_eq!(
t.content(&oid(2)).unwrap().unwrap().1,
b"after".to_vec(),
"the append landed at the last good boundary, not after the garbage"
);
}
#[test]
fn graph_keeps_commits_and_trees_and_drops_blobs() {
let p = ExplodePolicy::Graph;
assert!(p.wants(GitObjectKind::Commit));
assert!(p.wants(GitObjectKind::Tree));
assert!(!p.wants(GitObjectKind::Blob));
assert!(!ExplodePolicy::Off.wants(GitObjectKind::Commit));
assert!(ExplodePolicy::Full.wants(GitObjectKind::Blob));
assert!(ExplodePolicy::Commits.wants(GitObjectKind::Commit));
assert!(!ExplodePolicy::Commits.wants(GitObjectKind::Tree));
assert_eq!(
ExplodePolicy::default(),
ExplodePolicy::Commits,
"the shipped default is the lane Q one: commits and tags eager, trees \
and blobs lazy and budgeted. `Full` is what put 22.4x the pack into \
/data on the rig; `Graph` is what put 28.24 GB of tree payloads there \
and 50.9 GiB of them in RAM."
);
}
/// ★ THE SETTING: a declined object is not stored, and is COUNTED — so a
/// table short by policy and one short by fault do not look alike.
#[test]
fn a_declined_object_is_not_stored_and_is_counted() {
let t = ExplodedArchive::open_with_policy(&tmp("policy"), ExplodePolicy::Graph).unwrap();
t.explode(&oid(1), GitObjectKind::Commit, b"c").unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"bbbb").unwrap();
t.flush().unwrap();
assert!(t.has(&oid(1)).unwrap());
assert!(!t.has(&oid(2)).unwrap());
assert_eq!(t.skipped(), 1);
assert_eq!(t.rows().unwrap(), 1);
}
/// `Off` writes no file at all — the whole point of the free tier.
#[test]
fn off_stores_nothing_and_leaves_no_file() {
let p = tmp("off");
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Off).unwrap();
t.explode(&oid(1), GitObjectKind::Commit, b"c").unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"b").unwrap();
assert_eq!(t.flush().unwrap(), 0);
assert_eq!(t.rows().unwrap(), 0);
assert_eq!(t.skipped(), 2);
assert!(!p.exists(), "a disabled table must not create a file");
}
// ════════════════════════════════════════════════════════════════════════
// The stree in front of the lookup
// ════════════════════════════════════════════════════════════════════════
/// `n` oids in `groups` of `per` that share their first EIGHT bytes and
/// differ only after them — the collision is CONSTRUCTED, never hoped for.
///
/// The prefixes deliberately straddle `0x80`: half the oids in a real repo
/// start there, and they are the half a non-order-preserving key would sort
/// onto the wrong side of the keyspace.
fn colliding_oids(groups: usize, per: usize) -> Vec<[u8; 20]> {
let mut out = Vec::with_capacity(groups * per);
for g in 0..groups {
let mut z = (g as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15);
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
let prefix = (z ^ (z >> 31)).to_be_bytes();
for k in 0..per {
let mut o = [0u8; 20];
o[..8].copy_from_slice(&prefix);
o[8] = k as u8;
o[9..13].copy_from_slice(&(g as u32).to_be_bytes());
out.push(o);
}
}
out
}
/// ★ THE STREE RETURNS THE ROW THE BINARY SEARCH RETURNED — including on
/// 8-byte prefix collisions, of which this table has 1000.
///
/// Three answers are compared for every oid: the `stree` probe, the
/// `binary_search` over the identical oid array that this file shipped with,
/// and the **applied output** — the payload `content` hands back, which is
/// unique per oid, so a wrong row is a wrong payload and not a passing
/// assertion about a `Located` nobody read.
///
/// Seen RED by deleting the full-oid comparison from `OidTree::verify` (i.e.
/// `.find(|&i| self.oid_at(i) == oid)` → `.next()`, the "stree already found
/// it" mistake): **"the stree and the binary search disagree at
/// 0000000000000000010000000000000000000000: Some(Located { batch: 0, row: 0,
/// kind: Blob }) vs Some(Located { batch: 0, row: 1, kind: Blob })"** — the
/// probe was handed the row of the prefix-mate stored one slot earlier.
#[test]
fn the_stree_agrees_with_the_binary_search_it_replaced_on_colliding_prefixes() {
let t = ExplodedArchive::open_with_policy(&tmp("stree-agree"), ExplodePolicy::Full).unwrap();
let oids = colliding_oids(1000, 4);
for (i, o) in oids.iter().enumerate() {
t.explode(o, GitObjectKind::Blob, format!("payload-{i}").as_bytes())
.unwrap();
}
t.flush().unwrap();
// Force the index, then reach into it: the two finders must sit over the
// SAME arrays, or this compares two indexes rather than two finders.
t.index_now().unwrap();
let idx = t.index.lock().unwrap();
let tree = idx.as_ref().expect("index_now built it");
assert_eq!(tree.len(), oids.len());
// The premise: the collisions are real in the built tree.
let mut colliding_runs = 0usize;
for o in &oids {
let run = tree.candidate_run(key_for_oid(o));
assert_eq!(
run.len(),
4,
"expected a 4-entry candidate run for {}, got {run:?} — the collision \
the rest of this test rests on is not there",
hex::encode(o)
);
colliding_runs += 1;
}
assert_eq!(colliding_runs, oids.len());
// And the sign boundary is really crossed, or the key's top-bit flip is
// untested by this corpus.
assert!(
oids.iter().any(|o| o[0] >= 0x80) && oids.iter().any(|o| o[0] < 0x80),
"premise: the prefixes must straddle 0x80"
);
for o in &oids {
assert_eq!(
tree.find(o),
tree.find_by_binary_search(o),
"the stree and the binary search disagree at {}: {:?} vs {:?}",
hex::encode(o),
tree.find(o),
tree.find_by_binary_search(o)
);
assert!(tree.find(o).is_some(), "{} vanished", hex::encode(o));
}
// An oid on a colliding prefix that was never stored must MISS, in both
// finders — this is the query a verify-less probe answers with somebody
// else's row.
for o in oids.iter().step_by(4) {
let mut absent = *o;
absent[8] = 0xff;
assert_eq!(tree.find(&absent), None, "{} was never stored", hex::encode(absent));
assert_eq!(tree.find_by_binary_search(&absent), None);
}
drop(idx);
// Applied output: each oid's own payload, not its prefix-mate's.
for (i, o) in oids.iter().enumerate() {
assert_eq!(
t.content(o).unwrap().unwrap().1,
format!("payload-{i}").into_bytes(),
"{} came back with another object's bytes",
hex::encode(o)
);
}
}
/// ★ A TOMBSTONE STILL RETIRES AN OID WITH THE STREE IN FRONT — and its
/// prefix-mates do not go with it, nor stand in for it.
///
/// Two failure modes live here and both return a wrong answer rather than an
/// error: a fold that keeps the *first* sighting rather than the last leaves
/// the retired oid resolvable, and a probe that trusts the 8-byte key answers
/// the retired oid with the row of the live oid it collides with.
///
/// Seen RED three times:
///
/// * `Keep::Last` → `Keep::First` in `index_now`: **"a reopen resurrected a
/// retired oid: 0000000000000000000000000000000000000000"**. The same break
/// also reds the older, coarser
/// `retain_drops_only_what_is_dead_and_the_drop_survives_a_reopen`.
/// * the full-oid comparison dropped from `OidTree::verify`:
/// **"0000000000000000000000000000000000000000 was retired"** — the dead
/// oid answered `has` with its live prefix-mate's row, in-process, before
/// any reopen.
/// * the incremental fold in `flush` no longer filtering the retired set
/// (`if !retired.contains(oid)` → `if true`): the same message, from the
/// live process rather than from a reopen.
#[test]
fn a_tombstoned_oid_stays_absent_when_a_live_oid_shares_its_prefix() {
let p = tmp("stree-tomb");
// Four oids per prefix: one is retired, three stay.
let oids = colliding_oids(64, 4);
let dead: Vec<[u8; 20]> = oids.iter().copied().step_by(4).collect();
{
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
for (i, o) in oids.iter().enumerate() {
t.explode(o, GitObjectKind::Blob, format!("live-{i}").as_bytes())
.unwrap();
}
t.flush().unwrap();
let retired = t.retain(&|o: &[u8]| !dead.iter().any(|d| d == o)).unwrap();
assert_eq!(retired, dead.len() as u64);
for d in &dead {
assert!(!t.has(d).unwrap(), "{} was retired", hex::encode(d));
assert!(
t.content(d).unwrap().is_none(),
"the retired oid {} was answered with a prefix-mate's row",
hex::encode(d)
);
}
for (i, o) in oids.iter().enumerate() {
if i % 4 == 0 {
continue;
}
assert_eq!(
t.content(o).unwrap().unwrap().1,
format!("live-{i}").into_bytes(),
"{} is alive and must still answer with its OWN bytes",
hex::encode(o)
);
}
}
// ★ And a reopen — which rebuilds the tree from the file, tombstones and
// all — finds every live row and none of the dead.
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
for (i, o) in oids.iter().enumerate() {
if i % 4 == 0 {
assert!(
!t.has(o).unwrap(),
"a reopen resurrected a retired oid: {}",
hex::encode(o)
);
} else {
assert_eq!(
t.content(o).unwrap().unwrap().1,
format!("live-{i}").into_bytes(),
"a reopen lost or misrouted the live oid {}",
hex::encode(o)
);
}
}
assert_eq!(
t.of_kind(GitObjectKind::Blob).unwrap().len(),
oids.len() - dead.len()
);
// Only now — `rows()` answers from batch metadata until an index exists,
// and the file still holds the retired rows and their tombstones. This
// assertion measured 320 against 192 when it was made before the reads
// above, which is the documented behaviour of `rows()` and not the
// tombstone leaking.
assert_eq!(t.rows().unwrap(), (oids.len() - dead.len()) as u64);
}
/// The incremental extension in `flush` and the from-scratch rebuild in
/// `index_now` must land on the same tree — one of them runs while a push is
/// hot and the other after a restart, and a difference between them is a
/// bug that only appears to the second process.
///
/// Seen RED twice, once from each side, and the two reds are mirror images —
/// which is what tells you the test compares the folds rather than one of
/// them against itself:
///
/// * incremental path drops the `retired` filter over the entries it carries
/// forward (`if !retired.contains(oid)` → `if true`): **"the two folds kept
/// a different number of oids: left: 150, right: 200"** — the live index
/// kept the 50 it had been told to retire.
/// * `Keep::Last` → `Keep::First` in `index_now`: **"left: 200, right: 150"**
/// — the rebuild kept them instead.
#[test]
fn the_incremental_index_and_a_rebuilt_one_agree() {
let p = tmp("stree-incr");
let oids = colliding_oids(50, 4);
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
// Build the index FIRST, so every flush below takes the incremental path.
t.index_now().unwrap();
for (i, o) in oids.iter().enumerate() {
t.explode(o, GitObjectKind::Blob, format!("v-{i}").as_bytes())
.unwrap();
if i % 37 == 0 {
t.flush().unwrap();
}
}
t.flush().unwrap();
t.retain(&|o: &[u8]| o[8] != 3).unwrap();
let live: Vec<(Vec<u8>, Vec<u8>)> = {
let idx = t.index.lock().unwrap();
idx.as_ref()
.unwrap()
.iter()
.map(|(o, l)| (o.to_vec(), vec![l.batch as u8, l.row as u8]))
.collect()
};
drop(t);
let again = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
again.index_now().unwrap();
let rebuilt = again.index.lock().unwrap();
let rebuilt = rebuilt.as_ref().unwrap();
assert_eq!(rebuilt.len(), live.len(), "the two folds kept a different number of oids");
for (o, _) in &live {
assert!(
rebuilt.find(o).is_some(),
"the live process and a reopen disagree about {}: Some(..) vs None",
hex::encode(o)
);
}
for o in oids.iter().filter(|o| o[8] == 3) {
assert!(rebuilt.find(o).is_none(), "{} was retired", hex::encode(o));
}
}
/// The measurement behind the change, re-takeable: three finders over the
/// SAME 1 000 000 entries — the `Vec<(Vec<u8>, Located)>` this file shipped
/// with, a binary search over the flat array, and the stree.
///
/// `--ignored` because it is a measurement, not a guard: it asserts only that
/// the three agree, and prints the times. Run it with
/// `cargo test --release -p znippy-plugin-git --lib exploded_arrow -- --ignored --nocapture`
/// — a debug build measures the borrow checker, not the machine.
///
/// ## Measured, oden 2026-08-14, `--release`, 1 000 000 entries, 200 000
/// probes (half misses), three runs
///
/// | arm | ns/probe | vs shipped |
/// |---|---:|---:|
/// | `Vec<(Vec<u8>, Located)>` + `binary_search` (what shipped) | 836 / 862 / 961 | 1.00 |
/// | flat oid array + `binary_search` | 339 / 339 / 348 | 0.40 |
/// | **stree** | **164 / 175 / 185** | **0.21** |
///
/// So the structure is worth **~2×** on its own and the flattening another
/// **~2.5×**, and the two together take a point lookup to a fifth of what it
/// cost. **The box was NOT quiet** — `/proc/loadavg` 1-min 29.7–31.5 on 32
/// cores, other agents building throughout — so the absolute figures are
/// upper bounds. The ratios are reported instead, and each arm's own spread
/// across the three runs (14% / 2.7% / 12%) is far narrower than the 5×
/// and 2× between them.
#[test]
#[ignore = "measurement; run under --release with --ignored"]
fn lookup_cost_before_and_after() {
use std::time::Instant;
let n = 1_000_000usize;
// 4 oids per prefix, so ~25% of probes land in a run of four and the
// verify step is exercised rather than skipped.
let oids = colliding_oids(n / 4, 4);
let mut b = OidTreeBuilder::with_capacity(oids.len());
for (i, o) in oids.iter().enumerate() {
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 }))
.unwrap();
}
let build = Instant::now();
let tree = b.finish(Keep::Last);
let build = build.elapsed();
// The shape this replaced: one heap allocation per oid.
let mut old: Vec<(Vec<u8>, Located)> =
tree.iter().map(|(o, l)| (o.to_vec(), l)).collect();
old.sort_by(|a, b| a.0.cmp(&b.0));
// A miss-heavy probe order, and NOT in oid order — a `have` negotiation
// is random and half of it misses.
let mut probes: Vec<[u8; 20]> = Vec::with_capacity(200_000);
for i in (0..oids.len()).step_by(oids.len() / 100_000) {
probes.push(oids[i]);
let mut m = oids[i];
m[8] = 0xfe;
probes.push(m);
}
let mut sink = 0u64;
let t0 = Instant::now();
for p in &probes {
if let Ok(i) = old.binary_search_by(|(o, _)| o.as_slice().cmp(&p[..])) {
sink += old[i].1.row as u64;
}
}
let vec_of_vec = t0.elapsed();
let t0 = Instant::now();
for p in &probes {
if let Some(l) = tree.find_by_binary_search(p) {
sink += l.row as u64;
}
}
let flat_bsearch = t0.elapsed();
let t0 = Instant::now();
for p in &probes {
if let Some(l) = tree.find(p) {
sink += l.row as u64;
}
}
let stree = t0.elapsed();
// The arms must be the same question.
for p in &probes {
let want = tree.find_by_binary_search(p);
assert_eq!(tree.find(p), want, "arms disagree at {}", hex::encode(p));
}
let per = |d: std::time::Duration| d.as_secs_f64() * 1e9 / probes.len() as f64;
println!(
"exploded oid lookup, {} entries ({} groups of 4 colliding prefixes), \
{} probes half of them misses:\n \
Vec<(Vec<u8>,Located)> binary_search {:>7.1} ns/probe\n \
flat array binary_search {:>7.1} ns/probe\n \
stree {:>7.1} ns/probe\n \
tree build (sort + fold + stree) {:.3} s sink={sink}",
tree.len(),
oids.len() / 4,
probes.len(),
per(vec_of_vec),
per(flat_bsearch),
per(stree),
build.as_secs_f64(),
);
}
#[test]
fn an_unknown_policy_is_refused() {
assert_eq!(ExplodePolicy::parse("full").unwrap(), ExplodePolicy::Full);
let err = ExplodePolicy::parse("ful").expect_err("a typo must be refused");
assert!(format!("{err}").contains("unknown explode policy"), "{err}");
}
}