plugins/native/znippy-plugin-git/src/read_stack.rs
//! **One read stack, not three components.**
//!
//! A git object lookup in this crate goes through a single stack with three
//! surfaces, and the whole point of putting them in one type is that they cannot
//! be reasoned about — or drift — separately:
//!
//! 1. **The Ragnar `stree` is the fast path.** It is not a cache in front of an
//! index; it *is* the index's key surface, [`crate::oid_index`]'s 8-byte-prefix
//! static B-tree, reached through whichever [`ObjectIndex`] arm the stack was
//! parameterised with. No third oid index is written here (LAW 5).
//! 2. **The Arrow tables are what it points into.** The stree resolves an oid to
//! an ordinal; the ordinal addresses the five facts in the Arrow IPC columns
//! of [`FourTables`](crate::index_layout::FourTables) or
//! [`OneTableFourColumns`]. Ordinal and column live in the same object and are
//! replaced together.
//! 3. **redb is what it falls through to, and redb always answers.** Every object
//! ever appended is in redb. The stree+Arrow surface is a *projection* of a
//! prefix of it, rebuilt on a trigger. So a stree miss is never an answer — it
//! is a routing decision, and the row is fetched from redb. The single
//! exception is not an exception to the rule but a proof of it: when the
//! projection covers **every** row, "the stree missed" and "the repository
//! does not have it" are the same statement, and the trip is skipped. See
//! [`ObjectReadStack::projection_is_complete`].
//!
//! This is [`skade`'s arrangement](../../../../skade/src/static_index.rs) copied
//! rather than reinvented: `StaticIndex` (STree64 over snapshot ids) in front,
//! the redb commit log behind, `resolve_many` doing one pipelined tree pass and
//! then one redb transaction for the slots the tree did not fill. The names
//! differ because the keys differ (oids, not snapshot ids); the shape does not.
//!
//! ## Scope: one stack per REPOSITORY
//!
//! Not per account. A git negotiation — `have`/`want`, the push connectivity
//! check — is scoped to one repository, and an oid means nothing outside the
//! repository that stores it. [`ObjectReadStack::open`] therefore takes the path
//! of one repository's tail database.
//!
//! ## Why the projection may be incomplete and can still be trusted
//!
//! **The archive is append-only.** An object's five facts are written once and
//! never rewritten — [`ObjectReadStack::append`] *refuses* a rewrite that would
//! change them rather than accepting it (that refusal is what makes the rest of
//! this true, and it is asserted in
//! [`tests::an_append_only_violation_is_refused`]). Given that:
//!
//! * a **hit** in the stree is always valid — the row it points at cannot have
//! become stale, because nothing may change it;
//! * a **miss** carries no information at all — it means "not in this
//! projection", which is not the same as "not in this repository";
//! * so the projection is allowed to lag, and the only cost of lag is the redb
//! fall-through on the rows it has not absorbed yet.
//!
//! That asymmetry is the entire reason a *trigger* is sound where a
//! rebuild-per-write would be ruinous.
//!
//! ## The rebuild trigger
//!
//! Two triggers, both in [`RebuildTriggers`], either one fires:
//!
//! * **misses** — lookups the stree missed **and redb answered**, against a
//! threshold relative to the size of the projection. Genuine absences are
//! counted separately in `absent` and deliberately do NOT count: a `have`
//! negotiation is mostly misses, and if those drove the trigger, a busy
//! read-only repository would rebuild for ever without a single new object
//! having arrived.
//! * **volume** — stored object bytes appended since the last rebuild.
//!
//! Defaults, and the measurements they were derived from, are on
//! [`RebuildTriggers::DEFAULT_TAIL_HITS_PER_ROW`],
//! [`RebuildTriggers::DEFAULT_MIN_TAIL_HITS`] and
//! [`RebuildTriggers::DEFAULT_TAIL_BYTES`].
//!
//! ## Threads
//!
//! There are none here. A rebuild runs inline on the [`append`](ObjectReadStack::append)
//! that trips the trigger, or on an explicit [`rebuild`](ObjectReadStack::rebuild).
//! skade rebuilds on a detached thread; the sanctioned home for a detached thread
//! in this constellation is `gatling::background::Job` (LAW 3 — rayon is banned
//! and so is a hand-rolled pool), and `Job` is join-required with no completion
//! probe, so "in the background" would mean picking an arbitrary later call to
//! block on. Inline on a trigger is deterministic and, with the defaults below,
//! rare. An owner who wants the overlap can drive [`rebuild`](ObjectReadStack::rebuild)
//! from a `Job` itself — it takes `&self` and swaps under an `RwLock`.
use std::fs::{self, File};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use anyhow::{Context, Result, anyhow, bail};
use redb::{Database, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition};
use znippy_common::arrow::buffer::Buffer;
use crate::index_layout::{IndexEntry, IndexRow, ObjType, ObjectIndex, OneTableFourColumns};
// ── the durable tail ──────────────────────────────────────────────────────────
/// `oid → packed row`. The key is the raw oid (20 or 32 bytes), so redb's own
/// B-tree order is oid-lexicographic order — the same order the stree and the
/// Arrow columns are in, which makes a rebuild a straight ordered scan.
const OBJECTS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("objects");
/// Small counters. `arrival_seq` is the only one that must survive a reopen: it
/// is what gives a tail row an ordinal that no other tail row shares.
const META: TableDefinition<&str, u64> = TableDefinition::new("meta");
const META_ARRIVAL_SEQ: &str = "arrival_seq";
/// A durable, monotonically increasing marker of the OBJECTS table's **content
/// state**: bumped by one, inside the very same redb write transaction, whenever
/// a commit changes the set of oids — an [`ObjectReadStack::append`] that inserts
/// at least one new row, or a [`ObjectReadStack::retain`] that drops at least one.
///
/// It was introduced as the validity key of the `objects.ordinal` memo (S-082),
/// which is gone since 2026-09-04 — the persisted index's oid column is the
/// ordinal space now, current by construction. The epoch stays: it is the one
/// cheap, sound, crash-consistent "has the oid set changed" signal this store
/// has. `arrival_seq` alone will not do — it counts inserts and is blind to
/// `retain`'s removals — and a blake3 over the oid set is a full scan. Because
/// the bump shares the mutation's transaction, `epoch` names the OBJECTS content
/// uniquely across this store's life: equal epochs ⟹ an unchanged set. See
/// [`ObjectReadStack::tail_epoch_and_len`].
const META_TAIL_EPOCH: &str = "tail_epoch";
/// **The layout [`encode_row`] wrote these rows with.**
///
/// [`decode_row`] refuses a row whose width is not [`TAIL_ROW_BYTES`] — on
/// purpose, because reinterpreting an unknown layout is how a wrong offset gets
/// into an index and is then queried and believed. But a refusal with nothing
/// behind it is a store that has stopped answering, and until this key existed
/// that is exactly what a layout change produced: every read `bail`ing, with no
/// way back short of deleting the repository.
///
/// There is a way back, and it costs nothing to take, because **the tail is a
/// cache** (see [`decode_row`]) and the store already knows how to rebuild one:
/// `Absorber::adopt_journal` derives what is unabsorbed as *"its extent is in
/// the journal and its rows are not in the index"* and requeues it. Empty the
/// OBJECTS table and every pack is unabsorbed by that definition, so the drain
/// re-indexes the archive and reads fall back to the pack meanwhile — **slower,
/// never wrong**, which is the same path a crash mid-absorb already takes.
///
/// So a version mismatch is a **drop and rebuild**, not a failure. Absent is
/// treated as [`TAIL_FORMAT`] rather than as "old": every tail in existence when
/// this key was introduced was written by the layout it names, so stamping them
/// silently is correct and costs no rebuild. The next change to the row bumps
/// the constant and those stores rebuild once.
const META_TAIL_FORMAT: &str = "tail_format";
/// The current value of [`META_TAIL_FORMAT`]. **Bump this in the same commit as
/// any change to [`TAIL_ROW_BYTES`] or to what [`encode_row`] puts in it.**
///
/// `1` is the layout in use since the `delta_base` field landed (PLAN §13) — the
/// 41-byte row this constant was introduced beside, not a new one.
const TAIL_FORMAT: u64 = 1;
/// `seq | offset | len | type | uncompressed_size | delta_base`, little-endian,
/// fixed width.
///
/// ```text
/// 0 u64 seq (arrival, not a fact about the object)
/// 8 u64 offset
/// 16 u64 len
/// 24 u8 object type code
/// 25 u64 uncompressed_size
/// 33 u64 delta_base — archive offset of the base entry, 0 for none
/// ══
/// 41
/// ```
///
/// **This was 33 bytes before `delta_base` (PLAN §13).** A tail written by the
/// older encoding is refused by [`decode_row`] on its length rather than
/// misread: the tail is a cache the whole of §14 permits dropping and
/// rebuilding, so a hard refusal is the right failure and a silent
/// reinterpretation is not.
const TAIL_ROW_BYTES: usize = 41;
/// Set on the ordinal of every row that came from the tail rather than from the
/// Arrow projection.
///
/// An ordinal is a **row address within one generation**, not an identity — the
/// oid is the identity. A rebuild re-derives every ordinal as the new
/// oid-lexicographic rank, so an ordinal held across a rebuild is meaningless.
/// The high bit makes the dangerous version of that mistake impossible instead
/// of merely documented: a tail ordinal used to index an Arrow column is ≥ 2^31
/// and therefore out of bounds — a panic — rather than a silently wrong row.
pub const TAIL_ORDINAL_BIT: u32 = 0x8000_0000;
/// True when `row` was answered by the redb tail rather than by the projection.
pub fn is_tail_row(row: &IndexRow) -> bool {
is_tail_ordinal(row.ordinal)
}
/// The same question of a bare ordinal.
#[inline]
pub fn is_tail_ordinal(ordinal: u32) -> bool {
ordinal & TAIL_ORDINAL_BIT != 0
}
fn tail_ordinal(seq: u64) -> u32 {
TAIL_ORDINAL_BIT | (seq as u32 & !TAIL_ORDINAL_BIT)
}
fn encode_row(seq: u64, e: &IndexEntry) -> [u8; TAIL_ROW_BYTES] {
let mut b = [0u8; TAIL_ROW_BYTES];
b[0..8].copy_from_slice(&seq.to_le_bytes());
b[8..16].copy_from_slice(&e.offset.to_le_bytes());
b[16..24].copy_from_slice(&e.len.to_le_bytes());
b[24] = e.obj_type.code();
b[25..33].copy_from_slice(&e.uncompressed_size.to_le_bytes());
b[33..41].copy_from_slice(&e.delta_base.to_le_bytes());
b
}
/// One decoded tail row. `seq` is the arrival sequence, not a fact about the
/// object.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct TailRow {
seq: u64,
offset: u64,
len: u64,
obj_type: ObjType,
uncompressed_size: u64,
/// Archive offset of the base entry, `0` for none. An offset and not an
/// ordinal precisely because of this table: a tail ordinal is
/// [`TAIL_ORDINAL_BIT`]-tagged and a rebuild re-derives every ordinal, so an
/// ordinal stored here would be meaningless one rebuild later.
delta_base: u64,
}
fn decode_row(b: &[u8]) -> Result<TailRow> {
if b.len() != TAIL_ROW_BYTES {
bail!("tail row is {} bytes, expected {TAIL_ROW_BYTES}", b.len());
}
let u64_at = |i: usize| {
let mut w = [0u8; 8];
w.copy_from_slice(&b[i..i + 8]);
u64::from_le_bytes(w)
};
Ok(TailRow {
seq: u64_at(0),
offset: u64_at(8),
len: u64_at(16),
// A code the writer cannot produce must not be invented on the way out:
// a wrong type in an index is worse than a failure, because it is
// queried and believed.
obj_type: ObjType::from_code(b[24])
.ok_or_else(|| anyhow!("tail row carries object type code {}", b[24]))?,
uncompressed_size: u64_at(25),
delta_base: u64_at(33),
})
}
impl TailRow {
fn as_index_row(&self) -> IndexRow {
IndexRow {
ordinal: tail_ordinal(self.seq),
offset: self.offset,
len: self.len,
obj_type: self.obj_type,
uncompressed_size: self.uncompressed_size,
delta_base: self.delta_base,
}
}
/// The same facts as an [`IndexEntry`], for the rebuild scan.
fn as_entry(&self, oid: &[u8]) -> IndexEntry {
IndexEntry {
oid: oid.to_vec(),
offset: self.offset,
len: self.len,
obj_type: self.obj_type,
uncompressed_size: self.uncompressed_size,
delta_base: self.delta_base,
}
}
}
// ── the trigger ───────────────────────────────────────────────────────────────
/// When the Arrow/stree projection is rebuilt from the redb tail. Either
/// trigger fires; both are per repository and per generation.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct RebuildTriggers {
/// The miss trigger, as a multiple of the rows already in the projection.
/// See [`DEFAULT_TAIL_HITS_PER_ROW`](Self::DEFAULT_TAIL_HITS_PER_ROW) for why
/// it is relative and not a constant.
pub tail_hits_per_row: f64,
/// Floor and off-switch for the miss trigger: the threshold is never below
/// this, and **`0` disables the miss trigger entirely** whatever
/// `tail_hits_per_row` says.
pub min_tail_hits: u64,
/// Rebuild once this many stored object bytes have been appended since the
/// last rebuild. `0` disables the volume trigger. On a persisting stack this
/// is the **floor** of the volume threshold; see `fold_ratio`.
pub tail_bytes: u64,
/// **On a persisting stack, the volume threshold scales with the index**:
/// a fold fires once the object bytes appended since the last fold reach
/// `fold_ratio ×` the persisted index's own size (never below `tail_bytes`).
/// `0.0` leaves the fixed floor alone. See
/// [`DEFAULT_FOLD_RATIO`](Self::DEFAULT_FOLD_RATIO) for why the value is 1.
/// Ignored on a stack that does not persist — there a rebuild writes nothing
/// and the fixed 64 MiB is the right cadence.
pub fold_ratio: f64,
}
impl RebuildTriggers {
/// **1.0 — one tail-served lookup per row in the projection.**
///
/// The miss trigger is relative because the two costs it balances scale
/// differently, and a single constant is therefore wrong at one end or the
/// other. Both costs MEASURED on oden 2026-08-07, 1-min loadavg 2.1–2.2,
/// `examples/read_stack_bench.rs`, sha1 oids, batch 1000, 3 runs per cell,
/// worst run-to-run spread 24.9%:
///
/// | | 100 000 objects | 1 000 000 objects |
/// |---|---:|---:|
/// | lookup the projection answers | 80 ns | 221 ns |
/// | lookup the tail answers | 463 ns | 937 ns |
/// | **the fall-through costs** | **383 ns** | **716 ns** |
/// | rebuild (ordered tail scan + projection build) | 40.9 ms | 547.9 ms |
/// | **per row in the repository** | **409 ns** | **548 ns** |
/// | break-even, tail-served lookups per row | **1.07** | **0.77** |
///
/// The rebuild has paid for itself once the fall-through has carried roughly
/// one lookup per row, at both sizes and an order of magnitude apart — which
/// is why the ratio is the right shape for this trigger and **1.0** is the
/// right value in it. A fixed 4096 would rebuild a million-object repository
/// (0.55 s) to save 4096 × 716 ns ≈ 2.9 ms — 190× the wrong way.
///
/// Re-measure if the tail engine or the projection build changes: this is a
/// ratio of two measured costs and nothing else.
pub const DEFAULT_TAIL_HITS_PER_ROW: f64 = 1.0;
/// **4096 — the floor under the relative threshold.**
///
/// On an empty or nearly-empty projection the ratio above is ~0 and would
/// rebuild on the first miss, over and over, during exactly the period when
/// objects are still arriving. 4096 misses is ~1.6 ms of fall-through at the
/// measured 100 000-object price — cheap enough to be worth waiting for on
/// any repository, and enough that a burst of small pushes coalesces into
/// one rebuild.
pub const DEFAULT_MIN_TAIL_HITS: u64 = 4096;
/// **64 MiB of appended object bytes.**
///
/// A git push is a packfile, and the pack is the unit that lands in the tail.
/// 64 MiB is comfortably more than one ordinary push and less than a big
/// one, so a busy repository rebuilds on the order of once per large push
/// rather than once per push — while a repository taking a 2 GiB initial
/// import rebuilds ~32 times over that import instead of once at the end,
/// which is what keeps the fall-through from carrying the whole import.
/// It is the trigger that carries the normal case; the miss trigger is the
/// backstop for a repository that is read hard and written rarely.
pub const DEFAULT_TAIL_BYTES: u64 = 64 * 1024 * 1024;
/// **1.0 — fold when the bytes appended since the last fold equal the
/// index's own size.**
///
/// Since 2026-09-04 a rebuild on a persisting stack WRITES the projection
/// (`objects.index`), so the cadence above stopped being free. MEASURED on
/// the kernel mirror with the fixed 64 MiB floor alone: an 8.72 GB import
/// folded ≈136 times, each rewriting the whole growing file — ≈51 GB of
/// sequential writes for a 747 MB index (INFERRED from the arithmetic; ≥18
/// distinct file sizes OBSERVED at 10 s sampling). The ack did not slow on
/// oden's NVMe; on the appliance's disks it would.
///
/// A ratio of 1 gives a bound that is easy to say and holds by construction:
/// **the fold's total writes never exceed the push's own bytes.** Each fold
/// writes an index roughly the size of the bytes that triggered it, so the
/// sum of index writes over an import is ≈ the bytes imported — ~8.7 GB for
/// the kernel instead of ~51, and the number of folds falls from ~136 to
/// O(log n). The price is the redb tail's high-water mark: with the index at
/// 53.5 B/object and pack bytes at ~620 B/object, "appended bytes = index
/// bytes" is ≈ 8.6 % of the folded rows in redb before a fold, i.e. ~1.2 M
/// rows ≈ 250 MB of redb rows on the kernel mirror (redb allocates in
/// doublings, so a ~512 MiB file) against 270 MB with the floor alone.
/// Re-measure on the rig after changing either number; both are in
/// `.nornir/storage-levers-2026-09-03.md`.
pub const DEFAULT_FOLD_RATIO: f64 = 1.0;
/// The miss threshold for a projection of `projection_rows` rows, or `None`
/// when the miss trigger is off.
pub fn miss_threshold(&self, projection_rows: u64) -> Option<u64> {
if self.min_tail_hits == 0 {
return None;
}
let scaled = (self.tail_hits_per_row * projection_rows as f64) as u64;
Some(scaled.max(self.min_tail_hits))
}
/// Both triggers off: the projection is rebuilt only when asked.
pub fn manual() -> Self {
Self {
tail_hits_per_row: 0.0,
min_tail_hits: 0,
tail_bytes: 0,
fold_ratio: 0.0,
}
}
}
impl Default for RebuildTriggers {
fn default() -> Self {
Self {
tail_hits_per_row: Self::DEFAULT_TAIL_HITS_PER_ROW,
fold_ratio: Self::DEFAULT_FOLD_RATIO,
min_tail_hits: Self::DEFAULT_MIN_TAIL_HITS,
tail_bytes: Self::DEFAULT_TAIL_BYTES,
}
}
}
/// Which threshold fired, with the value that fired it. Returned rather than
/// logged, so a caller can record *why* a rebuild happened.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RebuildReason {
/// `n` lookups had to be answered by the tail.
TailHits(u64),
/// `n` bytes of objects were appended since the last rebuild.
TailBytes(u64),
/// [`ObjectReadStack::rebuild`] was called directly.
Explicit,
}
/// What [`ObjectReadStack::oids_raw`] hands back: the ordinal space and its
/// provenance.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OrdinalSpace {
/// Every oid, fixed width, back to back, in oid order.
pub raw: Vec<u8>,
/// Oids that came straight off the persisted projection's column.
pub folded_rows: u64,
/// Oids that came off a scan of the redb tail — the rows since the last
/// fold, or every row on a stack that does not persist.
pub tail_rows: u64,
}
/// A snapshot of the stack's counters. Everything here is applied output — rows
/// that exist, lookups that happened — not configuration echoed back.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct StackStats {
/// Objects in the Arrow/stree projection.
pub sealed_rows: u64,
/// Objects in the repository. Always ≥ `sealed_rows`; the difference is the
/// un-absorbed tail.
pub total_rows: u64,
/// Lookups the stree missed and redb answered, since the last rebuild.
pub tail_hits: u64,
/// Lookups nothing answered — the object is not in this repository. Never
/// drives the trigger.
pub absent: u64,
/// Rows the projection does not cover yet. **Zero means the projection *is*
/// the truth**, which is what lets a miss be answered without a redb round
/// trip at all.
pub unabsorbed_rows: u64,
/// redb read transactions opened by the read path since the last rebuild.
/// The number that shows whether the completeness fast path is actually
/// being taken; there is no other way to see it from outside.
pub tail_txns: u64,
/// Object bytes appended since the last rebuild.
pub tail_bytes: u64,
/// Projections built over this stack's lifetime, including the one at open.
pub rebuilds: u64,
/// Folds that persisted the projection and emptied redb of what it holds.
/// `0` on an in-memory stack and on an arm that cannot be persisted.
pub folds: u64,
/// `1` when this stack's open mapped `objects.index` and skipped the redb
/// scan-and-build; `0` when it had to rebuild. The applied-output proof
/// that a warm open is warm.
pub persisted_loads: u64,
/// Bumped by every rebuild. An [`IndexRow::ordinal`] is only meaningful
/// within one generation.
pub generation: u64,
}
// ── the stack ─────────────────────────────────────────────────────────────────
/// The read stack: Ragnar stree → Arrow columns → redb, in one object.
///
/// Parameterised by the Arrow layout so it drops into
/// `examples/index_layout_bench.rs` against
/// [`FourTables`](crate::index_layout::FourTables) and [`OneTableFourColumns`]
/// unchanged — it implements the same [`ObjectIndex`] trait, and
/// [`ObjectIndex::build`] produces a stack whose projection already covers every
/// entry, so on that benchmark it is those arms plus one redb round trip per
/// genuine miss.
pub struct ObjectReadStack<S: ObjectIndex = OneTableFourColumns> {
/// The durable truth. Every object ever appended is here.
tail: Arc<Database>,
/// The projection. Replaced wholesale by a rebuild, never mutated in place —
/// which is what lets a reader hold it briefly without coordinating.
projection: RwLock<Arc<S>>,
triggers: RebuildTriggers,
total_rows: AtomicU64,
tail_hits: AtomicU64,
absent: AtomicU64,
tail_bytes: AtomicU64,
rebuilds: AtomicU64,
generation: AtomicU64,
/// Uncompressed bytes of the rows the projection has **not** absorbed.
///
/// This is what makes [`ObjectIndex::sum_uncompressed`] exact without a scan.
/// The projection is, by definition, every row that existed at the last
/// rebuild; append-only means none of them can have changed since; so the
/// repository's total is the projection's column scan plus the rows appended
/// after it, and those are counted here as they arrive. Reset by a rebuild,
/// which is the moment they stop being un-absorbed.
unabsorbed_size: AtomicU64,
/// The same trick for [`ObjectIndex::count_type`], indexed by
/// [`ObjType::code`] (1–4, 6, 7; slots 0 and 5 stay zero).
unabsorbed_types: [AtomicU64; 8],
/// Rows the projection does not cover. **Zero is the interesting value**:
/// see [`ObjectReadStack::projection_is_complete`].
unabsorbed_rows: AtomicU64,
/// redb read transactions the read path has opened since the last rebuild.
tail_txns: AtomicU64,
/// **Where the projection is persisted** — `objects.index` beside the redb
/// tail — or `None` for an in-memory stack and for an arm whose
/// [`ObjectIndex::PERSISTS`] is false. When `Some`, every rebuild is a
/// **fold**: the projection is written here, then the rows it holds are
/// removed from redb, so redb is the tail in fact and not only in name.
persisted: Option<PathBuf>,
/// Serialises the fold against `append` and `retain`. Held for the whole
/// of a fold, and by `append` around its transaction, so the counters a
/// fold resets cannot be bumped underneath it by a row it did not see —
/// which would leave `unabsorbed_rows` at 0 over a row the projection
/// lacks, and `projection_is_complete` would then answer `absent` for an
/// object the repository holds.
fold: Mutex<()>,
/// Folds that wrote `persisted`, over this stack's lifetime.
folds: AtomicU64,
/// Opens that took the projection off `persisted` instead of rebuilding
/// it from a redb scan. `1` after a warm open, `0` after a cold one.
persisted_loads: AtomicU64,
}
impl<S: ObjectIndex> ObjectReadStack<S> {
/// True when this stack folds: it has somewhere to persist to and an arm
/// that can be persisted.
#[inline]
fn persists(&self) -> bool {
self.persisted.is_some()
}
/// Open (or create) one repository's stack at `tail_path`, warm-starting the
/// projection from everything already in the tail.
/// `cache_bytes` is redb's page-cache ceiling for **this** database. A
/// parameter and not a `getenv` down here, so the whole store costs one
/// environment read at construction and none below it; see
/// [`crate::arms::redb_cache_bytes`], and note that the stock
/// `Database::create` would take redb's 1 GiB default instead.
///
/// A tail written by redb 2 (every store before 2026-10-06) is migrated to
/// redb 4 here, once, before it is opened — see [`migrate_redb2_tail`].
pub fn open(tail_path: &Path, triggers: RebuildTriggers, cache_bytes: usize) -> Result<Self> {
let create = || Database::builder().set_cache_size(cache_bytes).create(tail_path);
let db = match create() {
Err(redb::DatabaseError::UpgradeRequired(found)) => {
migrate_redb2_tail(tail_path, found, cache_bytes)?;
create()
}
other => other,
}
.with_context(|| format!("opening object tail at {}", tail_path.display()))?;
// `objects.tail` → `objects.index`: the persisted projection sits beside
// the redb tail it is folded from. Only for an arm that can be written
// down; the others keep every row in redb, as before 2026-09-04.
let persisted = S::persists().then(|| tail_path.with_extension("index"));
Self::from_db(db, triggers, persisted, Some(tail_path.to_path_buf()))
}
/// Where this stack persists its projection, if it does.
pub fn persisted_path(&self) -> Option<&Path> {
self.persisted.as_deref()
}
/// **The name the built projection reports about itself** — the applied
/// output that says which [`crate::arms::IndexArm`] actually got built.
///
/// Deliberately not [`ObjectIndex::name`] on the stack, which answers
/// `"ObjectReadStack"` for every arm because that is what the stack is. `S`
/// is the layout, and only `S` can say which one it is. It is the value
/// [`crate::arms::IndexArm::projection_name`] is written to be compared
/// against, so a caller that selected an arm can prove the selection took
/// rather than echo the selector back to itself.
pub fn projection_name(&self) -> &'static str {
self.projection.read().expect("projection lock").name()
}
/// A stack whose tail has no file behind it. Used by [`ObjectIndex::build`]
/// and by benchmarks; the redb code path is identical, only the backend
/// differs, so a test of this stack is a test of the durable one.
pub fn in_memory(triggers: RebuildTriggers) -> Result<Self> {
let db = Database::builder()
.create_with_backend(redb::backends::InMemoryBackend::new())
.context("creating an in-memory object tail")?;
Self::from_db(db, triggers, None, None)
}
fn from_db(
db: Database,
triggers: RebuildTriggers,
persisted: Option<PathBuf>,
tail_path: Option<PathBuf>,
) -> Result<Self> {
// Materialise both tables so a read on a brand-new database does not
// trip `TableDoesNotExist`.
let w = db.begin_write()?;
let dropped = {
let _ = w.open_table(OBJECTS)?;
let mut meta = w.open_table(META)?;
let found = meta.get(META_TAIL_FORMAT)?.map(|v| v.value());
match found {
// Written by this layout, or by the layout in use when the key
// was introduced (see `META_TAIL_FORMAT`): nothing to do.
Some(TAIL_FORMAT) => 0,
None => {
meta.insert(META_TAIL_FORMAT, TAIL_FORMAT)?;
0
}
// A DIFFERENT layout. Every row in OBJECTS is unreadable by
// `decode_row`, so the tail is emptied here — in the same
// transaction that stamps the new version, so a crash between
// them cannot leave rows of one layout under the mark of
// another. `adopt_journal` then sees every pack as unabsorbed
// and requeues it.
Some(other) => {
let mut objects = w.open_table(OBJECTS)?;
let n = objects.len()?;
objects.retain(|_, _| false)?;
meta.insert(META_TAIL_FORMAT, TAIL_FORMAT)?;
// The oid set changed, so the content epoch moves with it.
let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
meta.insert(META_TAIL_EPOCH, epoch + 1)?;
tracing::warn!(
target: "znippy.git",
found = other,
expected = TAIL_FORMAT,
rows = n,
"the object tail was written in a different layout; dropping it and \
re-indexing from the journal. Reads fall back to the pack until the \
drain catches up — slower, never wrong"
);
n
}
}
};
w.commit()?;
let _ = dropped;
let mut db = db;
// ── the warm open ────────────────────────────────────────────────────
//
// If a persisted projection exists, it IS the projection: mapped, not
// rebuilt. redb then holds only the rows that arrived after the last
// fold, and those are served by the fall-through exactly as an
// un-absorbed row always was. The scan below is over that remainder —
// small by construction — not over the repository.
//
// The one thing to reconcile is the crash window a fold leaves: the
// file is renamed into place BEFORE redb forgets the rows it holds, so
// a crash between the two leaves the same row in both. Rows redb has
// that the file also has are removed here, in one transaction, so the
// two layers are disjoint again and `len` / `oids_in_order` count each
// object once. The other direction — file missing rows redb dropped —
// cannot happen, because the delete only ever follows a completed
// rename.
let folded: Option<(Arc<S>, u128, u128)> = match persisted.as_deref() {
Some(path) => {
// A `.tmp` left by a fold that died before its rename is not a
// projection and must not become one by accident.
let _ = fs::remove_file(tmp_path(path));
load_persisted::<S>(path)
}
None => None,
};
let t_scan = std::time::Instant::now();
let mut entries = scan(&db)?;
let mut migrated = 0u64;
let (projection, unfolded, loaded) = match folded {
Some((folded, map_ms, build_ms)) => {
let dup: Vec<Vec<u8>> = entries
.iter()
.filter(|e| folded.lookup(&e.oid).is_some())
.map(|e| e.oid.clone())
.collect();
if !dup.is_empty() {
let w = db.begin_write()?;
{
let mut objects = w.open_table(OBJECTS)?;
for oid in &dup {
objects.remove(oid.as_slice())?;
}
}
w.commit()?;
tracing::warn!(
target: "znippy.git",
rows = dup.len(),
"the object tail held rows the persisted index already has — a fold \
was interrupted between writing the index and emptying the tail; \
the duplicates are dropped from the tail, nothing is lost"
);
entries.retain(|e| folded.lookup(&e.oid).is_none());
}
// The three costs of a warm open, separately, so the next lever
// (persisting the stree section) is judged on the number it
// would actually remove: `build_ms` is the O(n) stree build over
// the mapped oid column; `map_ms` is the O(1) open+mmap;
// `tail_ms` is the redb scan of the rows since the last fold
// plus the reconcile. `stree_bytes` is what that build produces
// and what persisting it would have to write.
tracing::info!(
target: "znippy.git",
folded = folded.len(),
tail = entries.len(),
map_ms,
build_ms,
tail_ms = t_scan.elapsed().as_millis(),
stree_bytes = folded.resident_bytes().saturating_sub(folded.ipc_bytes()),
"mapped the persisted object index; the redb tail holds only what \
arrived after the last fold"
);
(folded, entries, 1u64)
}
None => {
let projection = Arc::new(S::build(&entries)?);
// ── the cold open of a store that predates persistence ──────
//
// No file, and redb holds rows: a store written before
// 2026-09-04, or one whose index was lost. The projection was
// just built from the full scan the old design paid on every
// open; fold it now, so this is the LAST time this store pays
// it. Nothing else holds `db` yet, so every row it has is a row
// the projection was just built from: drop the table rather
// than remove 14 M keys one by one. The kernel mirror pays this
// once — 14 M rows out of redb, ~850 MB written — and never
// again.
if let Some(path) = persisted.as_deref()
&& !entries.is_empty()
{
write_atomic(path, projection.persisted_bytes().expect("PERSISTS arm"))?;
let w = db.begin_write()?;
w.delete_table(OBJECTS)?;
let _ = w.open_table(OBJECTS)?;
w.commit()?;
migrated = 1;
tracing::info!(
target: "znippy.git",
rows = entries.len(),
path = %path.display(),
"folded the object tail into a persisted index on first open; the \
redb tail now holds only rows that arrive after this"
);
}
(projection, Vec::new(), 0u64)
}
};
// ── give redb its file back ──────────────────────────────────────────
//
// redb reuses freed pages but never shrinks its file. A store that just
// migrated has a redb at its old high-water mark with nothing in it —
// 4 GiB allocated, 2.9 GB of blocks, for the kernel mirror — and a store
// that has been folding for a while sits at one fold window of dead
// pages (270 MB on the same repository, over 54 k live rows). This is the
// one moment the `Database` is owned and no transaction is open, so
// compaction is free of coordination here and nowhere else.
compact_if_fragmented(&mut db, tail_path.as_deref());
// What the projection does not cover, counted the way `append` would
// have counted it, so the column-scan aggregates and the completeness
// fast path are exact from the first read.
let total = projection.len() as u64 + unfolded.len() as u64;
let mut unabsorbed_size = 0u64;
let mut unabsorbed_types = [0u64; 8];
let mut tail_bytes = 0u64;
for e in &unfolded {
unabsorbed_size += e.uncompressed_size;
unabsorbed_types[e.obj_type.code() as usize] += 1;
tail_bytes += e.len;
}
let db = Arc::new(db);
let stack = Self {
tail: db,
projection: RwLock::new(projection),
triggers,
total_rows: AtomicU64::new(total),
tail_hits: AtomicU64::new(0),
absent: AtomicU64::new(0),
tail_bytes: AtomicU64::new(tail_bytes),
rebuilds: AtomicU64::new(1),
generation: AtomicU64::new(1),
unabsorbed_size: AtomicU64::new(unabsorbed_size),
unabsorbed_types: unabsorbed_types.map(AtomicU64::new),
unabsorbed_rows: AtomicU64::new(unfolded.len() as u64),
tail_txns: AtomicU64::new(0),
persisted,
fold: Mutex::new(()),
folds: AtomicU64::new(migrated),
persisted_loads: AtomicU64::new(loaded),
};
Ok(stack)
}
/// **The projection covers every row in the repository**, so a miss in the
/// stree *is* an absence and the tail need not be asked.
///
/// Sound by construction, not by hope: the projection is built from a scan
/// of the tail, so its rows are always a subset of the tail's; the archive
/// is append-only, so no row ever leaves; therefore equal counts mean equal
/// sets. A concurrent append between this check and the lookup only means
/// the lookup answers as of an instant before that append, which is what a
/// lookup issued a microsecond earlier would have done anyway.
///
/// This is worth the paragraph because of what it buys. MEASURED on oden
/// 2026-08-07 (`read_stack_bench`, 100 000 objects, batch 1000): a `have`
/// negotiation — 10% hit, so 90% of the oids are in no repository at all —
/// cost **544 ns** per oid when every one of those absences took a redb
/// round trip, against **77 ns** for the bare Arrow arm. With this fast path
/// the same workload costs **68 ns**, i.e. the Arrow arm's own cost inside
/// the noise band. A `have` negotiation is *mostly* misses, so without this
/// the stack would be the dominant cost of the most common operation a git
/// server performs.
#[inline]
pub fn projection_is_complete(&self) -> bool {
self.unabsorbed_rows.load(Ordering::Acquire) == 0
}
/// Append objects to the tail.
///
/// **Append-only is enforced here, not assumed.** An oid already present with
/// identical facts is a no-op (a re-pushed pack repeats objects, and that is
/// normal); an oid already present with *different* facts is an error and
/// nothing in the batch is written. Without that refusal a stree hit could
/// return a superseded row and the projection would be not merely incomplete
/// but wrong, which is the one thing this design must not allow.
///
/// Runs a rebuild inline if the batch trips a threshold; the reason is
/// returned so the caller can see that it happened.
pub fn append(&self, entries: &[IndexEntry]) -> Result<Option<RebuildReason>> {
if entries.is_empty() {
return Ok(None);
}
let width = entries[0].oid.len();
if width != 20 && width != 32 {
bail!("oid width {width} is neither sha1 (20) nor sha256 (32)");
}
// Under the fold lock for the transaction AND the counter updates, so a
// fold cannot reset `unabsorbed_rows` between this batch landing in
// redb and its count landing in the atomics. Released before the
// trigger check below, because `rebuild` takes the same lock.
let guard = self.fold.lock().map_err(|_| anyhow!("the fold lock is poisoned"))?;
let w = self.tail.begin_write()?;
let mut added = 0u64;
let mut bytes = 0u64;
// Accumulated locally and applied only after the commit: a `bail!`
// half-way through the batch rolls redb back, and counters bumped inside
// the loop would survive a rollback and put the column-scan aggregates
// permanently out of step with the tail.
let mut added_size = 0u64;
let mut added_types = [0u64; 8];
{
let mut objects = w.open_table(OBJECTS)?;
let mut meta = w.open_table(META)?;
let mut seq = meta.get(META_ARRIVAL_SEQ)?.map(|v| v.value()).unwrap_or(0);
// The rows already known: the projection first, then redb. On a
// persisting stack the projection holds every folded row and redb
// does NOT, so a re-pushed or re-indexed object that was folded
// would otherwise be inserted into redb as new and exist in both
// layers — `len()` doubles, `oids_in_order` repeats, and the next
// fold's `validate` refuses the duplicate oid. MEASURED 2026-09-04
// on `dropping_the_exploded_table_and_reopening_still_answers`:
// 2704 rows for a 1352-object repository, exactly ×2. The stree
// lookup is also the cheaper check by an order of magnitude, so the
// common case of a re-push now never opens redb for it.
let projection = self.projection.read().expect("projection lock").clone();
for e in entries {
if e.oid.len() != width {
bail!(
"mixed oid widths in one append: {width} and {}",
e.oid.len()
);
}
let known: Option<TailRow> = match projection.lookup(&e.oid) {
Some(row) => Some(TailRow {
seq: 0,
offset: row.offset,
len: row.len,
obj_type: row.obj_type,
uncompressed_size: row.uncompressed_size,
delta_base: row.delta_base,
}),
None => match objects.get(e.oid.as_slice())? {
Some(existing) => Some(decode_row(existing.value())?),
None => None,
},
};
if let Some(old) = known {
// ── IDENTITY, not PLACEMENT ─────────────────────────
//
// `offset`, `len` and `delta_base` say WHERE this copy of
// the object landed. They are not what it IS. A re-pushed
// pack legitimately places the same oid somewhere else —
// that is the ordinary case, not a violation — so comparing
// them refused a normal push.
//
// MEASURED 2026-08-14: pushing an object the repository
// already held aborted the ingest and dropped the
// connection with `client_told=false`, so the client saw
// "send-pack: unexpected disconnect" and no reason at all.
// It bit `bare_lifecycle` and `endurance` on all four
// znippy columns of the bench, and it bites any real
// `--force` push that resends a known object.
//
// The guard's own doc three lines up already said what it
// meant to do — "An oid already present with identical
// facts is a no-op (a re-pushed pack repeats objects, and
// that is normal)" — so the code contradicted its contract.
//
// What the original comparison was protecting is still
// protected, and by a stronger argument: we KEEP THE OLD
// ROW. The archive is append-only and `gc` truncates
// nothing, so the offset and delta base already recorded
// still address bytes that are still there. The projection
// therefore never carries a base the archive did not write
// — the first writer's placement stands, forever.
//
// What CANNOT be waved through is a disagreement about the
// object itself. Same oid, different type or different
// inflated size, means either a hash collision or a
// corrupted row, and no push may quietly overwrite that.
// …and `obj_type` is a REPRESENTATION, not the object's
// kind. `ofs-delta` and `ref-delta` say how these bytes are
// ENCODED; what the object IS lives at the end of the delta
// chain. The same oid legitimately arrives whole in one
// pack and delta-encoded in another, so comparing those two
// is comparing encodings and calling them different
// objects. MEASURED here, `gunnar.import_export`, znippy:
//
// 6da81e5e… is already stored as (…, ofs-delta, size 2084)
// and cannot be redeclared as (…, tree, size 2084)
//
// — the inflated size agrees to the byte, which is the
// identity; only the encoding moved. The refusal poisoned
// the repository (`it stays un-indexed; reads will keep
// falling back`) and every later read answered `the
// repository is unavailable; try again`.
//
// So a type disagreement is a violation only when BOTH
// rows name a real object kind. The inflated size is
// compared unconditionally, and it is the half that
// actually catches a collision or a corrupt row.
let encoded = |t: ObjType| matches!(t, ObjType::OfsDelta | ObjType::RefDelta);
let kind_disagrees = !encoded(old.obj_type)
&& !encoded(e.obj_type)
&& old.obj_type != e.obj_type;
if kind_disagrees || old.uncompressed_size != e.uncompressed_size {
bail!(
"identity violation: {} is already stored as \
(offset {}, len {}, {}, size {}, delta_base {}) and cannot be \
redeclared as (offset {}, len {}, {}, size {}, delta_base {}) — \
the TYPE or the inflated SIZE differs, so the same oid is \
describing a different object. Placement (offset, len, \
delta_base) may differ freely: a re-pushed pack lands its \
copies elsewhere and the first writer's row stands.",
hex::encode(&e.oid),
old.offset,
old.len,
old.obj_type.as_str(),
old.uncompressed_size,
old.delta_base,
e.offset,
e.len,
e.obj_type.as_str(),
e.uncompressed_size,
e.delta_base,
);
}
continue;
}
objects.insert(e.oid.as_slice(), encode_row(seq, e).as_slice())?;
seq += 1;
added += 1;
bytes += e.len;
added_size += e.uncompressed_size;
added_types[e.obj_type.code() as usize] += 1;
}
meta.insert(META_ARRIVAL_SEQ, seq)?;
// Only a genuine insertion changes the oid set, so the content epoch
// holds still across a re-pushed pack that adds nothing (added == 0).
// The bump rides this same transaction, so the epoch and the rows it
// describes commit or roll back together.
if added > 0 {
let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
meta.insert(META_TAIL_EPOCH, epoch + 1)?;
}
}
w.commit()?;
self.total_rows.fetch_add(added, Ordering::AcqRel);
self.unabsorbed_rows.fetch_add(added, Ordering::AcqRel);
self.unabsorbed_size.fetch_add(added_size, Ordering::AcqRel);
for (slot, n) in self.unabsorbed_types.iter().zip(added_types) {
slot.fetch_add(n, Ordering::AcqRel);
}
let total_bytes = self.tail_bytes.fetch_add(bytes, Ordering::AcqRel) + bytes;
drop(guard);
if let Some(threshold) = self.volume_threshold()
&& total_bytes >= threshold
{
self.rebuild()?;
return Ok(Some(RebuildReason::TailBytes(total_bytes)));
}
self.maybe_rebuild()
}
/// Which threshold, if any, is currently tripped. Pure read; a caller on a
/// read-only repository can poll this from its own maintenance tick.
pub fn rebuild_due(&self) -> Option<RebuildReason> {
let hits = self.tail_hits.load(Ordering::Acquire);
if let Some(threshold) = self.triggers.miss_threshold(self.projection_len() as u64)
&& hits >= threshold
{
return Some(RebuildReason::TailHits(hits));
}
let bytes = self.tail_bytes.load(Ordering::Acquire);
if let Some(threshold) = self.volume_threshold()
&& bytes >= threshold
{
return Some(RebuildReason::TailBytes(bytes));
}
None
}
/// The volume trigger's threshold right now, or `None` when it is off.
///
/// The fixed floor everywhere; on a persisting stack, the larger of the
/// floor and `fold_ratio ×` the persisted index's current size — so a fold
/// on a big repository is paid for by a proportionate amount of new data,
/// and the sum of index writes over an import stays ≈ the bytes imported.
/// See [`RebuildTriggers::DEFAULT_FOLD_RATIO`].
fn volume_threshold(&self) -> Option<u64> {
let floor = self.triggers.tail_bytes;
if floor == 0 {
return None;
}
if !self.persists() || self.triggers.fold_ratio <= 0.0 {
return Some(floor);
}
let index_bytes = self.projection.read().expect("projection lock").ipc_bytes() as f64;
Some(floor.max((self.triggers.fold_ratio * index_bytes) as u64))
}
/// Rebuild if [`rebuild_due`](Self::rebuild_due) says so.
pub fn maybe_rebuild(&self) -> Result<Option<RebuildReason>> {
match self.rebuild_due() {
Some(reason) => {
self.rebuild()?;
Ok(Some(reason))
}
None => Ok(None),
}
}
/// Rebuild the Arrow/stree projection from a full ordered scan of the tail
/// and swap it in, resetting the counters and bumping the generation.
///
/// The old projection stays live for every reader until the swap; the write
/// lock is held only for the pointer store, not for the build.
pub fn rebuild(&self) -> Result<()> {
let _g = self.fold.lock().map_err(|_| anyhow!("the fold lock is poisoned"))?;
self.fold_locked(None).map(|_| ())
}
/// **The fold** — a rebuild that, on a persisting stack, also writes the
/// projection down and empties redb of what it wrote. Caller holds
/// [`fold`](Self::fold).
///
/// The merge is of two disjoint sorted runs: the current projection (which,
/// on a persisting stack, IS the file's content — the invariant every fold
/// re-establishes) and the rows redb holds right now. On a non-persisting
/// stack redb holds every row and the projection is not consulted, which is
/// the pre-2026-09-04 rebuild unchanged.
///
/// `live`, when given, is applied to the **projection's** rows only: those
/// are all older than the reachability set `retain`'s caller computed, so
/// the predicate has an answer for each of them. The redb rows are never
/// filtered here — `retain` has already removed the dead ones under this
/// same lock, and any row that landed after the caller's scan is one the
/// predicate does not know and must not be asked about (`gc`'s closure
/// answers `false` for an oid it never saw).
///
/// Durability order, and why it is that order: (1) the new index goes to
/// `objects.index.tmp`, is fsynced, and is renamed over `objects.index`;
/// (2) the rows it holds are removed from redb. A crash after (1) and
/// before (2) leaves them in both — reconciled at the next open. A crash
/// during (1) leaves a `.tmp` nobody reads and a redb that still has every
/// row. Nothing is ever lost, because a row leaves redb only after the
/// rename that made it durable elsewhere has returned.
///
/// Returns how many projection rows `live` rejected.
fn fold_locked(&self, live: Option<&dyn Fn(&[u8]) -> bool>) -> Result<u64> {
let unfolded = scan(&self.tail)?;
let mut dropped = 0u64;
let mut merged: Vec<IndexEntry>;
if self.persists() {
let cur = self.projection.read().expect("projection lock").clone();
merged = Vec::with_capacity(cur.len() + unfolded.len());
for o in 0..cur.len() as u32 {
let e = cur.entry_at(o);
if live.is_none_or(|f| f(&e.oid)) {
merged.push(e);
} else {
dropped += 1;
}
}
} else {
merged = Vec::with_capacity(unfolded.len());
}
let forget: Vec<Vec<u8>> = if self.persists() {
unfolded.iter().map(|e| e.oid.clone()).collect()
} else {
Vec::new()
};
merged.extend(unfolded);
let total = merged.len() as u64;
let fresh = Arc::new(S::build(&merged)?);
drop(merged);
if let Some(path) = self.persisted.as_deref() {
write_atomic(path, fresh.persisted_bytes().expect("PERSISTS arm"))?;
if !forget.is_empty() {
let w = self.tail.begin_write()?;
{
let mut objects = w.open_table(OBJECTS)?;
for oid in &forget {
objects.remove(oid.as_slice())?;
}
}
w.commit()?;
}
self.folds.fetch_add(1, Ordering::AcqRel);
}
*self
.projection
.write()
.map_err(|_| anyhow!("the projection lock is poisoned"))? = fresh;
self.total_rows.store(total, Ordering::Release);
self.tail_hits.store(0, Ordering::Release);
self.tail_bytes.store(0, Ordering::Release);
self.unabsorbed_size.store(0, Ordering::Release);
self.unabsorbed_rows.store(0, Ordering::Release);
self.tail_txns.store(0, Ordering::Release);
for slot in &self.unabsorbed_types {
slot.store(0, Ordering::Release);
}
self.rebuilds.fetch_add(1, Ordering::AcqRel);
self.generation.fetch_add(1, Ordering::AcqRel);
Ok(dropped)
}
pub fn stats(&self) -> StackStats {
StackStats {
sealed_rows: self.projection_len() as u64,
total_rows: self.total_rows.load(Ordering::Acquire),
tail_hits: self.tail_hits.load(Ordering::Acquire),
absent: self.absent.load(Ordering::Acquire),
tail_bytes: self.tail_bytes.load(Ordering::Acquire),
rebuilds: self.rebuilds.load(Ordering::Acquire),
folds: self.folds.load(Ordering::Acquire),
persisted_loads: self.persisted_loads.load(Ordering::Acquire),
generation: self.generation.load(Ordering::Acquire),
unabsorbed_rows: self.unabsorbed_rows.load(Ordering::Acquire),
tail_txns: self.tail_txns.load(Ordering::Acquire),
}
}
/// Rows in the projection right now. `stats().total_rows - this` is the
/// un-absorbed tail.
pub fn projection_len(&self) -> usize {
self.projection.read().expect("projection lock").len()
}
/// Every oid in the repository, in oid order — which is ordinal order, since
/// redb keys the tail by the raw oid.
///
/// A full scan, and it is here for the one caller that genuinely needs the
/// whole set at once: a GC, which has to name what is *not* live.
pub fn oids_in_order(&self) -> Result<Vec<Vec<u8>>> {
let read = self.tail.begin_read()?;
let objects = read.open_table(OBJECTS)?;
let mut tail = Vec::with_capacity(objects.len()? as usize);
for row in objects.iter()? {
let (k, _) = row?;
tail.push(k.value().to_vec());
}
if !self.persists() {
// redb holds every row, so its key order is the whole answer.
return Ok(tail);
}
// Two sorted runs — the persisted projection and the redb tail — merged
// once. Disjoint by the fold's invariant; the equal-key arm below exists
// for the window inside a fold where the file is renamed and redb has
// not yet forgotten, so a caller in that window counts each oid once.
let cur = self.projection.read().expect("projection lock").clone();
let n = cur.len();
let mut out = Vec::with_capacity(n + tail.len());
let (mut i, mut j) = (0usize, 0usize);
while i < n || j < tail.len() {
let a = (i < n).then(|| cur.oid_at(i as u32));
let b = tail.get(j).map(Vec::as_slice);
match (a, b) {
(Some(a), Some(b)) => match a.as_slice().cmp(b) {
std::cmp::Ordering::Less => {
out.push(a);
i += 1;
}
std::cmp::Ordering::Greater => {
out.push(b.to_vec());
j += 1;
}
std::cmp::Ordering::Equal => {
out.push(a);
i += 1;
j += 1;
}
},
(Some(a), None) => {
out.push(a);
i += 1;
}
(None, Some(b)) => {
out.push(b.to_vec());
j += 1;
}
(None, None) => break,
}
}
Ok(out)
}
/// **The ordinal space** — every oid in the repository, back to back, in oid
/// (= ordinal) order — and where each part came from.
///
/// This replaces the `objects.ordinal` memo (S-082). That file existed to
/// spare a cold open the full redb scan it took to rebuild the sorted oid
/// list; the scan no longer exists to be spared. On a persisting stack the
/// folded oids ARE the projection's oid column — one contiguous copy out of
/// the mapped buffer — and only the redb tail (the rows since the last fold)
/// is scanned and merged in. `folded_rows` / `tail_rows` say which is which,
/// so a test can assert the space came off the index and not off a scan
/// without comparing two byte strings that are equal either way.
///
/// Same overlap note as [`oids_in_order`](Self::oids_in_order): inside a
/// fold's rename-then-forget window an oid may sit in both layers for a
/// moment; the merge emits it once.
pub fn oids_raw(&self) -> Result<OrdinalSpace> {
let read = self.tail.begin_read()?;
let objects = read.open_table(OBJECTS)?;
let mut tail: Vec<Vec<u8>> = Vec::with_capacity(objects.len()? as usize);
for row in objects.iter()? {
let (k, _) = row?;
tail.push(k.value().to_vec());
}
let tail_rows = tail.len() as u64;
if !self.persists() {
let mut raw = Vec::with_capacity(tail.iter().map(Vec::len).sum());
for oid in &tail {
raw.extend_from_slice(oid);
}
return Ok(OrdinalSpace { raw, folded_rows: 0, tail_rows });
}
let cur = self.projection.read().expect("projection lock").clone();
let folded = cur.oids_raw();
let folded_rows = cur.len() as u64;
if tail.is_empty() {
return Ok(OrdinalSpace { raw: folded, folded_rows, tail_rows: 0 });
}
let w = match tail.first() {
Some(o) => o.len(),
None => unreachable!("tail is non-empty"),
};
let mut raw = Vec::with_capacity(folded.len() + tail.len() * w);
let (mut i, mut j) = (0usize, 0usize);
let n = folded.len() / w;
while i < n || j < tail.len() {
let a = (i < n).then(|| &folded[i * w..(i + 1) * w]);
let b = tail.get(j).map(Vec::as_slice);
match (a, b) {
(Some(a), Some(b)) => match a.cmp(b) {
std::cmp::Ordering::Less => {
raw.extend_from_slice(a);
i += 1;
}
std::cmp::Ordering::Greater => {
raw.extend_from_slice(b);
j += 1;
}
std::cmp::Ordering::Equal => {
raw.extend_from_slice(a);
i += 1;
j += 1;
}
},
(Some(a), None) => {
raw.extend_from_slice(a);
i += 1;
}
(None, Some(b)) => {
raw.extend_from_slice(b);
j += 1;
}
(None, None) => break,
}
}
Ok(OrdinalSpace { raw, folded_rows, tail_rows })
}
/// `(tail_epoch, object_count)` in one read transaction — the two cheap facts
/// that say whether the oid set has moved, **without** the ordered scan
/// [`Self::oids_in_order`] pays. (They were the `objects.ordinal` memo's
/// validity key until that file went, 2026-09-04.) `epoch` is [`META_TAIL_EPOCH`] (0 when a
/// store predates it, e.g. was populated by an older binary); the count is
/// redb's own O(1) table length. Both are metadata reads — a point lookup and
/// a stored length — so this is microseconds against the scan's hundreds of
/// milliseconds, which is the whole reason the memo can exist.
pub fn tail_epoch_and_len(&self) -> Result<(u64, u64)> {
let read = self.tail.begin_read()?;
let meta = read.open_table(META)?;
let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
let objects = read.open_table(OBJECTS)?;
// On a persisting stack the repository is the persisted projection plus
// the redb tail, disjoint. Inside a fold's rename-then-forget window the
// two overlap for a moment and this over-counts; the memo it validates
// then reads as stale and re-scans, and the next call is exact again —
// a transient re-scan, never a wrong answer, and not worth taking the
// fold lock on the read path for.
let folded = if self.persists() { self.projection_len() as u64 } else { 0 };
Ok((epoch, objects.len()? + folded))
}
/// **Which of these archive extents already have rows here** — the
/// crash-recovery diff, and the reason §13.12's `indexed` bit is derived
/// rather than stored.
///
/// A pack is unabsorbed **iff its extent is in the journal and its rows are
/// not in the index**. Both of those are already durable — the journal is
/// fsynced on the ack path, the tail is a redb commit — so the bit is a diff
/// of two durable facts and there is nothing for it to drift from. A *stored*
/// bit would be a third fact that can disagree with the two it describes,
/// which is precisely how a pack ends up marked absorbed with no rows behind
/// it: fast and wrong, in the one direction (`absent`) that loses a client's
/// objects during negotiation.
///
/// `out[i]` answers `extents[i]`. **The tail and not the projection**, because
/// the projection is a snapshot of a prefix of the tail and the question is
/// about what survived the crash.
///
/// # Cost
///
/// One ordered scan of the tail that **stops the moment every extent has been
/// hit**. Rows arrive in oid order, which is uncorrelated with the pack an
/// object came from, so a store whose packs are all absorbed answers after a
/// few rows per pack rather than after a full scan — the coupon-collector
/// case, and the one every clean reopen takes. The scan only runs to the end
/// when some pack genuinely has **no** rows, which is exactly the case where
/// the caller is about to re-resolve that whole pack anyway.
///
/// MEASURED on oden 2026-08-08, release, file-backed redb tail, 1-minute
/// loadavg 7.1–7.5 (another tenant's work — read these as ratios):
///
/// | packs | rows | every pack absorbed | one pack un-absorbed |
/// |---:|---:|---:|---:|
/// | 1 000 | 1 000 000 | **0.77 ms** | 120.6 ms |
/// | 10 000 | 1 000 000 | **11.7 ms** | 141.8 ms |
/// | 1 000 | 3 686 | **0.40 ms** | 0.39 ms |
///
/// The first column is what a clean reopen pays and it is bounded by the
/// **pack** count, not the object count — 1000 packs over a million objects
/// costs the same order as 1000 packs over four thousand. The second column
/// is the crash path, where the scan runs to the end: 120 ms over a million
/// rows, against the ~160 ms a *single* 2687-object pack takes to re-absorb
/// (`a_push_ends_in_object_rows_and_no_read_asked_for_them`). The diff is
/// therefore never the expensive half of a recovery.
/// The third row is the shape
/// [`crate::git_ops::tests::the_derive_on_open_diff_costs_milliseconds_at_a_realistic_pack_count`]
/// asserts on every run, over real pushed packs rather than synthetic rows.
///
/// A zero-length extent has no rows by construction and is answered `false`
/// without looking.
pub fn extents_with_rows(&self, extents: &[(u64, u64)]) -> Result<Vec<bool>> {
let mut hit = vec![false; extents.len()];
// Sorted by start, so one row is placed with a binary search instead of a
// pass over every extent: the scan below is the hot loop.
let mut order: Vec<usize> = (0..extents.len()).filter(|&i| extents[i].1 > 0).collect();
order.sort_unstable_by_key(|&i| extents[i].0);
let starts: Vec<u64> = order.iter().map(|&i| extents[i].0).collect();
let mut wanted = order.len();
if wanted == 0 {
return Ok(hit);
}
// One row's offset, placed against the sorted extents. `true` when every
// extent has now been hit and the scan can stop.
let mut place = |offset: u64| -> bool {
// The last extent that starts at or before this row.
let p = starts.partition_point(|&s| s <= offset);
if p == 0 {
return false;
}
let i = order[p - 1];
let (start, len) = extents[i];
if offset < start + len && !hit[i] {
hit[i] = true;
wanted -= 1;
}
wanted == 0
};
// The persisted projection first: on a persisting stack it holds every
// row older than the last fold, which is nearly all of them, and its
// offset column is one contiguous `u64` run — the cheapest scan there
// is. Then the redb tail for the rows that arrived since.
if self.persists() {
let cur = self.projection.read().expect("projection lock").clone();
for o in 0..cur.len() as u32 {
if place(cur.extent_at(o).0) {
return Ok(hit);
}
}
}
let read = self.tail.begin_read()?;
let objects = read.open_table(OBJECTS)?;
for row in objects.iter()? {
let (_, v) = row?;
if place(decode_row(v.value())?.offset) {
break;
}
}
Ok(hit)
}
/// **Drop every row `live` rejects.** Returns how many rows went.
///
/// This is the one operation that is not append-only, and it exists for
/// exactly one caller: `GitOps::gc`, which computes reachability and then has
/// to remove what is unreachable *before* base znippy compacts the payload —
/// so that base's notion of "live" already means what git means (§13.20).
///
/// The projection is rebuilt inside the same call. It has to be: it is a
/// snapshot of a tail that no longer says the same thing, and a stree that
/// still answers for a deleted oid would be **wrong** rather than merely
/// incomplete, which is the one failure this design does not tolerate.
pub fn retain(&self, live: &dyn Fn(&[u8]) -> bool) -> Result<u64> {
// The whole of `retain` — the redb pass, the fold that rewrites the
// persisted index without the dead rows, and the epoch bump — under one
// hold of the fold lock, so no fold can interleave and re-persist a row
// this call is removing.
let _g = self.fold.lock().map_err(|_| anyhow!("the fold lock is poisoned"))?;
let w = self.tail.begin_write()?;
let mut dropped = 0u64;
{
let mut objects = w.open_table(OBJECTS)?;
let dead: Vec<Vec<u8>> = objects
.iter()?
.filter_map(|row| row.ok())
.filter(|(k, _)| !live(k.value()))
.map(|(k, _)| k.value().to_vec())
.collect();
for oid in &dead {
objects.remove(oid.as_slice())?;
dropped += 1;
}
// A removal changes the oid set exactly as an insert does, so the
// content epoch must record it: bump it in this same transaction
// when anything actually went. `arrival_seq` is deliberately not
// touched — it is an *arrival* ordinal, not a live count.
if dropped > 0 {
let mut meta = w.open_table(META)?;
let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
meta.insert(META_TAIL_EPOCH, epoch + 1)?;
}
}
w.commit()?;
// The persisted projection's dead rows go in the fold: it rewrites the
// index from `live` rows only and swaps it in, so a stree that answered
// for a deleted oid a moment ago answers `None` now. On a persisting
// stack this is where most of what dies actually goes — those rows were
// never in redb — so the epoch is bumped again for them. (`len` in the
// memo's `(epoch, len)` key would catch it too; the bump keeps the
// contract that a changed oid set is a changed epoch.)
let folded_dead = self.fold_locked(Some(live))?;
if folded_dead > 0 {
let w = self.tail.begin_write()?;
{
let mut meta = w.open_table(META)?;
let epoch = meta.get(META_TAIL_EPOCH)?.map(|v| v.value()).unwrap_or(0);
meta.insert(META_TAIL_EPOCH, epoch + 1)?;
}
w.commit()?;
}
Ok(dropped + folded_dead)
}
/// One tail read transaction serving `oids`, filling only the slots that are
/// still `None`. This is the fall-through, and it is a single transaction for
/// the whole batch on purpose — skade's `resolve_many` phase 2.
///
/// **Not called at all when the projection is complete** — see
/// [`projection_is_complete`](Self::projection_is_complete). The unfilled
/// slots are then genuine absences and are counted as such by
/// [`absences_only`](Self::absences_only).
fn fill_from_tail(&self, oids: &[&[u8]], out: &mut [Option<IndexRow>]) -> Result<()> {
self.tail_txns.fetch_add(1, Ordering::AcqRel);
let read = self.tail.begin_read()?;
let objects = read.open_table(OBJECTS)?;
let mut hits = 0u64;
let mut absent = 0u64;
for (slot, oid) in out.iter_mut().zip(oids) {
if slot.is_some() {
continue;
}
match objects.get(*oid)? {
Some(v) => {
*slot = Some(decode_row(v.value())?.as_index_row());
hits += 1;
}
None => absent += 1,
}
}
self.tail_hits.fetch_add(hits, Ordering::AcqRel);
self.absent.fetch_add(absent, Ordering::AcqRel);
Ok(())
}
/// Count the unfilled slots as absences without asking the tail. Only ever
/// called when [`projection_is_complete`](Self::projection_is_complete),
/// where "the projection does not have it" and "the repository does not have
/// it" are the same statement.
fn absences_only(&self, out: &[Option<IndexRow>]) {
let n = out.iter().filter(|s| s.is_none()).count() as u64;
self.absent.fetch_add(n, Ordering::AcqRel);
}
}
/// Every row in the tail, in oid order, as index entries. redb's B-tree is
/// already ordered by the raw oid key, so this hands `S::build` its rows in the
/// order it wants them and the sort inside is a no-op scan.
fn scan(db: &Database) -> Result<Vec<IndexEntry>> {
let read = db.begin_read()?;
let objects = read.open_table(OBJECTS)?;
let mut out = Vec::with_capacity(objects.len()? as usize);
for row in objects.iter()? {
let (k, v) = row?;
out.push(decode_row(v.value())?.as_entry(k.value()));
}
Ok(out)
}
// ── the redb 2 tail, migrated once ───────────────────────────────────────────
/// `objects.tail` → `objects.tail.redb4-migrating`: where [`migrate_redb2_tail`]
/// builds the redb 4 copy before it is renamed over the old file.
fn migrating_path(tail: &Path) -> PathBuf {
let mut name = tail.file_name().map(|n| n.to_os_string()).unwrap_or_default();
name.push(".redb4-migrating");
tail.with_file_name(name)
}
/// Rows copied per redb 4 write transaction while migrating. The staging file
/// is not the tail until its rename, so a crash between chunks leaves the old
/// tail untouched and the next open starts the copy again.
const MIGRATE_CHUNK_ROWS: usize = 1 << 20;
/// **The tail on redb 4, and the redb 2 tail every store before 2026-10-06 has.**
///
/// # Why redb 4
///
/// MEASURED 2026-10-06 (oden, XFS — the appliance's filesystem; gunnar's
/// `tests/repo_overhead.rs`): a repository's whole fixed cost on disk was this
/// one file. redb 2.6.3 allocates **540 KB** when it creates a database and
/// **1.6 MB** (3.7 MB apparent) once that database has been closed, in file
/// format v2 and v3 alike, so a small repository cost 1 654 784 B of disk — about
/// 2 MB per repository on the load test's estate, ~20 GB at 10 000 accounts of
/// three small repositories. redb 4.3.0, same tables, same rows: **4 KB** at
/// create, 36-57 KB after use, and it gives trailing free pages back to the
/// filesystem when it closes. The same probe found redb 2 turning a
/// `retain`-everything over 20 000 rows into a 205 MB file (redb 4: 4 MB).
///
/// # The migration
///
/// redb 4 refuses a redb 2 file with `UpgradeRequired` before writing a byte of
/// it. Then, once per store:
///
/// 1. a staging file left by a migration that died is removed;
/// 2. the old tail is opened with redb 2 and both tables are copied, in chunks
/// of [`MIGRATE_CHUNK_ROWS`], into a fresh redb 4 database at
/// [`migrating_path`] — `objects` and `meta` verbatim, so `arrival_seq`,
/// `tail_epoch` and `tail_format` carry over and nothing downstream can tell;
/// 3. the last commit is durable (redb's default), the staging file is renamed
/// over `objects.tail`, and the directory is fsynced.
///
/// A crash before the rename leaves the redb 2 tail exactly as it was; after
/// it, the redb 4 tail is complete. There is no state in between to repair.
/// A file redb 2 cannot read either is reported, not dropped: it was
/// unreadable before this existed too.
fn migrate_redb2_tail(tail_path: &Path, found_format: u8, cache_bytes: usize) -> Result<()> {
const OBJECTS2: redb2::TableDefinition<'static, &[u8], &[u8]> =
redb2::TableDefinition::new("objects");
const META2: redb2::TableDefinition<'static, &str, u64> = redb2::TableDefinition::new("meta");
use redb2::{ReadableTable as _, ReadableTableMetadata as _};
let t = std::time::Instant::now();
let staging = migrating_path(tail_path);
match fs::remove_file(&staging) {
Ok(()) => tracing::warn!(
target: "znippy.git",
path = %staging.display(),
"removed a half-built redb 4 tail left by an interrupted migration; starting again"
),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => return Err(e).with_context(|| format!("removing {}", staging.display())),
}
let (before_len, before_alloc) = file_sizes(tail_path);
let old = redb2::Database::builder()
.set_cache_size(cache_bytes)
.create(tail_path)
.with_context(|| {
format!(
"opening the redb 2 object tail at {} (file format {found_format}) to migrate it",
tail_path.display()
)
})?;
let read = old.begin_read()?;
let mut rows = 0u64;
let copied = (|| -> Result<()> {
let new = Database::builder()
.set_cache_size(cache_bytes)
.create(&staging)
.with_context(|| format!("creating {}", staging.display()))?;
{
let w = new.begin_write()?;
{
let _ = w.open_table(OBJECTS)?;
let mut meta = w.open_table(META)?;
match read.open_table(META2) {
Ok(old_meta) => {
for kv in old_meta.iter()? {
let (k, v) = kv?;
meta.insert(k.value(), v.value())?;
}
}
Err(redb2::TableError::TableDoesNotExist(_)) => {}
Err(e) => return Err(e.into()),
}
}
w.commit()?;
}
match read.open_table(OBJECTS2) {
Ok(old_objects) => {
let mut it = old_objects.iter()?;
loop {
let w = new.begin_write()?;
let mut n = 0usize;
{
let mut objects = w.open_table(OBJECTS)?;
for kv in it.by_ref() {
let (k, v) = kv?;
objects.insert(k.value(), v.value())?;
n += 1;
if n == MIGRATE_CHUNK_ROWS {
break;
}
}
}
w.commit()?;
rows += n as u64;
if n < MIGRATE_CHUNK_ROWS {
break;
}
}
anyhow::ensure!(
rows == old_objects.len()?,
"the migration copied {rows} rows of {}",
old_objects.len()?
);
}
Err(redb2::TableError::TableDoesNotExist(_)) => {}
Err(e) => return Err(e.into()),
}
Ok(())
})();
drop(read);
drop(old);
if let Err(e) = copied {
let _ = fs::remove_file(&staging);
return Err(e.context(format!(
"migrating the object tail at {} from redb 2 to redb 4; the redb 2 tail is untouched",
tail_path.display()
)));
}
fs::rename(&staging, tail_path).with_context(|| {
format!("renaming {} over {}", staging.display(), tail_path.display())
})?;
if let Some(dir) = tail_path.parent().filter(|d| !d.as_os_str().is_empty()) {
File::open(dir)
.and_then(|d| d.sync_all())
.with_context(|| format!("syncing the directory of {}", tail_path.display()))?;
}
let (after_len, after_alloc) = file_sizes(tail_path);
tracing::info!(
target: "znippy.git",
tail = %tail_path.display(),
rows,
from_format = found_format,
before_bytes = before_len,
before_allocated = before_alloc,
after_bytes = after_len,
after_allocated = after_alloc,
ms = t.elapsed().as_millis(),
"migrated the object tail from redb 2 to redb 4 (once per store)"
);
Ok(())
}
/// `(apparent, allocated)` bytes of `path`, `(0, 0)` when it cannot be read —
/// for the migration's log line, never for a decision.
fn file_sizes(path: &Path) -> (u64, u64) {
match fs::metadata(path) {
#[cfg(unix)]
Ok(m) => {
use std::os::unix::fs::MetadataExt;
(m.len(), m.blocks() * 512)
}
#[cfg(not(unix))]
Ok(m) => (m.len(), m.len()),
Err(_) => (0, 0),
}
}
// ── the persisted projection on disk ─────────────────────────────────────────
/// `objects.index` → `objects.index.tmp`: where a fold writes before its rename.
fn tmp_path(path: &Path) -> PathBuf {
let mut name = path.file_name().map(|n| n.to_os_string()).unwrap_or_default();
name.push(".tmp");
path.with_file_name(name)
}
/// Write `bytes` so that `path` either holds all of them or is untouched:
/// sibling temp file, fsync, rename, fsync the directory. The same idiom
/// every durable writer in this crate uses, spelled once here for the index.
fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
let tmp = tmp_path(path);
{
let mut f = File::create(&tmp)
.with_context(|| format!("creating {}", tmp.display()))?;
f.write_all(bytes)
.with_context(|| format!("writing {} bytes to {}", bytes.len(), tmp.display()))?;
f.sync_all()
.with_context(|| format!("syncing {}", tmp.display()))?;
}
fs::rename(&tmp, path)
.with_context(|| format!("renaming {} over {}", tmp.display(), path.display()))?;
if let Some(dir) = path.parent().filter(|d| !d.as_os_str().is_empty()) {
File::open(dir)
.and_then(|d| d.sync_all())
.with_context(|| format!("syncing the directory of {}", path.display()))?;
}
Ok(())
}
/// Dead redb pages worth compacting for: below this an open leaves the file
/// alone, because a small repository's redb is a few pages either way and a
/// rewrite of it on every open would be churn for nothing. Compaction costs
/// what is LIVE, not what is dead — after a fold that is little — so the floor
/// is set by the file lying about the store, not by the work: 16 MiB is a redb
/// that has doubled itself four times past a few thousand live rows.
const COMPACT_FRAGMENTED_FLOOR: u64 = 16 << 20;
/// Shrink redb's file when it holds at least [`COMPACT_FRAGMENTED_FLOOR`] of dead
/// pages. Needs the `Database` owned and no transaction open, which is true at
/// exactly one place — the open — and nowhere else; that is why it lives here
/// rather than behind a maintain verb. Any failure is logged and swallowed: the
/// file being large is not a fault, and a store must open over it regardless.
fn compact_if_fragmented(db: &mut Database, tail_path: Option<&Path>) {
let Some(tail_path) = tail_path else { return }; // in-memory: nothing on disk to shrink
let file_len = match fs::metadata(tail_path) {
Ok(m) => m.len(),
Err(_) => return,
};
// `stats()` lives on the write transaction (redb 2 and 4); this one is opened to
// be read and aborted, never committed. The dead-page figure is taken from
// the FILE, not from `fragmented_bytes()`: right after a commit that freed
// pages (the migration fold's `delete_table`) redb still has them PENDING
// — two-phase free — and reports them as neither stored nor fragmented.
// MEASURED: the same 34 MB file read `fragmented = 12.8 MiB` inside the
// open and `33 MB` from a fresh handle a moment later. The file length
// minus what is stored is what the store is actually lying by, and
// `compact()` begins with the two commits that turn pending into free.
let (stored, metadata) = {
let w = match db.begin_write() {
Ok(w) => w,
Err(e) => {
tracing::debug!(target: "znippy.git", error = %e, "redb stats unavailable; not compacting");
return;
}
};
let s = w.stats().map(|s| (s.stored_bytes(), s.metadata_bytes()));
let _ = w.abort();
match s {
Ok(s) => s,
Err(e) => {
tracing::debug!(target: "znippy.git", error = %e, "redb stats unavailable; not compacting");
return;
}
}
};
let dead = file_len.saturating_sub(stored + metadata);
if dead < COMPACT_FRAGMENTED_FLOOR {
return;
}
let t = std::time::Instant::now();
match db.compact() {
Ok(did) => tracing::info!(
target: "znippy.git",
file_before = file_len,
file_after = fs::metadata(tail_path).map(|m| m.len()).unwrap_or(0),
stored_bytes = stored,
compacted = did,
ms = t.elapsed().as_millis(),
tail = %tail_path.display(),
"compacted the redb object tail: it was carrying dead pages from before a fold"
),
Err(e) => tracing::warn!(
target: "znippy.git",
file_len,
stored_bytes = stored,
error = %e,
"could not compact the redb object tail; it keeps its size, nothing else changes"
),
}
}
/// The persisted index as an Arrow [`Buffer`] — `None` when there is none.
///
/// Memory-mapped where the platform allows it, so the columns
/// [`ObjectIndex::from_persisted`] decodes are views into the page cache: the
/// index costs resident memory only for the pages a lookup touches, and none
/// of it is a copy. Elsewhere the file is read into an aligned buffer, which
/// keeps every other property (no rebuild, no redb scan) and gives up only
/// the sharing.
fn map_persisted(path: &Path) -> Result<Option<Buffer>> {
let f = match File::open(path) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => return Err(e).with_context(|| format!("opening {}", path.display())),
};
let len = f
.metadata()
.with_context(|| format!("stat {}", path.display()))?
.len();
if len == 0 {
// A persisted projection is never zero bytes — even an empty index
// carries its schema — so this is a truncated write, not an index.
return Ok(None);
}
#[cfg(target_os = "linux")]
{
// SAFETY: the file is replaced only by `write_atomic`'s rename, which
// swaps the directory entry and never writes into the inode this
// mapping is over; the old inode lives until the last mapping drops.
let map = unsafe { memmap2::Mmap::map(&f) }
.with_context(|| format!("mapping {}", path.display()))?;
let map = Arc::new(map);
let ptr = std::ptr::NonNull::new(map.as_ptr() as *mut u8)
.ok_or_else(|| anyhow!("mmap of {} returned a null pointer", path.display()))?;
// SAFETY: `ptr` and `len` describe exactly the mapping `map` owns, and
// `map` is kept alive inside the buffer as its allocation owner.
let buf = unsafe { Buffer::from_custom_allocation(ptr, map.len(), map) };
Ok(Some(buf))
}
#[cfg(not(target_os = "linux"))]
{
use std::io::Read;
let mut f = f;
let mut mb = znippy_common::arrow::buffer::MutableBuffer::with_capacity(len as usize);
let mut v = Vec::with_capacity(len as usize);
f.read_to_end(&mut v)
.with_context(|| format!("reading {}", path.display()))?;
mb.extend_from_slice(&v);
Ok(Some(mb.into()))
}
}
/// The projection at `path`, or `None` if there is none — or if what is there
/// cannot be read as one. The second case is a warning and not an error on
/// purpose: the index is a cache the journal can rebuild, and a store that
/// refuses to open over a damaged cache would be a store that has stopped
/// answering over something it knows how to recover from. The rows redb still
/// holds are served; the rest are re-indexed from the journal by
/// `adopt_journal`, which sees them as "extent journaled, rows absent".
/// Returns the projection and two wall-clock figures in milliseconds: the map
/// (open + mmap, which is O(1)) and the decode-plus-stree build (which is O(n)
/// in the oid count and is the whole of a warm open's real cost). They are
/// logged by the open, so the number the next lever has to beat is in the
/// server log rather than in a bench nobody reruns.
fn load_persisted<S: ObjectIndex>(path: &Path) -> Option<(Arc<S>, u128, u128)> {
let t0 = std::time::Instant::now();
match map_persisted(path) {
Ok(None) => None,
Ok(Some(buf)) => {
let map_ms = t0.elapsed().as_millis();
let t1 = std::time::Instant::now();
match S::from_persisted(buf) {
Ok(s) => Some((Arc::new(s), map_ms, t1.elapsed().as_millis())),
Err(e) => {
tracing::warn!(
target: "znippy.git",
path = %path.display(),
error = %format!("{e:#}"),
"the persisted object index could not be read; opening without it and \
re-indexing from the journal — slower until the drain catches up, \
never wrong"
);
None
}
}
}
Err(e) => {
tracing::warn!(
target: "znippy.git",
path = %path.display(),
error = %format!("{e:#}"),
"the persisted object index could not be opened; opening without it"
);
None
}
}
}
impl<S: ObjectIndex> ObjectIndex for ObjectReadStack<S> {
/// Build a stack over `entries` with an in-memory tail. Every entry lands in
/// redb and the projection is built from it, so the stack starts fully
/// absorbed — which is what makes it directly comparable with the two Arrow
/// arms on the layout benchmark.
fn build(entries: &[IndexEntry]) -> Result<Self> {
let stack = Self::in_memory(RebuildTriggers::default())?;
stack.append(entries)?;
stack.rebuild()?;
Ok(stack)
}
/// Projection first; **a miss falls through to redb**, which always knows.
/// `None` here means the tail said no, never that the stree said no.
fn lookup(&self, oid: &[u8]) -> Option<IndexRow> {
if let Some(row) = self.projection.read().expect("projection lock").lookup(oid) {
return Some(row);
}
if self.projection_is_complete() {
self.absent.fetch_add(1, Ordering::AcqRel);
return None;
}
let mut out = [None];
// A tail read that errors is reported as absent to this signature, which
// cannot carry an error. Callers that need the distinction use
// `stats().absent` against their own miss count, or `fill_from_tail`
// through `lookup_batch`'s caller. A corrupt tail is a `rebuild()`
// failure long before it is a wrong lookup.
let _ = self.fill_from_tail(&[oid], &mut out);
out[0]
}
/// Two phases, and the second is not optional: one pipelined stree pass, then
/// **one** redb transaction for every slot the projection left empty.
fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>> {
let mut out = self
.projection
.read()
.expect("projection lock")
.lookup_batch(oids);
if out.iter().any(Option::is_none) {
if self.projection_is_complete() {
self.absences_only(&out);
} else {
let _ = self.fill_from_tail(oids, &mut out);
}
}
out
}
/// The floor path — oid → ordinal, no payload column touched. The tail is
/// still consulted for the misses, because a miss is still not an answer;
/// a tail-served ordinal carries [`TAIL_ORDINAL_BIT`].
fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
let mut out = self
.projection
.read()
.expect("projection lock")
.ordinals_batch(oids);
if out.iter().any(Option::is_none) {
if self.projection_is_complete() {
let n = out.iter().filter(|s| s.is_none()).count() as u64;
self.absent.fetch_add(n, Ordering::AcqRel);
return out;
}
let mut rows: Vec<Option<IndexRow>> = out
.iter()
.map(|o| {
o.map(|ordinal| IndexRow {
ordinal,
offset: 0,
len: 0,
obj_type: ObjType::Blob,
uncompressed_size: 0,
delta_base: 0,
})
})
.collect();
let _ = self.fill_from_tail(oids, &mut rows);
for (slot, row) in out.iter_mut().zip(&rows) {
*slot = row.map(|r| r.ordinal);
}
}
out
}
/// The partial-row path — byte extent only.
fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>> {
let mut out = self
.projection
.read()
.expect("projection lock")
.extents_batch(oids);
if out.iter().any(Option::is_none) {
if self.projection_is_complete() {
let n = out.iter().filter(|s| s.is_none()).count() as u64;
self.absent.fetch_add(n, Ordering::AcqRel);
return out;
}
let mut rows: Vec<Option<IndexRow>> = out
.iter()
.map(|e| {
e.map(|(offset, len)| IndexRow {
ordinal: 0,
offset,
len,
obj_type: ObjType::Blob,
uncompressed_size: 0,
delta_base: 0,
})
})
.collect();
let _ = self.fill_from_tail(oids, &mut rows);
for (slot, row) in out.iter_mut().zip(&rows) {
*slot = row.map(|r| (r.offset, r.len));
}
}
out
}
/// The quota gate, over the whole **repository** and without a scan of the
/// tail: the projection's column scan plus the rows appended after it.
/// Exact because the archive is append-only — a row the projection already
/// holds can never gain or lose bytes, so the two terms cannot overlap.
fn sum_uncompressed(&self) -> u64 {
self.projection
.read()
.expect("projection lock")
.sum_uncompressed()
+ self.unabsorbed_size.load(Ordering::Acquire)
}
/// [`sum_uncompressed`](Self::sum_uncompressed)'s argument, per type.
fn count_type(&self, t: ObjType) -> usize {
self.projection
.read()
.expect("projection lock")
.count_type(t)
+ self.unabsorbed_types[t.code() as usize].load(Ordering::Acquire) as usize
}
fn name(&self) -> &'static str {
"ObjectReadStack"
}
/// Positional rows are the **projection's** rows. A tail ordinal carries
/// [`TAIL_ORDINAL_BIT`] and is a row address in no table — asking for it
/// positionally is the same programming error as indexing past `len()`,
/// and fails the same way, out loud.
fn oid_at(&self, ordinal: u32) -> Vec<u8> {
assert!(!is_tail_ordinal(ordinal), "ordinal {ordinal:#x} is tail-served, not positional");
self.projection.read().expect("projection lock").oid_at(ordinal)
}
fn entry_at(&self, ordinal: u32) -> IndexEntry {
assert!(!is_tail_ordinal(ordinal), "ordinal {ordinal:#x} is tail-served, not positional");
self.projection.read().expect("projection lock").entry_at(ordinal)
}
fn extent_at(&self, ordinal: u32) -> (u64, u64) {
assert!(!is_tail_ordinal(ordinal), "ordinal {ordinal:#x} is tail-served, not positional");
self.projection.read().expect("projection lock").extent_at(ordinal)
}
/// Objects in the **repository**, not in the projection — the stack answers
/// for the whole repository and this has to agree with what `lookup` will
/// resolve. [`projection_len`](Self::projection_len) is the other number.
fn len(&self) -> usize {
self.total_rows.load(Ordering::Acquire) as usize
}
fn ipc_bytes(&self) -> usize {
self.projection.read().expect("projection lock").ipc_bytes()
}
/// The projection's resident bytes. The tail is on disk (or in redb's own
/// page cache) and is not counted here — counting it would make this
/// incomparable with the two Arrow arms, which is the number's only use.
fn resident_bytes(&self) -> usize {
self.projection
.read()
.expect("projection lock")
.resident_bytes()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::arms::DEFAULT_REDB_CACHE_BYTES;
use crate::index_layout::{FourTables, synthetic_entries};
type Stack = ObjectReadStack<OneTableFourColumns>;
/// A stack with the thresholds off, so a test decides when a rebuild happens.
fn manual() -> Stack {
Stack::in_memory(RebuildTriggers::manual()).expect("in-memory stack")
}
/// Seal `sealed` into the projection, then append `tail` without absorbing
/// it. The returned stack is deliberately in the state the whole design is
/// about: a projection that is a strict subset of the truth.
fn split_stack(sealed: &[IndexEntry], tail: &[IndexEntry]) -> Stack {
let s = manual();
s.append(sealed).expect("seal half");
s.rebuild().expect("rebuild");
s.append(tail).expect("append tail");
s
}
/// **The property the whole stack rests on: incomplete, never wrong.**
///
/// The projection covers 300 of 500 objects. Asserted on applied output —
/// every one of the five facts of all 500 rows, through both the serial and
/// the batch path — plus the two things that would make the guard hollow:
/// that the projection really is short (its own `lookup` misses all 200 tail
/// objects, so the fall-through is genuinely being exercised) and that every
/// projection hit is byte-identical to what a complete index built over all
/// 500 returns.
///
/// Seen RED by making `lookup_batch`'s phase 2 conditional on
/// `out.iter().all(Option::is_none)` instead of `any` — i.e. falling through
/// only when the projection answered nothing: "the stack lost tail object
/// e50298d4cb83164fcbbd2a0d4fd99ebb43111e93, batch path". Restored.
///
/// Seen RED a second time by returning `None` from `lookup` on a projection
/// miss with no fall-through: "the stack lost tail object
/// e50298d4cb83164fcbbd2a0d4fd99ebb43111e93, serial path". Restored.
///
/// Seen RED a third time, for `delta_base`, by having [`encode_row`] write
/// a literal `0` into bytes 33..41 — the shape of a column that exists in
/// the struct and is never persisted: "the delta base of
/// b2af964cd8a34795ce5ad114d4e2a2144cf2a6b1 did not survive the projection
/// path / left: 0 / right: 6186". **Six** of this module's tests went red
/// on that one edit, on both sides of the fall-through, which is what says
/// the tail encoding is load-bearing rather than incidental. Restored.
#[test]
fn the_projection_is_incomplete_but_never_wrong() {
let all = synthetic_entries(500, 20, 0xC0FFEE);
let (sealed, tail) = all.split_at(300);
let stack = split_stack(sealed, tail);
assert_eq!(stack.projection_len(), 300, "the projection must be short");
assert_eq!(stack.len(), 500, "the stack answers for the repository");
// The tail half has to carry delta bases, or the `delta_base` assertion
// below is an assertion about zero on both sides of the fall-through.
assert!(
tail.iter().filter(|e| e.delta_base != 0).count() >= 10,
"the un-absorbed half carries no delta bases — the tail encoding would be untested"
);
// Two references. `sealed_only` is what the projection is *supposed* to
// be — an index over exactly the 300 sealed objects, so its ordinals are
// the ranks the projection must reproduce. `complete` is the index the
// stack must be indistinguishable from on the five facts.
let sealed_only = FourTables::build(sealed).expect("sealed-only index");
let complete = FourTables::build(&all).expect("complete index");
// The projection alone must MISS every tail object — otherwise the
// fall-through below proves nothing.
{
let proj = stack.projection.read().unwrap();
for e in tail {
assert!(
proj.lookup(&e.oid).is_none(),
"the projection already holds {} — this test would be vacuous",
hex::encode(&e.oid)
);
}
for e in sealed {
assert!(
proj.lookup(&e.oid).is_some(),
"the projection lost sealed object {}",
hex::encode(&e.oid)
);
}
}
let refs: Vec<&[u8]> = all.iter().map(|e| e.oid.as_slice()).collect();
let batched = stack.lookup_batch(&refs);
for (i, e) in all.iter().enumerate() {
let via_serial = stack.lookup(&e.oid).unwrap_or_else(|| {
panic!(
"the stack lost {} object {}, serial path",
if i < 300 { "sealed" } else { "tail" },
hex::encode(&e.oid)
)
});
let via_batch = batched[i].unwrap_or_else(|| {
panic!(
"the stack lost {} object {}, batch path",
if i < 300 { "sealed" } else { "tail" },
hex::encode(&e.oid)
)
});
assert_eq!(via_serial, via_batch, "serial and batch disagree");
// The five facts, against the entry that was appended.
assert_eq!(via_serial.offset, e.offset);
assert_eq!(via_serial.len, e.len);
assert_eq!(via_serial.obj_type, e.obj_type);
assert_eq!(via_serial.uncompressed_size, e.uncompressed_size);
assert_eq!(
via_serial.delta_base,
e.delta_base,
"the delta base of {} did not survive the {} path",
hex::encode(&e.oid),
if i < 300 { "projection" } else { "tail" }
);
// A HIT in the projection is valid: the five facts are the complete
// index's, and the ordinal is the rank within the generation the
// projection covers — which is what an ordinal means.
let truth = complete.lookup(&e.oid).expect("complete index has it");
assert_eq!(via_serial.offset, truth.offset);
assert_eq!(via_serial.len, truth.len);
assert_eq!(via_serial.obj_type, truth.obj_type);
assert_eq!(via_serial.uncompressed_size, truth.uncompressed_size);
assert_eq!(via_serial.delta_base, truth.delta_base);
if i < 300 {
assert!(
!is_tail_row(&via_serial),
"sealed row wearing a tail ordinal"
);
assert_eq!(
via_serial.ordinal,
sealed_only.lookup(&e.oid).unwrap().ordinal,
"the projection's ordinal for {} is not its rank in the generation the \
projection covers",
hex::encode(&e.oid)
);
} else {
assert!(
is_tail_row(&via_serial),
"tail row {} has no tail ordinal",
hex::encode(&e.oid)
);
}
}
}
/// A miss in the tree is not an answer, and an absence is.
///
/// Seen RED by having `fill_from_tail` skip its `objects.get` and count every
/// unfilled slot as absent: the tail-served block came back empty —
/// "assertion `left == right` failed / left: 0 / right: 80". Restored.
#[test]
fn only_a_tail_miss_is_absent_and_only_a_tail_hit_counts() {
let all = synthetic_entries(200, 20, 7);
let (sealed, tail) = all.split_at(120);
let stack = split_stack(sealed, tail);
let nowhere = synthetic_entries(64, 20, 0x00AB_5E47);
let mut refs: Vec<&[u8]> = Vec::new();
refs.extend(sealed.iter().map(|e| e.oid.as_slice()));
refs.extend(tail.iter().map(|e| e.oid.as_slice()));
refs.extend(nowhere.iter().map(|e| e.oid.as_slice()));
let rows = stack.lookup_batch(&refs);
assert_eq!(rows[..120].iter().filter(|r| r.is_some()).count(), 120);
assert_eq!(rows[120..200].iter().filter(|r| r.is_some()).count(), 80);
assert!(
rows[200..].iter().all(Option::is_none),
"an object in no repository resolved to a row"
);
let st = stack.stats();
assert_eq!(st.tail_hits, 80, "tail-served lookups miscounted");
assert_eq!(st.absent, 64, "genuine absences miscounted");
}
/// **Append-only, enforced.** The refusal is what makes a stree hit
/// trustworthy; if a fact could be rewritten, a stale projection would return
/// the old one.
///
/// **Identity is refused; placement is not.** REWRITTEN 2026-08-14 — the
/// old form compared `offset`, `len` and `delta_base` too, and those say
/// where a copy landed rather than what the object is. A re-pushed pack
/// places the same oid at a new offset, which is the ordinary case, so the
/// guard refused normal pushes: the ingest aborted and the connection
/// dropped with `client_told=false`, leaving the client with
/// "send-pack: unexpected disconnect" and no reason. It failed
/// `bare_lifecycle` and `endurance` on all four znippy bench columns and
/// bit any `--force` push that resent a known object.
///
/// The old comparison's stated fear — a projection carrying "a base offset
/// the archive never wrote" — is answered better by keeping the OLD row:
/// the archive is append-only and `gc` truncates nothing, so the first
/// writer's offset and base still address bytes that are still there.
///
/// Seen RED by comparing `offset` again: "a re-pushed pack at a new offset
/// was refused" — which is precisely the production failure, reproduced.
///
/// Seen RED the other way by dropping `old.obj_type != e.obj_type`: "an oid
/// that changed type was accepted" — the same oid describing a different
/// object, waved through.
#[test]
fn an_append_only_violation_is_refused() {
let entries = synthetic_entries(16, 20, 11);
let stack = manual();
stack.append(&entries).expect("first append");
// Re-appending the identical batch is a no-op, not an error: a re-pushed
// pack repeats objects.
stack
.append(&entries)
.expect("identical re-append is idempotent");
assert_eq!(
stack.len(),
16,
"an idempotent re-append changed the row count"
);
// ★ THE PRODUCTION CASE: the same oid, re-pushed, lands elsewhere.
// This must be a NO-OP, and the first writer's placement must stand.
let mut moved = entries[3].clone();
moved.offset += 1;
stack
.append(std::slice::from_ref(&moved))
.expect("a re-pushed pack at a new offset was refused");
let row = stack.lookup(&entries[3].oid).expect("still there");
assert_eq!(
row.offset, entries[3].offset,
"the second placement overwrote the first — the projection now names \
an offset that is not where the first writer put the bytes"
);
assert_eq!(stack.len(), 16, "a re-placement added a row");
// ★ IDENTITY still refuses: same oid, different object.
let mut retyped = entries[4].clone();
retyped.uncompressed_size += 1;
let err = stack
.append(std::slice::from_ref(&retyped))
.expect_err("an oid that changed size was accepted");
assert!(
err.to_string().contains("identity violation"),
"wrong error: {err}"
);
// ★ A DIFFERENT DELTA BASE IS PLACEMENT TOO, and the first row stands.
//
// The old form refused this, on the reasoning that a row naming a base
// the archive never wrote would be left behind. Keeping the OLD row
// answers that completely: the base the FIRST writer recorded is still
// in the archive, because the blob is append-only and `gc` truncates
// nothing. The second pack's base is simply not adopted.
//
// Refusing it instead is what broke real pushes — a re-pushed pack
// re-deltas against whatever is in ITS window, so a differing base is
// the ordinary case, not evidence of corruption.
let based = entries
.iter()
.find(|e| e.delta_base != 0)
.expect("the fixture must carry a delta");
let mut rebased = based.clone();
rebased.delta_base += 8;
stack
.append(std::slice::from_ref(&rebased))
.expect("a re-pushed pack that re-deltaed was refused");
assert_eq!(
stack.lookup(&based.oid).unwrap().delta_base,
based.delta_base,
"the second pack's base was adopted — the projection now names a base \
the FIRST writer never recorded, which is the thing the old guard \
was right to fear"
);
}
/// The volume trigger fires on appended bytes and the miss trigger on
/// tail-served lookups, and neither fires on a genuine absence.
///
/// Seen RED by counting `absent` into `tail_hits` in `fill_from_tail`:
/// "assertion `left == right` failed: absences tripped the miss trigger /
/// left: Some(TailHits(1000)) / right: None". Restored.
#[test]
fn each_trigger_fires_on_its_own_signal() {
// ── volume ──
let entries = synthetic_entries(64, 20, 21);
let bytes: u64 = entries.iter().map(|e| e.len).sum();
let stack = Stack::in_memory(RebuildTriggers {
tail_bytes: bytes,
..RebuildTriggers::manual()
})
.unwrap();
let g0 = stack.stats().generation;
let reason = stack.append(&entries).unwrap();
assert!(
matches!(reason, Some(RebuildReason::TailBytes(_))),
"the volume trigger did not fire at exactly its threshold: {reason:?}"
);
assert_eq!(stack.projection_len(), 64, "the rebuild absorbed nothing");
assert_eq!(stack.stats().generation, g0 + 1);
assert_eq!(
stack.stats().tail_bytes,
0,
"the byte counter was not reset"
);
// ── misses ──
let all = synthetic_entries(200, 20, 22);
let (sealed, tail) = all.split_at(100);
// A flat floor of 100 misses: `tail_hits_per_row` 0 means the threshold
// is exactly the floor, which makes the arithmetic in this test the
// trigger's and not the ratio's.
let stack = Stack::in_memory(RebuildTriggers {
tail_hits_per_row: 0.0,
min_tail_hits: 100,
tail_bytes: 0,
fold_ratio: 0.0,
})
.unwrap();
stack.append(sealed).unwrap();
stack.rebuild().unwrap();
stack.append(tail).unwrap();
let g1 = stack.stats().generation;
assert_eq!(stack.projection_len(), 100);
// 1000 lookups of oids in NO repository: 1000 tail misses, zero
// tail-served hits, and no rebuild may follow.
let nowhere = synthetic_entries(1000, 20, 23);
let refs: Vec<&[u8]> = nowhere.iter().map(|e| e.oid.as_slice()).collect();
assert!(stack.lookup_batch(&refs).iter().all(Option::is_none));
assert_eq!(stack.stats().absent, 1000);
assert_eq!(
stack.rebuild_due(),
None,
"absences tripped the miss trigger"
);
stack.maybe_rebuild().unwrap();
assert_eq!(
stack.stats().generation,
g1,
"1000 absences must not rebuild"
);
assert_eq!(stack.projection_len(), 100);
// 100 lookups the tail DOES answer: the trigger fires.
let refs: Vec<&[u8]> = tail.iter().map(|e| e.oid.as_slice()).collect();
assert_eq!(stack.lookup_batch(&refs).iter().flatten().count(), 100);
assert_eq!(
stack.rebuild_due(),
Some(RebuildReason::TailHits(100)),
"100 tail-served lookups did not trip a threshold of 100"
);
assert!(stack.maybe_rebuild().unwrap().is_some());
assert_eq!(
stack.projection_len(),
200,
"the rebuild did not absorb the tail"
);
assert_eq!(stack.stats().generation, g1 + 1);
}
/// The two column scans answer for the **repository**, not for the
/// projection — including the rows the projection has not absorbed, and
/// without scanning the tail to find them.
///
/// Seen RED by dropping the `unabsorbed_size` term from `sum_uncompressed`:
/// "assertion `left == right` failed: sum_uncompressed answered for the
/// projection, not the repository / left: 1179629 / right: 1915881".
/// Restored.
///
/// The rejected-append half of this test is what pins the *ordering* in
/// `append`: the counters are accumulated locally and applied after the
/// commit, because a `bail!` half-way through a batch rolls redb back and
/// counters bumped inside the loop would not roll back with it. That one is
/// fixed by construction rather than watched — there is one place the
/// counters are written and it is after `commit()`.
#[test]
fn the_column_scans_answer_for_the_repository_not_the_projection() {
let all = synthetic_entries(300, 20, 71);
let (sealed, tail) = all.split_at(180);
let stack = split_stack(sealed, tail);
let complete = OneTableFourColumns::build(&all).unwrap();
assert!(
stack.projection.read().unwrap().sum_uncompressed() < complete.sum_uncompressed(),
"the projection already sums to the whole repository — this test would be vacuous"
);
assert_eq!(
stack.sum_uncompressed(),
complete.sum_uncompressed(),
"sum_uncompressed answered for the projection, not the repository"
);
for t in ObjType::ALL {
assert_eq!(
stack.count_type(t),
complete.count_type(t),
"count_type({}) answered for the projection",
t.as_str()
);
}
// A rejected append must not move either aggregate.
let mut rewritten = all[0].clone();
rewritten.uncompressed_size += 1_000_000;
let batch = [all[7].clone(), rewritten];
assert!(stack.append(&batch).is_err());
assert_eq!(stack.sum_uncompressed(), complete.sum_uncompressed());
// And a rebuild leaves them where they were.
stack.rebuild().unwrap();
assert_eq!(stack.sum_uncompressed(), complete.sum_uncompressed());
for t in ObjType::ALL {
assert_eq!(stack.count_type(t), complete.count_type(t));
}
}
/// The two narrow batch paths fall through to the tail exactly as the
/// full-row one does. A miss is not an answer on those either.
///
/// Seen RED by returning the projection's `ordinals_batch` unchanged:
/// "assertion `left == right` failed: the ordinal path lost 120 of 300
/// objects / left: 180 / right: 300". Restored.
#[test]
fn the_narrow_batch_paths_also_fall_through() {
let all = synthetic_entries(300, 20, 81);
let (sealed, tail) = all.split_at(180);
let stack = split_stack(sealed, tail);
let refs: Vec<&[u8]> = all.iter().map(|e| e.oid.as_slice()).collect();
let ordinals = stack.ordinals_batch(&refs);
assert_eq!(
ordinals.iter().filter(|o| o.is_some()).count(),
300,
"the ordinal path lost {} of 300 objects",
300 - ordinals.iter().filter(|o| o.is_some()).count()
);
for (o, e) in ordinals[180..].iter().zip(tail) {
assert!(
o.unwrap() & TAIL_ORDINAL_BIT != 0,
"tail object {} got a projection ordinal",
hex::encode(&e.oid)
);
}
let extents = stack.extents_batch(&refs);
for (x, e) in extents.iter().zip(&all) {
assert_eq!(
*x,
Some((e.offset, e.len)),
"extent lost or wrong for {}",
hex::encode(&e.oid)
);
}
}
/// A rebuild re-derives every ordinal as the new lexicographic rank, and the
/// tail's high-bit ordinals disappear as their rows are absorbed. This is the
/// generation caveat, asserted rather than only documented.
///
/// Seen RED by having `rebuild` compute the fresh projection and never swap
/// it in: "an absorbed row kept its tail ordinal". Restored.
#[test]
fn a_rebuild_re_derives_ordinals_and_clears_the_tail_bit() {
let all = synthetic_entries(120, 32, 31);
let (sealed, tail) = all.split_at(60);
let stack = split_stack(sealed, tail);
let before = stack.lookup(&tail[0].oid).unwrap();
assert!(is_tail_row(&before));
stack.rebuild().unwrap();
let after = stack.lookup(&tail[0].oid).unwrap();
assert!(
!is_tail_row(&after),
"an absorbed row kept its tail ordinal"
);
assert_eq!(after.offset, before.offset, "absorption changed a fact");
assert_eq!(after.len, before.len);
assert_eq!(after.obj_type, before.obj_type);
assert_eq!(after.uncompressed_size, before.uncompressed_size);
assert_eq!(after.delta_base, before.delta_base);
// The delta base is an ARCHIVE OFFSET and absorption must not touch it.
// An ordinal in this column would have had to be re-derived by the
// rebuild — silently, and to a different number, because the rebuild
// re-ranks every row. That is §13's argument, asserted on a row that
// actually carries a base rather than on the 0 sentinel.
let based = tail
.iter()
.find(|e| e.delta_base != 0)
.expect("the tail half must carry at least one delta");
let row = stack.lookup(&based.oid).unwrap();
assert_eq!(
row.delta_base,
based.delta_base,
"the rebuild moved {}'s delta base from {} to {}",
hex::encode(&based.oid),
based.delta_base,
row.delta_base
);
assert_ne!(
row.delta_base, row.ordinal as u64,
"a delta base that equals a row ordinal is the mistake §13 forbids"
);
let complete = OneTableFourColumns::build(&all).unwrap();
assert_eq!(
after.ordinal,
complete.lookup(&tail[0].oid).unwrap().ordinal,
"the rebuilt ordinal is not the lexicographic rank"
);
}
/// **The crash-recovery diff is exact at the pack boundary.**
///
/// Three packs' worth of rows laid end to end, the middle one never appended:
/// [`ObjectReadStack::extents_with_rows`] must answer `true, false, true`.
/// The middle pack's extent is the interesting one — it is bounded on both
/// sides by rows that *are* in the tail, so an off-by-one in either direction
/// reports it absorbed, its objects are never re-queued after a restart, and a
/// durable pack becomes unreadable for ever.
///
/// Asserted per extent, not as a count.
///
/// Seen RED by `let p = starts.partition_point(|&s| s <= offset)` →
/// `partition_point(|&s| s < offset)`, i.e. placing a row against the
/// *previous* extent when it lands on a pack's first byte: "left: [false] /
/// right: [true]" on the single-row extent. Restored.
///
/// Seen RED a second time by `offset < start + len` → `offset <= start + len`:
/// "a row one byte past pack B's extent was counted as B's — left: [true,
/// true], right: [true, false]". **That mutation stayed green against the
/// four-pack question**, and the reason is worth writing down: when every
/// extent is in the question and the packs are adjacent, the row at B's end
/// offset is also C's first row, and the binary search assigns it to C before
/// the bound is ever consulted. The bound only bites when a pack's successor
/// is *not* in the question — a journal whose tail was torn, which is the
/// state this whole diff exists to survive. The `[..2]` case below is
/// therefore not an extra assertion, it is the one that tests the bound at
/// all. Restored.
#[test]
fn the_crash_recovery_diff_is_exact_at_the_pack_boundary() {
let all = synthetic_entries(400, 20, 0xB17_5E7);
// Four packs, back to back, the way `SafeWriter` appends them.
let bounds = [(0usize, 100usize), (100, 200), (200, 300), (300, 400)];
let extents: Vec<(u64, u64)> = bounds
.iter()
.map(|&(a, b)| {
let start = all[a].offset;
let end = all[b - 1].offset + all[b - 1].len;
(start, end - start)
})
.collect();
assert_eq!(
extents[1].0,
extents[0].0 + extents[0].1,
"the fixture's packs must be adjacent or the boundary is not under test"
);
let stack = manual();
for (i, &(a, b)) in bounds.iter().enumerate() {
if i != 1 {
stack.append(&all[a..b]).unwrap();
}
}
assert_eq!(stack.len(), 300, "the middle pack must be the missing one");
let hit = stack.extents_with_rows(&extents).unwrap();
assert_eq!(
hit,
vec![true, false, true, true],
"pack B has no rows in the tail and was reported absorbed"
);
// **The containment bound, made observable.** Asking only about A and B
// is the torn-journal case: pack C's rows are in the tail but its extent
// is not in the question, and C's first row sits at exactly B's end
// offset. That row belongs to C and must not answer for B.
assert_eq!(
stack.extents_with_rows(&extents[..2]).unwrap(),
vec![true, false],
"a row one byte past pack B's extent was counted as B's"
);
// A zero-length extent has no rows by construction, and an extent past
// everything the tail knows about has none either.
let past = all[399].offset + all[399].len;
assert_eq!(
stack.extents_with_rows(&[(0, 0), (past, 4096)]).unwrap(),
vec![false, false]
);
// And an extent that covers exactly one row is that row's pack.
assert_eq!(
stack
.extents_with_rows(&[(all[7].offset, all[7].len)])
.unwrap(),
vec![true]
);
}
/// A file-backed tail survives being closed and reopened, and the reopened
/// stack warm-starts a projection over everything — including rows that were
/// only in the tail when the process went away.
///
/// Seen RED by having `from_db` build the projection from an empty slice
/// instead of `scan(&db)`: "reopened projection covers 0 of 90". Restored.
#[test]
fn a_file_backed_tail_reopens_warm() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("objects.tail.redb");
let all = synthetic_entries(90, 20, 41);
{
let stack =
ObjectReadStack::<OneTableFourColumns>::open(
&path,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
stack.append(&all[..40]).unwrap();
stack.rebuild().unwrap();
stack.append(&all[40..]).unwrap();
assert_eq!(stack.projection_len(), 40);
}
let stack =
ObjectReadStack::<OneTableFourColumns>::open(
&path,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
).unwrap();
// **What "warm" means since 2026-09-04.** This used to assert
// `projection_len() == 90`: the old open rebuilt the projection from a
// full redb scan, so everything was in it. Now the open MAPS the folded
// index (the 40 rows the rebuild persisted) and leaves the 50 that
// arrived after it in redb, where the fall-through serves them — the
// same routing an un-absorbed row always had. The repository is still
// 90 rows and every one of them answers; what changed is that nothing
// was rebuilt to make that so, and `persisted_loads` is the applied
// output that says so.
assert_eq!(stack.len(), 90, "the repository is {} of 90 rows", stack.len());
assert_eq!(stack.projection_len(), 40, "the mapped index holds what the fold wrote");
assert_eq!(stack.stats().persisted_loads, 1, "the open must have mapped, not rebuilt");
assert_eq!(stack.stats().unabsorbed_rows, 50, "the 50 later rows are redb's to serve");
for e in &all {
let row = stack
.lookup(&e.oid)
.unwrap_or_else(|| panic!("reopen lost {}", hex::encode(&e.oid)));
assert_eq!(row.offset, e.offset);
assert_eq!(row.uncompressed_size, e.uncompressed_size);
assert_eq!(row.delta_base, e.delta_base, "reopen lost a delta base");
}
assert!(
all.iter().filter(|e| e.delta_base != 0).count() >= 5,
"the fixture must carry delta bases across the reopen"
);
}
// ── the fold ─────────────────────────────────────────────────────────────
/// A file-backed stack over `all[..40]` folded, `all[40..]` appended after.
/// Returns the tail path and the index path.
fn folded_then_appended(
dir: &Path,
all: &[IndexEntry],
split: usize,
) -> (PathBuf, PathBuf, ObjectReadStack<OneTableFourColumns>) {
let tail = dir.join("objects.tail");
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
stack.append(&all[..split]).unwrap();
stack.rebuild().unwrap();
stack.append(&all[split..]).unwrap();
let index = stack.persisted_path().expect("the default arm persists").to_path_buf();
(tail, index, stack)
}
/// How many rows redb holds right now — the thing the fold exists to keep
/// small, read straight off the table so the test cannot be fooled by the
/// stack's own bookkeeping.
fn redb_rows(stack: &ObjectReadStack<OneTableFourColumns>) -> u64 {
let r = stack.tail.begin_read().unwrap();
r.open_table(OBJECTS).unwrap().len().unwrap()
}
/// **A rebuild on a file-backed stack is a fold**: the projection lands in
/// `objects.index`, and redb forgets the rows it holds. Asserted on the
/// file and on redb's own row count, not on the stack's counters.
///
/// Seen RED by skipping the `objects.remove` loop in `fold_locked`: redb
/// kept 40 rows after the fold ("redb must hold only what arrived after the
/// fold / left: 40 / right: 0"). Seen RED by skipping `write_atomic`: the
/// index file was absent. Restored.
#[test]
fn a_rebuild_on_a_file_backed_stack_is_a_fold() {
let dir = tempfile::tempdir().unwrap();
let all = synthetic_entries(90, 20, 43);
let (_tail, index, stack) = folded_then_appended(dir.path(), &all, 40);
assert!(index.is_file(), "the fold must persist the projection at {}", index.display());
assert!(
fs::metadata(&index).unwrap().len() > 40 * 61,
"the persisted index must carry at least the rows' own bytes"
);
assert_eq!(stack.stats().folds, 1);
assert_eq!(redb_rows(&stack), 50, "redb must hold only what arrived after the fold");
assert_eq!(stack.projection_len(), 40);
assert_eq!(stack.len(), 90);
assert!(!stack.projection_is_complete());
for e in &all {
let row = stack.lookup(&e.oid).unwrap_or_else(|| panic!("lost {}", hex::encode(&e.oid)));
assert_eq!((row.offset, row.len, row.delta_base), (e.offset, e.len, e.delta_base));
}
// The tail rows are the tail's: served with the tail bit, as before.
assert!(is_tail_row(&stack.lookup(&all[80].oid).unwrap()));
assert!(!is_tail_row(&stack.lookup(&all[10].oid).unwrap()));
// A second fold takes the 50 with it and redb is empty.
stack.rebuild().unwrap();
assert_eq!(redb_rows(&stack), 0);
assert_eq!(stack.projection_len(), 90);
assert!(stack.projection_is_complete());
assert_eq!(stack.stats().folds, 2);
}
/// **Re-appending a folded row creates nothing.** `append` dedupes against
/// the projection AND redb, because on a persisting stack the folded rows
/// are in the projection only — a check against redb alone would insert
/// every re-pushed or re-indexed object a second time, into redb, and the
/// two layers would each hold it.
///
/// Seen RED before the projection check existed, on the real store:
/// `git_ops::tests::dropping_the_exploded_table_and_reopening_still_answers`
/// re-indexed a pack from the journal after a reopen and the row count
/// read **2704 for 1352 objects** — exactly twice. Seen RED here by
/// consulting redb only: `redb_rows` read 40 where 0 was expected and
/// `len()` read 80. Restored.
#[test]
fn re_appending_a_folded_row_creates_no_second_row() {
let dir = tempfile::tempdir().unwrap();
let all = synthetic_entries(40, 20, 50);
let tail = dir.path().join("objects.tail");
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
stack.append(&all).unwrap();
stack.rebuild().unwrap();
assert_eq!(redb_rows(&stack), 0, "precondition: everything folded");
// The same objects again — a re-push, or the journal re-indexing a
// pack after a reopen. Placement may differ; identity must agree.
let mut again = all.clone();
for e in &mut again {
e.offset += 1_000_000;
}
stack.append(&again).unwrap();
assert_eq!(redb_rows(&stack), 0, "a folded row re-appended must not land in redb");
assert_eq!(stack.len(), 40);
assert_eq!(stack.oids_in_order().unwrap().len(), 40);
// And the first writer's placement stands, as `append`'s contract says.
assert_eq!(stack.lookup(&all[3].oid).unwrap().offset, all[3].offset);
// The identity check still bites through the projection path: a
// different inflated size for a folded oid is refused, not inserted.
let mut lie = all[7].clone();
lie.uncompressed_size += 1;
let err = stack.append(&[lie]).unwrap_err().to_string();
assert!(err.contains("identity violation"), "{err}");
assert_eq!(redb_rows(&stack), 0);
}
/// **A warm open maps the index and rebuilds nothing.** The proof is
/// `persisted_loads == 1` plus a projection whose size is exactly what the
/// fold wrote — a rebuilt projection would hold all 90.
///
/// Seen RED by making `load_persisted` return `None` unconditionally: the
/// open fell back to the scan-and-build, `persisted_loads` read 0, and the
/// projection held 90. Restored.
#[test]
fn a_warm_open_maps_the_folded_index_and_does_not_rebuild() {
let dir = tempfile::tempdir().unwrap();
let all = synthetic_entries(90, 20, 44);
let (tail, _index, stack) = folded_then_appended(dir.path(), &all, 60);
drop(stack);
let again = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
let s = again.stats();
assert_eq!(s.persisted_loads, 1, "the open must have taken the index off disk");
assert_eq!(s.folds, 0, "a warm open folds nothing — there is nothing to migrate");
assert_eq!(again.projection_len(), 60, "the mapped projection is what the fold wrote");
assert_eq!(again.len(), 90);
assert_eq!(s.unabsorbed_rows, 30);
assert_eq!(redb_rows(&again), 30);
// The column-scan aggregates are exact from the first read — computed
// over the redb remainder at open, not left at zero.
let expect_size: u64 = all.iter().map(|e| e.uncompressed_size).sum();
assert_eq!(again.sum_uncompressed(), expect_size);
for t in [ObjType::Commit, ObjType::Tree, ObjType::Blob, ObjType::OfsDelta] {
assert_eq!(
again.count_type(t),
all.iter().filter(|e| e.obj_type == t).count(),
"count_type({t:?}) over both layers"
);
}
for e in &all {
let row = again.lookup(&e.oid).unwrap_or_else(|| panic!("lost {}", hex::encode(&e.oid)));
assert_eq!(row.uncompressed_size, e.uncompressed_size);
}
}
/// **The crash window a fold leaves is reconciled at open.** The fold
/// renames the index into place and THEN forgets the rows in redb; a crash
/// between the two leaves every folded row in both. Staged here by putting
/// the folded rows back into redb by hand after a fold, exactly as that
/// crash would leave them.
///
/// Seen RED by removing the `dup` pass in `from_db`: `oids_in_order` came
/// back with 90 entries for a 60-object repository and `len()` read 90.
/// Restored.
#[test]
fn an_interrupted_fold_is_reconciled_at_the_next_open() {
let dir = tempfile::tempdir().unwrap();
let all = synthetic_entries(60, 20, 45);
let (tail, _index, stack) = folded_then_appended(dir.path(), &all, 30);
// Put the 30 folded rows back — the state after "renamed, then died".
{
let w = stack.tail.begin_write().unwrap();
{
let mut objects = w.open_table(OBJECTS).unwrap();
for (i, e) in all[..30].iter().enumerate() {
objects.insert(e.oid.as_slice(), encode_row(1000 + i as u64, e).as_slice()).unwrap();
}
}
w.commit().unwrap();
}
assert_eq!(redb_rows(&stack), 60, "precondition: every row is now in both layers");
drop(stack);
let again = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(redb_rows(&again), 30, "the duplicates must leave redb at open");
assert_eq!(again.len(), 60);
assert_eq!(again.tail_epoch_and_len().unwrap().1, 60);
let oids = again.oids_in_order().unwrap();
assert_eq!(oids.len(), 60, "each object once");
assert!(oids.windows(2).all(|w| w[0] < w[1]), "oid order, no repeats");
}
/// **The tail-wide answers cover both layers.** `oids_in_order` is the
/// merge of the persisted projection and redb; `tail_epoch_and_len` counts
/// both; `extents_with_rows` finds a row whichever layer holds it.
///
/// Seen RED by returning only redb's keys from `oids_in_order` on a
/// persisting stack: 50 of 90. Seen RED by scanning only redb in
/// `extents_with_rows`: the folded extent read `false`. Restored.
#[test]
fn the_tail_wide_answers_cover_both_layers() {
let dir = tempfile::tempdir().unwrap();
let all = synthetic_entries(90, 20, 46);
let (_tail, _index, stack) = folded_then_appended(dir.path(), &all, 40);
let mut expect: Vec<Vec<u8>> = all.iter().map(|e| e.oid.clone()).collect();
expect.sort();
assert_eq!(stack.oids_in_order().unwrap(), expect, "the merge of both layers, in oid order");
assert_eq!(stack.tail_epoch_and_len().unwrap().1, 90);
// One extent that only the persisted half can answer, one only redb can,
// and one nobody can.
let folded = &all[3];
let tailed = &all[77];
let nowhere = (u64::MAX / 2, 7u64);
let hit = stack
.extents_with_rows(&[(folded.offset, folded.len), (tailed.offset, tailed.len), nowhere])
.unwrap();
assert_eq!(hit, vec![true, true, false], "folded / tail / absent");
}
/// **The ordinal space is the merge of both layers, and says which is which.**
/// After a fold with nothing appended, every oid comes off the index and
/// none off redb; after more appends the tail rows are counted as such; in
/// every state the bytes equal `oids_in_order` concatenated.
///
/// Seen RED by returning the redb tail alone on a persisting stack:
/// `folded_rows` read 0 where 40 was expected and the space held 50 of 90
/// oids. Restored.
#[test]
fn the_ordinal_space_is_both_layers_with_provenance() {
let dir = tempfile::tempdir().unwrap();
let all = synthetic_entries(90, 20, 51);
let (_tail, _index, stack) = folded_then_appended(dir.path(), &all, 40);
let expect: Vec<u8> = stack.oids_in_order().unwrap().concat();
let space = stack.oids_raw().unwrap();
assert_eq!(space.raw, expect, "the space must be oids_in_order, concatenated");
assert_eq!((space.folded_rows, space.tail_rows), (40, 50));
assert_eq!(space.raw.len(), 90 * 20);
stack.rebuild().unwrap();
let space = stack.oids_raw().unwrap();
assert_eq!(space.raw, expect, "a fold does not change the space");
assert_eq!((space.folded_rows, space.tail_rows), (90, 0), "everything off the index now");
}
/// **On a persisting stack the volume trigger scales with the index.** With
/// a 1-byte floor and `fold_ratio = 1`, appending fewer object bytes than the
/// persisted index is long does not fold; appending past it does. On an
/// in-memory stack the same triggers fold at the floor, because there a
/// rebuild writes nothing and the fixed cadence is right.
///
/// Seen RED by making `volume_threshold` return the floor unconditionally:
/// the small append folded ("a small append must not fold a big index /
/// left: 2 / right: 1"). Restored.
#[test]
fn the_volume_trigger_scales_with_the_persisted_index() {
let dir = tempfile::tempdir().unwrap();
let mut all = synthetic_entries(400, 20, 52);
for e in &mut all {
e.len = 100;
}
let triggers = RebuildTriggers {
tail_bytes: 1,
fold_ratio: 1.0,
..RebuildTriggers::manual()
};
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&dir.path().join("objects.tail"),
triggers,
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
// The first append trips the 1-byte floor at once: there is no index yet
// to scale against. After it the index is 300 rows ≈ 16 kB of IPC.
stack.append(&all[..300]).unwrap();
let folds = stack.stats().folds;
assert_eq!(folds, 1, "the first batch folds at the floor");
let index_bytes = stack.ipc_bytes() as u64;
assert!(index_bytes > 10_000, "precondition: a real index, got {index_bytes} B");
// 50 rows × 100 B = 5 000 B of object bytes: well under the index size.
stack.append(&all[300..350]).unwrap();
assert_eq!(stack.stats().folds, folds, "a small append must not fold a big index");
assert_eq!(stack.stats().unabsorbed_rows, 50, "the 50 rows wait in redb");
// Enough more to pass the index size: 50 rows × 100 B on top → 10 000 B
// is still under 16 kB, so lift the lens for the last batch instead.
let mut big: Vec<IndexEntry> = all[350..400].to_vec();
for e in &mut big {
e.len = index_bytes; // one row is enough to cross it
}
stack.append(&big[..1]).unwrap();
assert_eq!(stack.stats().folds, folds + 1, "crossing the index size folds");
assert!(stack.projection_is_complete());
// The same triggers on an in-memory stack: the floor alone decides.
let mem = ObjectReadStack::<OneTableFourColumns>::in_memory(triggers).unwrap();
mem.append(&all[..300]).unwrap();
let g = mem.stats().generation;
mem.append(&all[300..350]).unwrap();
assert!(
mem.stats().generation > g,
"without persistence the fixed floor still rebuilds on every batch"
);
assert_eq!(mem.stats().folds, 0, "and nothing is written");
}
/// **`retain` rewrites the persisted index without the dead rows**, whichever
/// layer they were in, and the count it returns is both halves.
///
/// Seen RED by passing `None` instead of `Some(live)` to `fold_locked` from
/// `retain`: the dead folded oid was still answered after the call, and the
/// count was 1 where 2 rows had died. Restored.
#[test]
fn retain_rewrites_the_persisted_index_without_the_dead_rows() {
let dir = tempfile::tempdir().unwrap();
let all = synthetic_entries(90, 20, 47);
let (tail, index, stack) = folded_then_appended(dir.path(), &all, 40);
let dead_folded = all[5].oid.clone();
let dead_tailed = all[70].oid.clone();
let before = fs::metadata(&index).unwrap().len();
let epoch_before = stack.tail_epoch_and_len().unwrap().0;
let dropped = stack
.retain(&|oid: &[u8]| oid != dead_folded.as_slice() && oid != dead_tailed.as_slice())
.unwrap();
assert_eq!(dropped, 2, "one from each layer");
assert!(stack.lookup(&dead_folded).is_none(), "the folded dead row must be gone");
assert!(stack.lookup(&dead_tailed).is_none(), "the tail dead row must be gone");
assert_eq!(stack.len(), 88);
assert_eq!(stack.projection_len(), 88, "retain folds: everything live is in the index");
assert_eq!(redb_rows(&stack), 0);
assert!(
fs::metadata(&index).unwrap().len() > before,
"the index was rewritten (88 rows now, was 40)"
);
assert!(
stack.tail_epoch_and_len().unwrap().0 > epoch_before,
"a changed oid set is a changed epoch"
);
for e in all.iter().filter(|e| e.oid != dead_folded && e.oid != dead_tailed) {
assert!(stack.lookup(&e.oid).is_some(), "live row {} lost", hex::encode(&e.oid));
}
// And it survives a reopen off the rewritten file.
drop(stack);
let again = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(again.stats().persisted_loads, 1);
assert_eq!(again.len(), 88);
assert!(again.lookup(&dead_folded).is_none());
}
/// **A damaged index is ignored, loudly, and the store still opens.** The
/// rows redb has are served; the rest are the journal's to re-index. What
/// this asserts is the shape of the failure: open succeeds, the projection
/// is empty, redb's rows answer, and nothing panicked on a half-written file.
///
/// Seen RED by making `load_persisted` propagate the decode error: the open
/// failed. Restored.
#[test]
fn a_damaged_persisted_index_is_ignored_and_redb_still_answers() {
let dir = tempfile::tempdir().unwrap();
let all = synthetic_entries(60, 20, 48);
let (tail, index, stack) = folded_then_appended(dir.path(), &all, 30);
drop(stack);
// Truncate the index mid-stream: an IPC prefix, not an index.
let bytes = fs::read(&index).unwrap();
fs::write(&index, &bytes[..bytes.len() / 2]).unwrap();
let again = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(again.stats().persisted_loads, 0, "a damaged index must not count as loaded");
assert_eq!(again.len(), 30, "the 30 redb rows are what the store knows");
for e in &all[30..] {
assert!(again.lookup(&e.oid).is_some(), "redb row {} lost", hex::encode(&e.oid));
}
for e in &all[..30] {
assert!(
again.lookup(&e.oid).is_none(),
"a row only the damaged index held must read absent, so the journal re-indexes it"
);
}
// And the store keeps working: the next fold overwrites the damage.
again.append(&all[..30]).unwrap();
again.rebuild().unwrap();
assert_eq!(again.len(), 60);
drop(again);
let third = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(third.stats().persisted_loads, 1);
assert_eq!(third.len(), 60);
}
/// **A store from before persistence folds once, at its first open, and
/// never pays the scan again.** Staged by building a redb-only tail (the
/// old shape: every row in redb, no index file) and opening it.
///
/// Seen RED by removing the cold-open fold in `from_db`: `folds` read 0,
/// the index file was absent, and a second open rebuilt from redb again
/// (`persisted_loads == 0`). Restored.
#[test]
fn a_cold_open_of_a_pre_fold_store_folds_it_once() {
let dir = tempfile::tempdir().unwrap();
let tail = dir.path().join("objects.tail");
let all = synthetic_entries(70, 20, 49);
// The pre-2026-09-04 shape, written by hand: rows in redb, stamped with
// this layout, no index anywhere.
{
let db = Database::create(&tail).unwrap();
let w = db.begin_write().unwrap();
{
let mut objects = w.open_table(OBJECTS).unwrap();
for (i, e) in all.iter().enumerate() {
objects.insert(e.oid.as_slice(), encode_row(i as u64, e).as_slice()).unwrap();
}
let mut meta = w.open_table(META).unwrap();
meta.insert(META_TAIL_FORMAT, TAIL_FORMAT).unwrap();
meta.insert(META_ARRIVAL_SEQ, all.len() as u64).unwrap();
}
w.commit().unwrap();
}
let index = tail.with_extension("index");
assert!(!index.exists(), "precondition: no persisted index");
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(stack.stats().persisted_loads, 0, "a cold open has nothing to map");
assert_eq!(stack.stats().folds, 1, "and folds exactly once");
assert!(index.is_file(), "the fold must have written {}", index.display());
assert_eq!(redb_rows(&stack), 0, "redb is emptied by the migration fold");
assert_eq!(stack.len(), 70);
assert!(stack.projection_is_complete());
for e in &all {
assert!(stack.lookup(&e.oid).is_some(), "migrated row {} lost", hex::encode(&e.oid));
}
drop(stack);
let again = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(again.stats().persisted_loads, 1, "the second open is warm");
assert_eq!(again.stats().folds, 0);
assert_eq!(again.len(), 70);
}
// ── the redb 2 tail, migrated (2026-10-06) ───────────────────────────────
/// Both tables of a tail, as bytes: `(objects, meta)`.
type TailTables = (Vec<(Vec<u8>, Vec<u8>)>, Vec<(String, u64)>);
fn read_tail(db: &Database) -> TailTables {
let r = db.begin_read().unwrap();
let objects = r
.open_table(OBJECTS)
.unwrap()
.iter()
.unwrap()
.map(|kv| {
let (k, v) = kv.unwrap();
(k.value().to_vec(), v.value().to_vec())
})
.collect();
let meta = r
.open_table(META)
.unwrap()
.iter()
.unwrap()
.map(|kv| {
let (k, v) = kv.unwrap();
(k.value().to_owned(), v.value())
})
.collect();
(objects, meta)
}
/// Rewrite the redb 4 tail at `path` as the redb 2 file every store made
/// before 2026-10-06 has: same tables, same rows, redb 2.6's own writer.
fn rewrite_as_redb2(path: &Path) -> TailTables {
let tables = read_tail(&Database::open(path).unwrap());
fs::remove_file(path).unwrap();
let db = redb2::Database::create(path).unwrap();
let w = db.begin_write().unwrap();
{
let objects_def: redb2::TableDefinition<&[u8], &[u8]> =
redb2::TableDefinition::new("objects");
let meta_def: redb2::TableDefinition<&str, u64> = redb2::TableDefinition::new("meta");
let mut objects = w.open_table(objects_def).unwrap();
for (k, v) in &tables.0 {
objects.insert(k.as_slice(), v.as_slice()).unwrap();
}
let mut meta = w.open_table(meta_def).unwrap();
for (k, v) in &tables.1 {
meta.insert(k.as_str(), *v).unwrap();
}
}
w.commit().unwrap();
tables
}
fn allocated(path: &Path) -> u64 {
file_sizes(path).1
}
/// **An old store opens, answers every object it answered, and gives the
/// disk back.** The tail is in the shape a busy store leaves it: 40 rows
/// folded into `objects.index`, 50 that arrived after the fold still in the
/// tail — written by redb 2, as every tail before 2026-10-06 was.
#[test]
fn a_redb2_tail_is_migrated_once_and_answers_the_same() {
let dir = tempfile::tempdir().unwrap();
let tail = dir.path().join("objects.tail");
let all = synthetic_entries(91, 20, 41);
{
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
stack.append(&all[..40]).unwrap();
stack.rebuild().unwrap();
stack.append(&all[40..90]).unwrap();
}
let before = rewrite_as_redb2(&tail);
assert_eq!(before.0.len(), 50, "fixture: the 50 rows after the fold are in the tail");
let old_alloc = allocated(&tail);
// Opened and closed once, as a running server does to every store.
drop(redb2::Database::create(&tail).unwrap());
let old_closed = allocated(&tail);
assert!(
old_closed > 1_000_000,
"fixture: a closed redb 2 tail allocates {old_closed} B (create: {old_alloc} B) — \
the cost this migration exists to remove"
);
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.expect("an old store must open");
assert!(!migrating_path(&tail).exists(), "the staging file is renamed, not left");
assert_eq!(read_tail(&stack.tail), before, "every row and every counter carried over");
assert_eq!(stack.len(), 90);
assert_eq!(stack.stats().persisted_loads, 1, "the folded index is still mapped");
assert_eq!(stack.stats().unabsorbed_rows, 50);
for e in &all[..90] {
let row = stack
.lookup(&e.oid)
.unwrap_or_else(|| panic!("the migration lost {}", hex::encode(&e.oid)));
assert_eq!(row.offset, e.offset);
assert_eq!(row.delta_base, e.delta_base);
}
// It keeps working: a new row lands after the old ones.
stack.append(&all[90..]).unwrap();
assert_eq!(stack.len(), 91);
drop(stack);
let new_closed = allocated(&tail);
assert!(
new_closed <= 100 * 1024,
"the migrated tail allocates {new_closed} B (redb 2: {old_closed} B)"
);
// Once: the second open is a plain redb 4 open.
let again = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(again.len(), 91);
assert!(again.lookup(&all[90].oid).is_some());
}
/// A migration that died before its rename left a half-built redb 4 file
/// beside the untouched redb 2 tail. The next open throws the half away and
/// migrates again; nothing is read from it.
#[test]
fn an_interrupted_migration_is_started_again_from_the_redb2_tail() {
let dir = tempfile::tempdir().unwrap();
let tail = dir.path().join("objects.tail");
let all = synthetic_entries(30, 32, 7);
{
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
stack.append(&all).unwrap();
}
let before = rewrite_as_redb2(&tail);
fs::write(migrating_path(&tail), b"half a redb 4 file, cut by a power loss").unwrap();
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.expect("the open must migrate over the leftover");
assert!(!migrating_path(&tail).exists());
assert_eq!(stack.len(), 30);
for e in &all {
assert!(stack.lookup(&e.oid).is_some(), "lost {}", hex::encode(&e.oid));
}
let _ = before;
}
/// **A brand-new tail is small.** The per-repository floor this crate puts
/// on a disk, created and closed, then after one small append.
#[test]
fn a_new_tail_allocates_kilobytes_not_megabytes() {
let dir = tempfile::tempdir().unwrap();
let tail = dir.path().join("objects.tail");
let open = || {
ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap()
};
drop(open());
let empty = allocated(&tail);
let stack = open();
stack.append(&synthetic_entries(3, 20, 3)).unwrap();
drop(stack);
drop(open());
let small = allocated(&tail);
assert!(empty <= 64 * 1024, "an empty tail allocates {empty} B");
assert!(small <= 64 * 1024, "a three-object tail allocates {small} B");
}
/// **A migrated store gets its redb file back.** A pre-fold store big enough
/// to matter (here 200 k rows, ~34 MB of redb) is opened once: the migration
/// fold empties redb, and the open compacts the file because it now carries
/// more than [`COMPACT_FRAGMENTED_FLOOR`] of dead pages. Asserted on the
/// file's size on disk, not on a flag. A small store (the 70-row fixture next
/// door) is deliberately left alone — that case is covered by the floor, and
/// this test's precondition asserts the fixture is above it.
///
/// Seen RED by making `compact_if_fragmented` return before `db.compact()`:
/// "the migrated redb must shrink" with equal before/after sizes. Restored.
#[test]
fn a_migrated_store_gets_its_redb_file_back() {
let dir = tempfile::tempdir().unwrap();
let tail = dir.path().join("objects.tail");
let n = 200_000usize;
let all = synthetic_entries(n, 20, 53);
{
let db = Database::create(&tail).unwrap();
let w = db.begin_write().unwrap();
{
let mut objects = w.open_table(OBJECTS).unwrap();
for (i, e) in all.iter().enumerate() {
objects.insert(e.oid.as_slice(), encode_row(i as u64, e).as_slice()).unwrap();
}
let mut meta = w.open_table(META).unwrap();
meta.insert(META_TAIL_FORMAT, TAIL_FORMAT).unwrap();
meta.insert(META_ARRIVAL_SEQ, n as u64).unwrap();
}
w.commit().unwrap();
}
let before = fs::metadata(&tail).unwrap().len();
assert!(
before >= COMPACT_FRAGMENTED_FLOOR,
"precondition: the fixture must be above the compaction floor, is {before} B"
);
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&tail,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(stack.stats().folds, 1, "the migration fold ran");
assert_eq!(redb_rows(&stack), 0);
let after = fs::metadata(&tail).unwrap().len();
assert!(
after < before / 4,
"the migrated redb must shrink: {after} >= {before} / 4"
);
// And it still works, in both directions.
assert!(stack.lookup(&all[123].oid).is_some());
stack.append(&synthetic_entries(10, 20, 54)).unwrap();
assert_eq!(redb_rows(&stack), 10);
}
/// The packed tail row rejects the two things that would make it silently
/// wrong: a short buffer and a type code git does not use.
///
/// Seen RED by dropping the length check from `decode_row`: the short-buffer
/// case panicked — "range end index 33 out of range for slice of length 32"
/// — instead of returning an error. Restored.
///
/// Seen RED for the `delta_base` field by decoding it from byte 25 (where
/// `uncompressed_size` starts) instead of byte 33: "the delta base did not
/// survive the tail row / left: 2448 / right: 4242" — it came back holding
/// the *size*. Both are plausible u64s, so only a comparison against the
/// entry that was encoded can tell them apart. Restored.
///
/// A row written by the 33-byte encoding is a **length** error here rather
/// than a misread: `decode_row` refuses anything that is not exactly
/// `TAIL_ROW_BYTES`, and the assertion below is the same check the old width
/// now trips.
#[test]
fn a_packed_tail_row_round_trips_and_refuses_nonsense() {
let mut e = synthetic_entries(1, 20, 51)[0].clone();
// A one-entry workload has nothing to delta against, so the base is set
// here: a round trip that only ever saw 0 would not test the field.
e.delta_base = 4242;
let e = &e;
let packed = encode_row(9, e);
let back = decode_row(&packed).unwrap();
assert_eq!(back.seq, 9);
assert_eq!(back.offset, e.offset);
assert_eq!(back.len, e.len);
assert_eq!(back.obj_type, e.obj_type);
assert_eq!(back.uncompressed_size, e.uncompressed_size);
assert_eq!(
back.delta_base, e.delta_base,
"the delta base did not survive the tail row"
);
assert_ne!(
back.delta_base, back.uncompressed_size,
"the fixture must not let the two u64 fields alias"
);
assert!(decode_row(&packed[..TAIL_ROW_BYTES - 1]).is_err());
// 33 bytes is exactly the pre-`delta_base` row width, and it is refused
// rather than read as a row with the field missing.
assert!(
decode_row(&packed[..33]).is_err(),
"a row in the old 33-byte encoding must be refused, not reinterpreted"
);
let mut bad = packed;
bad[24] = 5; // git uses 1-4, 6, 7; 5 is unassigned
assert!(
decode_row(&bad).is_err(),
"an unassigned object type code was accepted"
);
}
/// The stack is the same index as the two Arrow arms once it is fully
/// absorbed, which is what lets it be dropped into
/// `examples/index_layout_bench.rs` unchanged.
///
/// Seen RED by having `ObjectIndex::build` skip its final `rebuild()`, so
/// every row was served from the tail instead of from the projection:
/// "assertion `left == right` failed: build left rows in the tail / left: 0
/// / right: 400". Restored.
#[test]
fn a_fully_absorbed_stack_agrees_with_both_arrow_arms() {
let entries = synthetic_entries(400, 20, 61);
let absent = synthetic_entries(100, 20, 62);
let stack = Stack::build(&entries).unwrap();
let a = FourTables::build(&entries).unwrap();
let b = OneTableFourColumns::build(&entries).unwrap();
assert_eq!(stack.len(), 400);
assert_eq!(stack.projection_len(), 400, "build left rows in the tail");
let mut refs: Vec<&[u8]> = entries.iter().map(|e| e.oid.as_slice()).collect();
refs.extend(absent.iter().map(|e| e.oid.as_slice()));
let rs = stack.lookup_batch(&refs);
assert_eq!(
rs,
a.lookup_batch(&refs),
"the stack disagrees with FourTables"
);
assert_eq!(
rs,
b.lookup_batch(&refs),
"the stack disagrees with OneTableFourColumns"
);
assert_eq!(
stack.stats().tail_hits,
0,
"a fully absorbed stack still went to the tail for a row"
);
assert_eq!(stack.stats().absent, 100);
}
/// **A complete projection answers an absence without opening a redb
/// transaction at all** — and the moment one row is appended, it stops doing
/// that.
///
/// This is the fast path the bench found was worth having: a `have`
/// negotiation is mostly misses, and paying a redb round trip for each of
/// them made the stack six times the cost of the bare Arrow arm on the most
/// common git operation. `tail_txns` is the applied output — an assertion on
/// timing would be a flake, and an assertion on the returned rows cannot
/// distinguish the two paths, because both return the same rows. That is the
/// point: the fast path is only allowed to exist because it is
/// indistinguishable in its answers.
///
/// Seen RED by deleting the `projection_is_complete()` branch from
/// `lookup_batch`: "assertion `left == right` failed: a complete projection
/// opened 1 redb transactions to say 'no' / left: 1 / right: 0". Restored.
///
/// Seen RED the other way — the dangerous way — by making
/// `projection_is_complete` return `true` unconditionally: the
/// `the_projection_is_incomplete_but_never_wrong` test failed with "the
/// stack lost tail object …, batch path", because an incomplete projection
/// then declared its misses absent. Restored.
#[test]
fn a_complete_projection_answers_absent_without_asking_the_tail() {
let entries = synthetic_entries(200, 20, 91);
let nowhere = synthetic_entries(500, 20, 92);
let stack = Stack::build(&entries).unwrap();
assert!(stack.projection_is_complete());
let refs: Vec<&[u8]> = nowhere.iter().map(|e| e.oid.as_slice()).collect();
assert!(stack.lookup_batch(&refs).iter().all(Option::is_none));
assert!(stack.lookup(&nowhere[0].oid).is_none());
assert!(stack.ordinals_batch(&refs).iter().all(Option::is_none));
assert!(stack.extents_batch(&refs).iter().all(Option::is_none));
let st = stack.stats();
assert_eq!(st.unabsorbed_rows, 0);
assert_eq!(
st.tail_txns, 0,
"a complete projection opened {} redb transactions to say 'no'",
st.tail_txns
);
assert_eq!(st.absent, 500 + 1 + 500 + 500, "absences went uncounted");
// One appended row and the fall-through is on again — the fast path is
// conditional on completeness and not on having been complete once.
let more = synthetic_entries(1, 20, 93);
stack.append(&more).unwrap();
assert!(!stack.projection_is_complete());
assert!(stack.lookup_batch(&refs).iter().all(Option::is_none));
assert!(
stack.stats().tail_txns > 0,
"an incomplete projection did not consult the tail"
);
assert!(
stack.lookup(&more[0].oid).is_some(),
"the appended row is not reachable"
);
}
/// **A tail written in a different layout is DROPPED and rebuilt, not
/// refused for ever.**
///
/// `decode_row` refuses an unknown row width on purpose — reinterpreting one
/// puts a wrong offset into an index that is then queried and believed. But
/// until `META_TAIL_FORMAT` existed the refusal had nothing behind it: every
/// read `bail`ed and the only way back was deleting the repository.
///
/// The tail is a cache, so the way back is to empty it and let
/// `Absorber::adopt_journal` requeue every pack — the same path a crash
/// mid-absorb already takes.
///
/// RED-FIRST: with the version check removed, the reopened stack keeps the
/// 90 foreign rows and the first `get` through them fails on the row width
/// instead. The assertion is on the APPLIED state — the row count and a
/// working `append`/`get` afterwards — not on a returned flag.
#[test]
fn a_tail_in_a_foreign_layout_is_dropped_and_the_store_keeps_working() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("objects.tail.redb");
let all = synthetic_entries(90, 20, 41);
{
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&path,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
stack.append(&all).unwrap();
assert_eq!(stack.tail_epoch_and_len().unwrap().1, 90);
}
// Stamp a version this build does not know, exactly as a future layout
// change would leave behind.
{
let db = Database::create(&path).unwrap();
let w = db.begin_write().unwrap();
{
let mut meta = w.open_table(META).unwrap();
meta.insert(META_TAIL_FORMAT, TAIL_FORMAT + 7).unwrap();
}
w.commit().unwrap();
}
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&path,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(
stack.tail_epoch_and_len().unwrap().1,
0,
"a foreign-layout tail was kept; the next read would bail on the row width"
);
assert_eq!(stack.projection_len(), 0);
// And the store is USABLE, which is the half that says "dropped" and not
// "broken": the same oids re-appended come back.
stack.append(&all).unwrap();
assert_eq!(stack.tail_epoch_and_len().unwrap().1, 90);
let got = stack.lookup(&all[7].oid).expect("re-appended row");
assert_eq!(got.offset, all[7].offset);
// The stamp is now this build's, so a second reopen drops nothing.
drop(stack);
let again = ObjectReadStack::<OneTableFourColumns>::open(
&path,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(
again.tail_epoch_and_len().unwrap().1,
90,
"the rewritten version did not stick, so every open would drop the tail"
);
}
/// A tail written BEFORE the version key existed carries no stamp, and must
/// be adopted rather than dropped: it holds rows of exactly this layout, and
/// rebuilding every deployed store for a key that was merely absent would be
/// a self-inflicted re-index of the whole fleet.
#[test]
fn a_tail_with_no_version_stamp_is_adopted_and_stamped_not_dropped() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("objects.tail.redb");
let all = synthetic_entries(50, 20, 41);
{
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&path,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
stack.append(&all).unwrap();
}
// Remove the stamp: this is what every tail on disk looked like before
// the key existed.
{
let db = Database::create(&path).unwrap();
let w = db.begin_write().unwrap();
{
let mut meta = w.open_table(META).unwrap();
meta.remove(META_TAIL_FORMAT).unwrap();
}
w.commit().unwrap();
}
let stack = ObjectReadStack::<OneTableFourColumns>::open(
&path,
RebuildTriggers::manual(),
DEFAULT_REDB_CACHE_BYTES,
)
.unwrap();
assert_eq!(stack.tail_epoch_and_len().unwrap().1, 50, "an unstamped tail was dropped");
let r = stack.tail.begin_read().unwrap();
let meta = r.open_table(META).unwrap();
assert_eq!(
meta.get(META_TAIL_FORMAT).unwrap().map(|v| v.value()),
Some(TAIL_FORMAT),
"the tail was adopted but not stamped, so it is re-examined for ever"
);
}
}