znippy-common/src/meta_sink_append.rs
//! `ArrowIpcSinkAppend` — the **append/resume-capable v2 clone** of
//! [`ArrowIpcSink`](crate::ArrowIpcSink).
//!
//! ## Why a clone and not a refactor
//! The original [`ArrowIpcSink`] is the A baseline of the arrow-ipc write/seal
//! A/B audit — its hot path (`push_subindex` + `finish`) must stay byte- and
//! perf-identical. So this is a *near-identical copy* that adds the two new
//! capabilities **without touching the original**:
//!
//! * **fresh path** — [`ArrowIpcSinkAppend::new`] + `push_subindex` + `finish`
//! is a line-for-line clone of `ArrowIpcSink`. It writes the **same v0.7
//! on-disk bytes** (same sub-index serialisation, same sorted lookup, same
//! fst trie, same manifest, same `ZNPYMIDX` footer). The A/B parity test
//! proves this clone is a zero-cost superset on the normal write path.
//!
//! * **resume/append path** — [`ArrowIpcSinkAppend::open_existing`] reopens an
//! already-sealed `.znippy`, recovers its existing data rows, repositions the
//! cursor at the **end of the blob region** (truncating the old metadata
//! tail), and lets the caller `push_subindex` more blobs' rows. `finish()`
//! then re-seals the merged old+new row set — the **first-class native blob
//! append** that the iceberg lifecycle test (#23) previously had to do by
//! hand.
//!
//! ### The resume mechanism, in bytes
//! A sealed v0.7 archive is:
//! ```text
//! [ blob_0 … blob_N ][ data sub-idx(es) ][ lookup sub-idx ][ trie ][ manifest ][ ZNPYMIDX ][off]
//! ^0 ^blob_end (the whole metadata tail is rebuildable)
//! ```
//! To append we must:
//! 1. read the full manifest (incl. reserved lookup/trie entries),
//! 2. find `blob_end` = the lowest `index_offset` over all sub-index entries
//! (where the blob region stops and the rebuildable tail begins),
//! 3. recover the existing **data** rows from the sorted lookup sub-index (one
//! cheap read of the already-sorted reserved section — no per-sub-index
//! re-scan), seeding the lookup accumulator,
//! 4. set the cursor to `blob_end` so the caller's new blob bytes + new data
//! sub-index overwrite the old (now-stale) metadata tail,
//! 5. on `finish()`, re-sort the merged rows and re-emit lookup + trie +
//! manifest + footer.
//!
//! The new blob bytes are written by the caller (the compress pipeline / a
//! library append entrypoint such as [`append_files`]) to the file at
//! `blob_end()` exactly as a fresh archive
//! writes blobs at offset 0; this sink owns only the metadata tail, identical to
//! the original's contract.
use std::fs::File;
use std::os::unix::fs::FileExt;
use std::path::Path;
use std::sync::Arc;
use anyhow::{Result, anyhow};
use arrow::array::{
BooleanArray, BooleanBuilder, FixedSizeBinaryArray, FixedSizeBinaryBuilder, StringArray,
StringBuilder, UInt32Array, UInt32Builder, UInt64Array, UInt64Builder,
};
use arrow::datatypes::Schema;
use arrow::ipc::writer::StreamWriter;
use arrow::record_batch::RecordBatch;
use crate::index::{
ChunkLoc, LOOKUP_MODULE, MULTI_INDEX_MAGIC, ManifestEntry, RESERVED_PKG_TYPE, TRIE_MODULE,
data_subindex_schema, is_reserved_module, lookup_schema, read_znippy_full_manifest,
write_manifest_bytes,
};
use crate::index::{
CARRIED_RESERVED_MODULES, META_MODULE, ZNIPPY_DELTA_MODULE, read_reserved_section_bytes,
};
use crate::meta_index::{
MetaTable, build_meta_batch, decode_meta_section, meta_schema,
};
use crate::meta_sink::{ArchiveMetaSink, GroupKey};
/// Append/resume-capable v2 clone of [`ArrowIpcSink`](crate::ArrowIpcSink).
///
/// Field-for-field identical to the original; the only added surface is the
/// [`open_existing`](Self::open_existing) constructor and the [`blob_end`](Self::blob_end)
/// accessor. The fresh-write path is byte-identical to the original.
pub struct ArrowIpcSinkAppend {
file: Arc<File>,
cursor: u64,
entries: Vec<ManifestEntry>,
lookup_paths: Vec<String>,
lookup_locs: Vec<ChunkLoc>,
/// On resume, the recovered pre-existing data rows. They are re-emitted as a
/// data sub-index in `finish()` (so the ordinary reader, which reads data
/// sub-indexes, still sees them) AND merged into the rebuilt lookup. Empty
/// for a fresh sink. Kept separate from `lookup_*` so they aren't
/// double-counted before the re-emit.
carried: Vec<(String, ChunkLoc)>,
/// Searchable metadata to seal as the `META_MODULE` sub-index.
///
/// `None` means "emit no section", which is what a reader later reports as
/// `ArchiveMeta::NoMetadata`; `Some(empty)` means "emit a present, empty
/// index". The two are different archives on disk and different answers on
/// read, and this field is the only place that decision is made.
///
/// On resume this is **seeded from the archive's existing section**, because
/// `open_existing` truncates the whole metadata tail — without carrying it,
/// every append would silently erase the metadata the archive already had.
meta: Option<MetaTable>,
/// Reserved sections carried forward verbatim across an append, as
/// `(module_name, raw bytes)`.
///
/// The same failure `meta` above fixes, for the reserved sections that are
/// **independent logs** rather than derivations — see
/// [`CARRIED_RESERVED_MODULES`]. MEASURED 2026-08-04: an object-carrying
/// push dropped `__gunnar_refs__` (144 928 -> 146 396 bytes, section gone),
/// because this sink re-supplies no reserved section it was not handed and
/// nothing hands it one.
///
/// Derived sections are deliberately NOT carried: when the objects change,
/// `__gunnar_graph__` / `__gunnar_reach__` / `__gunnar_oid__` are wrong, and
/// carrying a stale one forward is worse than dropping it.
carried_reserved: Vec<(String, Vec<u8>)>,
/// The delta map, as rows: `(relative_path, chunk_seq, base_path)`.
///
/// Decoded rather than carried raw, because an append may ADD rows to it and
/// the manifest cannot hold two sections under one module name. Same reason
/// `meta` travels as a `MetaTable`.
delta_map: Vec<(String, u32, String)>,
}
impl ArrowIpcSinkAppend {
/// Fresh archive: identical to [`ArrowIpcSink::new`](crate::ArrowIpcSink::new).
/// `blob_end_offset` is the byte offset just past the last blob.
pub fn new(file: Arc<File>, blob_end_offset: u64) -> Self {
Self {
file,
cursor: blob_end_offset,
entries: Vec::new(),
lookup_paths: Vec::new(),
lookup_locs: Vec::new(),
carried: Vec::new(),
meta: None,
carried_reserved: Vec::new(),
delta_map: Vec::new(),
}
}
/// Seal `meta` as this archive's searchable metadata sub-index, replacing
/// anything carried from a resumed archive.
pub fn with_meta(mut self, meta: MetaTable) -> Self {
self.meta = Some(meta);
self
}
/// Add rows to the metadata to be sealed, keeping whatever a resume carried.
/// Creates the section if the archive had none.
pub fn merge_meta(&mut self, rows: impl IntoIterator<Item = crate::meta_index::MetaEntry>) {
self.meta.get_or_insert_with(MetaTable::new).extend(rows);
}
/// The metadata this sink will seal — `None` when it will emit no section.
pub fn meta(&self) -> Option<&MetaTable> {
self.meta.as_ref()
}
/// Reopen an **already-sealed** v0.7 `.znippy` for append/resume.
///
/// Recovers the existing data rows from the sorted lookup sub-index, drops
/// the (rebuildable) metadata tail by positioning the cursor at the end of
/// the blob region, and returns a sink ready to accept more `push_subindex`
/// calls. The caller appends its new blob bytes to the same file starting at
/// [`blob_end`](Self::blob_end) before pushing the matching index rows.
///
/// The file is opened read+write; nothing is mutated until `push_subindex` /
/// `finish` overwrite the old tail.
pub fn open_existing(path: &Path) -> Result<Self> {
let file = Arc::new(
std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(path)
.map_err(|e| anyhow!("append: open {} for resume: {e}", path.display()))?,
);
// 1. Full manifest (incl. reserved lookup/trie entries).
let (entries, _manifest_offset) = read_znippy_full_manifest(path)?;
if entries.is_empty() {
return Err(anyhow!("append: archive {} has an empty manifest", path.display()));
}
// 2. blob_end = lowest index_offset over all sub-index/reserved sections.
// Everything from there to EOF is the rebuildable metadata tail.
let blob_end = entries
.iter()
.map(|e| e.index_offset)
.min()
.ok_or_else(|| anyhow!("append: no sections in manifest"))?;
// 3. Recover existing DATA rows from the sorted lookup sub-index (one
// read of the already-sorted reserved section). Fall back to scanning
// the data sub-indexes if (unexpectedly) no lookup section is present.
let (paths, locs) = recover_rows(path, &entries)?;
let carried: Vec<(String, ChunkLoc)> = paths.into_iter().zip(locs).collect();
// 4. Carry the searchable metadata section forward. The tail we are about
// to overwrite contains it, so a resume that did not recover it would
// quietly turn an archive WITH metadata into one without — and the
// reader would then honestly report `NoMetadata` about an archive that
// used to have some. Absent stays absent; present-but-empty stays
// present-but-empty.
let meta = match read_reserved_section_bytes(path, META_MODULE)? {
None => None,
Some(bytes) => Some(decode_meta_section(&bytes)?.to_table()),
};
// 5. Carry the INDEPENDENT reserved sections forward verbatim. The tail
// about to be overwritten holds them, and nothing else in this process
// can reproduce them: they are per-push logs, not derivations of the
// blobs. Derived sections are left to be rebuilt.
let mut carried_reserved = Vec::new();
for module in CARRIED_RESERVED_MODULES {
if let Some(bytes) = read_reserved_section_bytes(path, module)? {
carried_reserved.push(((*module).to_string(), bytes));
}
}
// 6. The delta map, decoded. Dropping it would turn every delta chunk in
// the archive back into a stored chunk on the next append, and the
// reader would then hand a delta's instruction stream to a caller as
// file content.
let delta_map = read_delta_map(path)?;
Ok(Self {
file,
cursor: blob_end,
entries: Vec::new(), // rebuilt fresh by push_subindex + finish
lookup_paths: Vec::new(),
lookup_locs: Vec::new(),
carried,
meta,
carried_reserved,
delta_map,
})
}
/// Byte offset where the blob region ends in a resumed archive — where the
/// caller writes its newly-appended blob bytes (and where the first new data
/// sub-index will be placed). For a fresh sink this is the `blob_end_offset`
/// passed to [`new`](Self::new) until the first `push_subindex`.
pub fn blob_end(&self) -> u64 {
self.cursor
}
/// Number of pre-existing data rows recovered on resume (0 for a fresh sink).
pub fn recovered_rows(&self) -> usize {
self.carried.len()
}
/// Drop every carried (pre-existing) row whose `relative_path` is about to be
/// re-written by this append, giving last-writer-wins **replace** semantics.
/// Returns the number of rows dropped.
///
/// Without this, appending a path the archive already contains left TWO row
/// sets for it in the re-sealed index — the stale one and the new one — and
/// nothing downstream treated that as an error: the reader that concatenates
/// chunks returned both copies back to back at twice the real length, and the
/// reader that places chunks at `fdata_offset` wrote both to offset 0, so the
/// carried STALE copy (re-emitted last, in `finish`) won and `znippy get`
/// silently handed back the old file. Every chunk's blake3 is individually
/// correct in both cases, so even verified reads passed.
fn drop_carried_paths(&mut self, replacing: &std::collections::HashSet<&str>) -> usize {
if self.carried.is_empty() || replacing.is_empty() {
return 0;
}
let before = self.carried.len();
self.carried.retain(|(p, _)| !replacing.contains(p.as_str()));
before - self.carried.len()
}
/// Re-emit the carried (recovered) rows as a single data sub-index so the
/// ordinary reader — which reads data sub-indexes, not the lookup — still
/// lists them after the re-seal. `push_subindex` also folds them into the
/// rebuilt lookup accumulator. No-op for a fresh sink.
fn emit_carried(&mut self) -> Result<()> {
if self.carried.is_empty() {
return Ok(());
}
let carried = std::mem::take(&mut self.carried);
let (paths, locs): (Vec<String>, Vec<ChunkLoc>) = carried.into_iter().unzip();
let batch = base_batch_from_rows(&paths, &locs)?;
self.push_subindex(data_subindex_schema().as_ref(), &[batch], GroupKey {
pkg_type: 0,
repo: String::new(),
module_name: String::new(),
})
}
// ── below: a line-for-line clone of ArrowIpcSink's private machinery ──
fn accumulate_lookup(&mut self, batch: &RecordBatch) {
let cols = (|| {
Some((
batch.column_by_name("relative_path")?.as_any().downcast_ref::<StringArray>()?,
batch.column_by_name("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()?,
batch.column_by_name("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()?,
batch.column_by_name("compressed")?.as_any().downcast_ref::<BooleanArray>()?,
batch.column_by_name("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()?,
batch.column_by_name("blob_offset")?.as_any().downcast_ref::<UInt64Array>()?,
batch.column_by_name("blob_size")?.as_any().downcast_ref::<UInt64Array>()?,
batch.column_by_name("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()?,
))
})();
let Some((paths, chunk_seq, fdata, compressed, usz, blob_off, blob_sz, checksum)) = cols
else { return; };
for i in 0..batch.num_rows() {
let mut ck = [0u8; 32];
ck.copy_from_slice(checksum.value(i));
self.lookup_paths.push(paths.value(i).to_string());
self.lookup_locs.push(ChunkLoc {
chunk_seq: chunk_seq.value(i),
fdata_offset: fdata.value(i),
blob_offset: blob_off.value(i),
blob_size: blob_sz.value(i),
uncompressed_size: usz.value(i),
compressed: compressed.value(i),
checksum: ck,
});
}
}
fn write_lookup_and_trie(&mut self) -> Result<()> {
let n = self.lookup_paths.len();
let mut order: Vec<usize> = (0..n).collect();
order.sort_by(|&a, &b| {
self.lookup_paths[a].cmp(&self.lookup_paths[b])
.then(self.lookup_locs[a].chunk_seq.cmp(&self.lookup_locs[b].chunk_seq))
});
let schema = lookup_schema();
let batch = base_batch_permuted(
schema.clone(),
&self.lookup_paths,
&self.lookup_locs,
&order,
)?;
self.push_subindex(&schema, &[batch], GroupKey {
pkg_type: RESERVED_PKG_TYPE,
repo: String::new(),
module_name: LOOKUP_MODULE.to_string(),
})?;
let mut builder = fst::MapBuilder::memory();
let mut prev: Option<&str> = None;
for (sorted_idx, &orig) in order.iter().enumerate() {
let p = self.lookup_paths[orig].as_str();
if prev != Some(p) {
builder.insert(p.as_bytes(), sorted_idx as u64)
.map_err(|e| anyhow!("trie insert: {e}"))?;
prev = Some(p);
}
}
let trie_bytes = builder.into_inner().map_err(|e| anyhow!("trie finish: {e}"))?;
self.write_raw_section(&trie_bytes, GroupKey {
pkg_type: RESERVED_PKG_TYPE,
repo: String::new(),
module_name: TRIE_MODULE.to_string(),
})
}
/// Emit the searchable metadata sub-index, when there is one to emit.
///
/// Reserved module, so the data readers skip it and an older znippy simply
/// ignores the entry. `None` writes NOTHING — that absence is exactly what
/// `ArchiveMeta::NoMetadata` reports, and it is why an archive sealed without
/// metadata stays byte-identical to one sealed before this module existed.
fn write_meta_subindex(&mut self) -> Result<()> {
let Some(table) = self.meta.take() else {
return Ok(());
};
let batch = build_meta_batch(&table)?;
let schema = meta_schema();
self.push_subindex(schema.as_ref(), &[batch], GroupKey {
pkg_type: RESERVED_PKG_TYPE,
repo: String::new(),
module_name: META_MODULE.to_string(),
})
}
/// Re-emit the reserved sections a resume carried, byte for byte.
///
/// After `write_meta_subindex`, so a `__meta__` section this sink was handed
/// wins over anything else; the carried list never contains `__meta__`,
/// which travels as a decoded `MetaTable` instead.
fn write_carried_reserved(&mut self) -> Result<()> {
for (module, bytes) in std::mem::take(&mut self.carried_reserved) {
self.write_raw_section(&bytes, GroupKey {
pkg_type: RESERVED_PKG_TYPE,
repo: String::new(),
module_name: module,
})?;
}
Ok(())
}
/// Seal the delta map, if there is one.
fn write_delta_map(&mut self) -> Result<()> {
if self.delta_map.is_empty() {
return Ok(());
}
let rows = std::mem::take(&mut self.delta_map);
let paths = StringArray::from(rows.iter().map(|r| r.0.as_str()).collect::<Vec<_>>());
let seqs = UInt32Array::from(rows.iter().map(|r| r.1).collect::<Vec<_>>());
let bases = StringArray::from(rows.iter().map(|r| r.2.as_str()).collect::<Vec<_>>());
let schema = crate::index::delta_map_schema();
let batch = RecordBatch::try_new(
Arc::clone(&schema),
vec![Arc::new(paths), Arc::new(seqs), Arc::new(bases)],
)
.map_err(|e| anyhow!("delta map batch: {e}"))?;
self.push_subindex(schema.as_ref(), &[batch], GroupKey {
pkg_type: RESERVED_PKG_TYPE,
repo: String::new(),
module_name: ZNIPPY_DELTA_MODULE.to_string(),
})
}
/// The archive file, for a caller appending blob bytes at [`blob_end`](Self::blob_end).
pub fn file(&self) -> &Arc<File> {
&self.file
}
/// Move the blob cursor on after a caller wrote `n` bytes at `blob_end`.
pub fn advance_blob_end(&mut self, n: u64) {
self.cursor += n;
}
/// Replace every carried row of `path` with `locs`.
///
/// Used to turn a stored entry into a delta entry in place: its old chunk
/// rows go, one delta row arrives. Rows of other paths are untouched.
pub fn replace_carried(&mut self, path: &str, locs: Vec<ChunkLoc>) {
self.carried.retain(|(p, _)| p != path);
for loc in locs {
self.carried.push((path.to_string(), loc));
}
}
/// Record that `(path, chunk_seq)` is a delta against `base`.
pub fn push_delta_map_row(&mut self, path: String, chunk_seq: u32, base: String) {
self.delta_map.retain(|(p, s, _)| !(p == &path && *s == chunk_seq));
self.delta_map.push((path, chunk_seq, base));
}
fn write_raw_section(&mut self, bytes: &[u8], key: GroupKey) -> Result<()> {
let start = self.cursor;
self.file.write_all_at(bytes, start)?;
self.cursor += bytes.len() as u64;
self.entries.push(ManifestEntry {
pkg_type: key.pkg_type,
repo: key.repo,
module_name: key.module_name,
index_offset: start,
index_len: bytes.len() as u64,
row_count: 0,
});
Ok(())
}
}
impl ArchiveMetaSink for ArrowIpcSinkAppend {
fn push_subindex(
&mut self,
schema: &Schema,
batches: &[RecordBatch],
key: GroupKey,
) -> Result<()> {
let sub_start = self.cursor;
let mut sub_bytes: Vec<u8> = Vec::new();
let mut sw = StreamWriter::try_new(&mut sub_bytes, schema)
.map_err(|e| anyhow!("sub-index writer: {e}"))?;
let mut row_count = 0u64;
for batch in batches {
row_count += batch.num_rows() as u64;
sw.write(batch).map_err(|e| anyhow!("sub-index write: {e}"))?;
}
sw.finish().map_err(|e| anyhow!("sub-index finish: {e}"))?;
// Accumulate base columns for the lookup layer from DATA sub-indexes only.
// Widened from "not lookup, not trie" to "not reserved" when META_MODULE
// arrived: the metadata sub-index is Arrow IPC and does come through here,
// and its rows are key/value facts, not chunk locations — folding them
// into the lookup would corrupt random access. Behaviour for every
// pre-existing module is unchanged (sign sections are raw, never pushed).
if !is_reserved_module(&key.module_name) {
for batch in batches {
self.accumulate_lookup(batch);
}
}
let sub_len = sub_bytes.len() as u64;
self.file.write_all_at(&sub_bytes, sub_start)?;
self.cursor += sub_len;
self.entries.push(ManifestEntry {
pkg_type: key.pkg_type,
repo: key.repo,
module_name: key.module_name,
index_offset: sub_start,
index_len: sub_len,
row_count,
});
Ok(())
}
fn finish(mut self: Box<Self>) -> Result<u64> {
// Resume: re-emit recovered rows as a data sub-index (the ordinary reader
// reads data sub-indexes, not the lookup). No-op on the fresh path.
self.emit_carried()?;
self.write_lookup_and_trie()?;
self.write_meta_subindex()?;
self.write_carried_reserved()?;
self.write_delta_map()?;
let manifest_offset = self.cursor;
let manifest_bytes =
write_manifest_bytes(&self.entries).map_err(|e| anyhow!("manifest: {e}"))?;
self.file.write_all_at(&manifest_bytes, manifest_offset)?;
let after = manifest_offset + manifest_bytes.len() as u64;
self.file.write_all_at(&MULTI_INDEX_MAGIC, after)?;
self.file.write_all_at(
&manifest_offset.to_le_bytes(),
after + MULTI_INDEX_MAGIC.len() as u64,
)?;
// Resume overwrites the old (longer-or-shorter) tail in place; if the new
// tail is shorter than the old one, truncate so no stale footer lingers.
let final_len = after + MULTI_INDEX_MAGIC.len() as u64 + 8;
self.file.set_len(final_len)?;
self.file.sync_all()?;
Ok(final_len)
}
}
/// Read the delta map out of an archive, as rows. Empty when the section is
/// absent, which is every archive that holds no delta chunk.
///
/// **Public because the chain is the caller's to bound.**
/// [`supersede_as_delta`] takes `chain_depth` — how many links already sit
/// behind the base — and refuses past [`MAX_GENERATION_CHAIN`]. A caller that
/// supersedes one generation at a time (which is what a repository `gc` does)
/// cannot supply that number from what it did in this process: the chain was
/// built by earlier runs and the archive is the only record of it. Without this
/// the caller must keep a sidecar counter, and a counter that drifts low walks
/// the chain past the cap the writer exists to enforce.
pub fn read_delta_map(path: &Path) -> Result<Vec<(String, u32, String)>> {
let Some(bytes) = read_reserved_section_bytes(path, ZNIPPY_DELTA_MODULE)? else {
return Ok(Vec::new());
};
let reader = crate::ipc_guard::stream_reader(bytes)
.map_err(|e| anyhow!("delta map: {e}"))?;
let mut out = Vec::new();
for batch in reader {
let batch = batch.map_err(|e| anyhow!("delta map batch: {e}"))?;
let paths = batch
.column_by_name("relative_path")
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
.ok_or_else(|| anyhow!("delta map: missing relative_path"))?;
let seqs = batch
.column_by_name("chunk_seq")
.and_then(|c| c.as_any().downcast_ref::<UInt32Array>())
.ok_or_else(|| anyhow!("delta map: missing chunk_seq"))?;
let bases = batch
.column_by_name("base_path")
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
.ok_or_else(|| anyhow!("delta map: missing base_path"))?;
for r in 0..batch.num_rows() {
out.push((paths.value(r).to_string(), seqs.value(r), bases.value(r).to_string()));
}
}
Ok(out)
}
/// **The writer: re-store a superseded entry as a delta against a live one.**
///
/// Given an archive holding generation N and generation N+1, this replaces N's
/// stored bytes with a delta against N+1 and records the reference in
/// [`ZNIPPY_DELTA_MODULE`]. N's old blob bytes become dead space in the file;
/// they are reclaimed by a rewrite, not by this call.
///
/// # Why the OLD one is delta'd against the NEW one, and never the other way
///
/// The obvious direction is forward — store N+1 as a delta against N — and it is
/// wrong for gunnar, for a measured reason that is not znippy's. The current
/// generation is what `P-001`'s `write_pack_copy` copies entries out of on every
/// clone and fetch; putting it behind a delta chain would move it onto the
/// entry-copy path, measured at 18x the bytes and 221x the CPU of the path that
/// avoids it.
///
/// Delta-ing backwards keeps the property that matters: **the live generation is
/// always a whole entry, at chain depth zero**, whatever the archive's history.
/// The k-th-oldest generation sits at depth k, and depth there is free because
/// nothing on the serve path reads a superseded generation — the live pack holds
/// every reachable object.
///
/// # The chain bound
///
/// A repository `gc`'d weekly reaches [`MAX_RECONSTRUCT_DEPTH`] in fifteen
/// months, and the reader would then refuse the oldest generation rather than
/// serve it slowly. So the writer bounds itself first: past
/// [`MAX_GENERATION_CHAIN`] it declines and returns
/// [`SupersedeOutcome::ChainTooLong`], and the caller keeps that generation
/// whole. A whole generation is a keyframe: the chain restarts from it and every
/// generation is reachable in at most `MAX_GENERATION_CHAIN` links, for ever.
/// That is a policy decided here rather than a limit discovered in production.
///
/// `chain_depth` is how many links the base already sits behind a whole entry —
/// the caller knows this because it wrote them. Zero for a base that is whole.
pub fn supersede_as_delta(
archive: &Path,
superseded: &str,
base: &str,
chain_depth: usize,
compression_level: i32,
) -> Result<SupersedeOutcome> {
if superseded == base {
return Err(anyhow!("an entry cannot be a delta against itself: {superseded}"));
}
if chain_depth + 1 > MAX_GENERATION_CHAIN {
return Ok(SupersedeOutcome::ChainTooLong);
}
let (old_bytes, base_bytes) = {
let ar = crate::ZnippyArchive::open(archive)?;
// VERIFIED reads: the bytes about to become a checksum and a delta base
// must be the bytes the archive claims, or the delta is correct against
// something nobody stored.
(
ar.extract_file_verified(superseded)?,
ar.extract_file_verified(base)?,
)
};
let delta = crate::archive::encode_delta_against(&base_bytes, &old_bytes);
// The same two cutoffs `plan_version` applies, on real bytes: a delta that
// does not clearly win is not worth a chain link that is paid on every later
// read of this entry.
if (delta.len() as f64) >= crate::archive::DELTA_SIZE_ALPHA * (old_bytes.len() as f64) {
return Ok(SupersedeOutcome::NotSmaller {
delta_bytes: delta.len() as u64,
stored_bytes: old_bytes.len() as u64,
});
}
let mut sink = ArrowIpcSinkAppend::open_existing(archive)?;
let at = sink.blob_end();
// A delta between two packs is not obviously compressible — pack bytes are
// already deflated, and a diff of them mostly is too. So the codec is
// consulted and its answer taken, exactly as the append path does, rather
// than a rule being asserted either way (`precompressed.rs`).
let mut ctx = crate::codec::CompressCtx::new(compression_level)?;
let frame = ctx.compress(&delta).ok();
let (on_disk, compressed): (&[u8], bool) = match frame.as_deref() {
Some(f) if f.len() < delta.len() => (f, true),
_ => (&delta, false),
};
sink.file().write_all_at(on_disk, at)?;
sink.advance_blob_end(on_disk.len() as u64);
// Drop every old row of the superseded entry, and put one delta row in.
sink.replace_carried(superseded, vec![ChunkLoc {
chunk_seq: 0,
fdata_offset: 0,
blob_offset: at,
blob_size: on_disk.len() as u64,
// What the entry reconstructs TO, which is what the reader sizes and
// what the chunk's blake3 is over.
uncompressed_size: old_bytes.len() as u64,
compressed,
checksum: *blake3::hash(&old_bytes).as_bytes(),
}]);
sink.push_delta_map_row(superseded.to_string(), 0, base.to_string());
Box::new(sink).finish()?;
Ok(SupersedeOutcome::Delta {
stored_bytes: old_bytes.len() as u64,
delta_bytes: on_disk.len() as u64,
chain_depth: chain_depth + 1,
})
}
/// What one [`compact_archive`] reclaimed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CompactReport {
pub bytes_before: u64,
pub bytes_after: u64,
/// Live data rows carried across. A compaction never changes this.
pub rows: u64,
/// Delta-map rows carried across. Also never changed.
pub delta_rows: u64,
}
/// **Rewrite a sealed archive keeping only the bytes something still points at.**
///
/// The other half of [`supersede_as_delta`], and without it that writer cannot
/// deliver a byte. Superseding an entry writes the delta and drops the old
/// entry's index rows, but the old *blob* stays where it was, unreferenced —
/// MEASURED on oden 2026-08-05: an 8 056 624-byte archive went to **8 057 731**
/// after a supersede replaced 4 000 000 bytes of live payload with 11. An archive
/// fed one generation at a time therefore grows by a whole generation each time,
/// which is exactly what storing them whole would have cost. The saving
/// `supersede_as_delta` measures is in the archive's LIVE bytes; this is what
/// makes it the archive's size.
///
/// # It copies, and that is the point
///
/// Every live chunk's on-disk bytes are copied **verbatim** to the new file — no
/// decode, no re-encode, no delta re-computation, and the `compressed` flag,
/// `uncompressed_size` and blake3 of each chunk are carried unchanged. A delta
/// chunk stays a delta chunk at the same depth, and `__znippy_delta__` travels as
/// its decoded rows exactly as an append carries it. So a compaction costs one
/// pass over the live bytes and cannot change what any entry reads back as.
///
/// The obvious alternative — read every entry out and write a fresh archive —
/// cannot work here at all, and that is worth recording because it is what a
/// caller reaches for first: rebuilding stores every entry WHOLE, and superseding
/// them again reproduces exactly the dead payload the rebuild was meant to
/// remove.
///
/// Staged beside the destination and renamed over it, so an interruption at any
/// byte leaves the original archive, dead payload and all, serving.
pub fn compact_archive(archive: &Path) -> Result<CompactReport> {
let bytes_before = std::fs::metadata(archive)?.len();
let src = ArrowIpcSinkAppend::open_existing(archive)?;
let staged = {
let unique = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
let mut p = archive.as_os_str().to_owned();
p.push(format!(".compact-{}-{unique}", std::process::id()));
std::path::PathBuf::from(p)
};
let out = Arc::new(
std::fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(true)
.open(&staged)
.map_err(|e| anyhow!("compact: staging {}: {e}", staged.display()))?,
);
let mut sink = ArrowIpcSinkAppend::new(Arc::clone(&out), 0);
sink.meta = src.meta.clone();
sink.carried_reserved = src.carried_reserved.clone();
sink.delta_map = src.delta_map.clone();
let delta_rows = sink.delta_map.len() as u64;
let mut cursor = 0u64;
// Buffer reused across chunks: a repository's generations are packs, and
// allocating one per chunk is the kind of thing that turns a copy into a
// profile.
let mut buf: Vec<u8> = Vec::new();
for (path, loc) in &src.carried {
let n = loc.blob_size as usize;
buf.clear();
buf.resize(n, 0);
src.file
.read_exact_at(&mut buf, loc.blob_offset)
.map_err(|e| anyhow!("compact: reading {path} at {}: {e}", loc.blob_offset))?;
out.write_all_at(&buf, cursor)?;
let mut moved = loc.clone();
moved.blob_offset = cursor;
cursor += loc.blob_size;
sink.carried.push((path.clone(), moved));
}
let rows = sink.carried.len() as u64;
sink.cursor = cursor;
Box::new(sink).finish()?;
out.sync_all()?;
drop(out);
drop(src);
std::fs::rename(&staged, archive)?;
if let Some(parent) = archive.parent() {
if let Ok(f) = std::fs::File::open(parent) {
let _ = f.sync_all();
}
}
Ok(CompactReport {
bytes_before,
bytes_after: std::fs::metadata(archive)?.len(),
rows,
delta_rows,
})
}
/// What [`supersede_as_delta`] did, and why. Never silent: a caller that asked
/// for a delta and got a whole entry has to be able to see which.
#[derive(Debug, PartialEq, Eq)]
pub enum SupersedeOutcome {
/// Stored as a delta. `delta_bytes` is what is on disk now.
Delta { stored_bytes: u64, delta_bytes: u64, chain_depth: usize },
/// The delta did not clear [`crate::archive::DELTA_SIZE_ALPHA`]; the entry is
/// untouched.
NotSmaller { delta_bytes: u64, stored_bytes: u64 },
/// The chain would exceed [`MAX_GENERATION_CHAIN`]; keep this one whole. It
/// becomes the keyframe the next chain is built against.
ChainTooLong,
}
/// How many generations may chain before one is kept whole as a keyframe.
///
/// **32**, half the reader's [`MAX_RECONSTRUCT_DEPTH`] of 64. The reader's bound
/// is about a hostile index and is a refusal; this is a writer policy about cost,
/// and it is set below the refusal so a legitimate archive never approaches it.
///
/// The cost it bounds is MEASURED on a real eight-generation `nornir` chain
/// (oden, 2026-08-05, ~7.3-8.3 MB packs): reading the live generation is
/// **12.8 ms** and every link behind it adds **18.2 ms** — 6.4 ms of delta
/// application and decompression, 11.8 ms of blake3 over the reconstructed
/// entry. Thirty-two links is therefore ~0.6 s to read the OLDEST generation in
/// a full chain, and the reader's bound of 64 is ~1.2 s.
///
/// That is a cold archival read that nothing on a serve path makes: the live
/// generation holds every reachable object and sits at depth 0, whole, for ever.
/// Against it, each link saves a whole generation — measured **5.26x** over the
/// chain as a whole, 62 350 306 bytes of packs down to 11 860 793.
///
/// **The per-link blake3 is kept, deliberately.** It is 56% of that cold read and
/// it is the only thing that can catch a delta applied to the wrong base — which
/// is exactly the mistake a generation chain invites, because every entry in it
/// is a pack of the same repository at a similar size, and a base swapped for
/// its neighbour would apply cleanly. Trading that for 80 ms on a read nobody
/// makes is the wrong way round.
pub const MAX_GENERATION_CHAIN: usize = 32;
/// One base-schema data batch from `(path, ChunkLoc)` rows, in the order given.
///
/// The single builder for base-schema index rows. Four call sites used to carry
/// a copy of this loop — `emit_carried`, `write_lookup_and_trie`, `ArrowIpcSink`'s
/// two — and a column appended in one order in one of them and another order in
/// the next is a silent index corruption no checksum catches, because every
/// individual chunk still hashes correctly (LAW 5, by construction).
/// `pub` rather than `pub(crate)` since 2026-08-08: `znippy-plugin-git`'s
/// generation-0 seal needs exactly these columns in exactly this order for the
/// verbatim packs it already holds on disk, and a second copy of this loop in
/// that crate is the drift this function was extracted to prevent (LAW 5).
pub fn base_batch_from_rows(paths: &[String], locs: &[ChunkLoc]) -> Result<RecordBatch> {
let order: Vec<usize> = (0..paths.len()).collect();
base_batch_permuted(data_subindex_schema(), paths, locs, &order)
}
/// [`base_batch_from_rows`] emitting rows in `order` under an explicit `schema` —
/// the sorted-lookup case, which is the same columns in a different order.
pub(crate) fn base_batch_permuted(
schema: Arc<Schema>,
paths: &[String],
locs: &[ChunkLoc],
order: &[usize],
) -> Result<RecordBatch> {
let n = order.len();
let mut path_b = StringBuilder::with_capacity(n, n * 16);
let mut seq_b = UInt32Builder::with_capacity(n);
let mut fdata_b = UInt64Builder::with_capacity(n);
let mut comp_b = BooleanBuilder::with_capacity(n);
let mut usz_b = UInt64Builder::with_capacity(n);
let mut boff_b = UInt64Builder::with_capacity(n);
let mut bsz_b = UInt64Builder::with_capacity(n);
let mut ck_b = FixedSizeBinaryBuilder::with_capacity(n, 32);
for &i in order {
let loc = &locs[i];
path_b.append_value(&paths[i]);
seq_b.append_value(loc.chunk_seq);
fdata_b.append_value(loc.fdata_offset);
comp_b.append_value(loc.compressed);
usz_b.append_value(loc.uncompressed_size);
boff_b.append_value(loc.blob_offset);
bsz_b.append_value(loc.blob_size);
ck_b.append_value(loc.checksum).expect("checksum is 32 bytes");
}
Ok(RecordBatch::try_new(
schema,
vec![
Arc::new(path_b.finish()),
Arc::new(seq_b.finish()),
Arc::new(fdata_b.finish()),
Arc::new(comp_b.finish()),
Arc::new(usz_b.finish()),
Arc::new(boff_b.finish()),
Arc::new(bsz_b.finish()),
Arc::new(ck_b.finish()),
],
)?)
}
/// Compress-or-store each file and write its blob at `cursor`, returning the
/// index rows and the new cursor. **No metadata is touched** — this is the blob
/// half alone, shared by the re-sealing [`append_files`] path and by the hot
/// journal path, so the two cannot drift on the skip decision, the blake3 domain
/// (original bytes) or the store-raw rule.
pub(crate) fn write_blobs(
file: &File,
cursor: u64,
files: &[(String, Vec<u8>)],
ctx: &mut crate::codec::CompressCtx,
policy: crate::SkipPolicy,
) -> Result<(Vec<String>, Vec<ChunkLoc>, u64)> {
let mut paths = Vec::with_capacity(files.len());
let mut locs = Vec::with_capacity(files.len());
let mut cursor = cursor;
for (rel, bytes) in files {
let checksum = *blake3::hash(bytes).as_bytes();
// The skip decision, which the append path did not make at all until
// 2026-08-03: every byte went to the codec and the frame was thrown away
// whenever it came out no smaller. For already-compressed input — a
// `.pack`, a `.jar`, a `.crate` — that is the entire codec cost paid to
// learn what the file's name already said.
//
// MEASURED on gunnar's cold tier (oden, 2026-08-03): appending a
// 34.1 MiB consolidated packfile cost 0.34 s of CPU compressing and
// 0.03 s skipping — 11.3x — for byte-identical output.
let skip = policy.skip_by_path(std::path::Path::new(rel.as_str()));
let frame = if skip { Vec::new() } else { ctx.compress(bytes)? };
let (on_disk, compressed): (&[u8], bool) = if !skip && frame.len() < bytes.len() {
(&frame, true)
} else {
(bytes, false)
};
let blob_offset = cursor;
file.write_all_at(on_disk, blob_offset)?;
cursor += on_disk.len() as u64;
paths.push(rel.clone());
locs.push(ChunkLoc {
chunk_seq: 0,
fdata_offset: 0,
blob_offset,
blob_size: on_disk.len() as u64,
uncompressed_size: bytes.len() as u64,
compressed,
checksum,
});
}
Ok((paths, locs, cursor))
}
/// Outcome of a native [`append_files`] call.
#[derive(Debug, Clone)]
pub struct AppendReport {
/// Data rows that existed in the archive before the append (recovered),
/// including any that the append then replaced.
pub rows_before: u64,
/// Pre-existing rows dropped because the append re-wrote the same
/// `relative_path` (replace semantics). Always 0 on the fresh-create path.
pub rows_replaced: u64,
/// New rows (one per appended file/chunk) written by the append.
pub rows_added: u64,
/// Byte offset where the appended blob region started (old blob_end).
pub blob_append_offset: u64,
/// Bytes of new blob payload appended (compressed/stored).
pub blob_bytes_added: u64,
/// Final size of the re-sealed archive.
pub sealed_total_bytes: u64,
}
/// First-class **native blob append** — the caller-facing library primitive that
/// the iceberg lifecycle test (#23) previously had to perform by hand. (There is
/// no `compress --append` CLI verb today; this is the in-process entry point.)
///
/// Opens an existing sealed v0.7 `.znippy`, compresses each `(relative_path,
/// bytes)` in `new_files` with the znippy codec, appends the resulting blobs to
/// the same file past the existing blob region, then re-seals (merged old+new
/// lookup + trie + manifest + footer) via [`ArrowIpcSinkAppend`]. The original
/// blob bytes and existing rows are reused verbatim — nothing is recompressed.
///
/// **Replace semantics:** a `relative_path` in `new_files` that the archive
/// already contains REPLACES the existing entry — the pre-existing rows for that
/// path are dropped from the re-sealed index (counted in
/// [`AppendReport::rows_replaced`]) and its old blob bytes become unreferenced
/// dead payload. Appending the same path twice never leaves two live copies.
///
/// Mirrors the real compress path's per-blob accounting: blake3 over the
/// ORIGINAL bytes, store-raw when the codec frame is not smaller, one chunk per
/// file (`chunk_seq = 0`). `compression_level` is the codec level (e.g. 3).
pub fn append_files(
archive: &Path,
new_files: &[(String, Vec<u8>)],
compression_level: i32,
) -> Result<AppendReport> {
append_files_with_meta(archive, new_files, compression_level, None)
}
/// [`append_files`] with an explicit [`SkipPolicy`](crate::SkipPolicy).
///
/// The batch-level form, for a caller that already knows what it is appending —
/// `SkipPolicy::already_compressed()` stores every entry raw without inspecting
/// a byte, which is exact, free, and better informed than any probe. gunnar's
/// cold tier appends packfiles and uses it.
pub fn append_files_with_policy(
archive: &Path,
new_files: &[(String, Vec<u8>)],
compression_level: i32,
policy: crate::SkipPolicy,
) -> Result<AppendReport> {
let sink = ArrowIpcSinkAppend::open_existing(archive)?;
let rows_before = sink.recovered_rows() as u64;
let blob_append_offset = sink.blob_end();
write_files_into_sink(
sink,
new_files,
compression_level,
rows_before,
blob_append_offset,
policy,
)
}
/// [`append_files`], additionally merging `meta` rows into the archive's
/// searchable metadata sub-index.
///
/// `None` leaves the metadata exactly as the archive had it — including having
/// none. `Some(rows)` merges into whatever was already there (the resume carries
/// the old section forward), creating the section if the archive had none.
pub fn append_files_with_meta(
archive: &Path,
new_files: &[(String, Vec<u8>)],
compression_level: i32,
meta: Option<MetaTable>,
) -> Result<AppendReport> {
let mut sink = ArrowIpcSinkAppend::open_existing(archive)?;
if let Some(table) = meta {
sink.merge_meta(table.rows().to_vec());
}
let rows_before = sink.recovered_rows() as u64;
let blob_append_offset = sink.blob_end();
write_files_into_sink(
sink,
new_files,
compression_level,
rows_before,
blob_append_offset,
// Resolve per entry from its name, then from its bytes — the same
// default `compress_dir` has. Deliberately NOT "always compress", which
// is what this path did before and which is never the right answer for
// an entry whose extension already says it is compressed.
crate::SkipPolicy::resolve(),
)
}
/// Create a fresh `.znippy` archive from in-memory `files` — the bootstrap inverse
/// of [`append_files`], which requires an already-sealed archive (it rejects an
/// empty manifest). Seed a new writable archive with this, then grow it with
/// [`append_files`]. Overwrites `archive` if it already exists.
pub fn create_archive(
archive: &Path,
files: &[(String, Vec<u8>)],
compression_level: i32,
) -> Result<AppendReport> {
create_archive_with_meta(archive, files, compression_level, None)
}
/// [`create_archive`], additionally sealing a searchable metadata sub-index.
///
/// `None` seals **no section** — the archive reads back as
/// `ArchiveMeta::NoMetadata` and is byte-identical to one from
/// [`create_archive`]. `Some(table)` seals the section even when the table is
/// empty, which reads back as a present-but-empty index: "searched, records
/// nothing", a different statement from "never had an index".
pub fn create_archive_with_meta(
archive: &Path,
files: &[(String, Vec<u8>)],
compression_level: i32,
meta: Option<MetaTable>,
) -> Result<AppendReport> {
let blob_file = Arc::new(
File::create(archive)
.map_err(|e| anyhow!("create archive {}: {e}", archive.display()))?,
);
let mut sink = ArrowIpcSinkAppend::new(blob_file, 0);
sink.meta = meta;
write_files_into_sink(sink, files, compression_level, 0, 0, crate::SkipPolicy::resolve())
}
/// Build a complete `.znippy` archive **entirely in memory** from in-memory
/// `files` and return its bytes — no staging directory, no named output file. The
/// sink needs a positioned-write fd, so this seals into an **anonymous temp file**
/// (`O_TMPFILE` where the OS supports it → never linked into the filesystem
/// namespace), then reads the sealed bytes straight back out. The returned `Vec`
/// is byte-identical to what [`create_archive`] would write to a path.
///
/// Use this for zero-disk pipelines: build a release/airgap archive from product
/// bytes held in RAM and stream it across the gap without ever touching disk on
/// the build side. Pair with [`append_files`] (path) or hold the bytes and re-seal.
pub fn create_archive_to_vec(
files: &[(String, Vec<u8>)],
compression_level: i32,
) -> Result<(Vec<u8>, AppendReport)> {
let anon = Arc::new(
tempfile::tempfile().map_err(|e| anyhow!("anonymous archive fd: {e}"))?,
);
let sink = ArrowIpcSinkAppend::new(anon.clone(), 0);
let report =
write_files_into_sink(sink, files, compression_level, 0, 0, crate::SkipPolicy::resolve())?;
// The Arc keeps the anonymous fd alive past `finish()`; read the sealed bytes.
let mut bytes = vec![0u8; report.sealed_total_bytes as usize];
anon.read_exact_at(&mut bytes, 0)
.map_err(|e| anyhow!("read back anonymous archive: {e}"))?;
Ok((bytes, report))
}
/// Shared core of [`append_files`] / [`create_archive`]: compress each file's bytes
/// (store-raw if not smaller), write the blobs at the sink's running blob cursor,
/// then push one base-schema data sub-index and seal. `sink` is either a fresh
/// [`ArrowIpcSinkAppend::new`] or an [`ArrowIpcSinkAppend::open_existing`].
fn write_files_into_sink(
mut sink: ArrowIpcSinkAppend,
new_files: &[(String, Vec<u8>)],
compression_level: i32,
rows_before: u64,
blob_append_offset: u64,
policy: crate::SkipPolicy,
) -> Result<AppendReport> {
use crate::codec::CompressCtx;
// Replace, don't duplicate: a path being (re-)written now supersedes whatever
// rows the archive already held for it. The old blob bytes stay in the file as
// dead payload — they are simply no longer referenced by any index row.
let incoming: std::collections::HashSet<&str> =
new_files.iter().map(|(rel, _)| rel.as_str()).collect();
let rows_replaced = sink.drop_carried_paths(&incoming) as u64;
// Append the new blob bytes to the file at the running blob cursor, mirroring
// the compress pipeline (hash original bytes; store-raw if not smaller). One
// writer, shared with the hot journal path.
let blob_file = sink.file.clone();
let mut ctx = CompressCtx::new(compression_level)?;
let (paths, locs, cursor) =
write_blobs(&blob_file, blob_append_offset, new_files, &mut ctx, policy)?;
let blob_bytes_added = cursor - blob_append_offset;
blob_file.sync_all()?;
// Advance the sink's cursor past the freshly-written blob region so the new
// data sub-index lands after the appended blobs (not over them).
sink.cursor = cursor;
// A DATA sub-index: seal it with the format-version-stamped schema. With the
// bare `lookup_schema()` every archive this path writes — i.e. every archive
// a writable holger repo or `cargo publish` produces — recorded no format
// version, so the reader-side version pin had nothing to check.
let batch = base_batch_from_rows(&paths, &locs)?;
let schema = data_subindex_schema();
let rows_added = batch.num_rows() as u64;
sink.push_subindex(
schema.as_ref(),
&[batch],
GroupKey { pkg_type: 0, repo: String::new(), module_name: String::new() },
)?;
let sealed_total_bytes = Box::new(sink).finish()?;
Ok(AppendReport {
rows_before,
rows_replaced,
rows_added,
blob_append_offset,
blob_bytes_added,
sealed_total_bytes,
})
}
/// Recover existing DATA rows for the resume accumulator. Reads the sorted
/// lookup reserved section (the cheapest source — already the exact per-chunk
/// rows, sorted) when present; otherwise concatenates the data sub-indexes.
pub(crate) fn recover_rows(
path: &Path,
entries: &[ManifestEntry],
) -> Result<(Vec<String>, Vec<ChunkLoc>)> {
use std::io::{Read, Seek, SeekFrom};
let mut file = File::open(path)?;
// Prefer the sorted lookup section: it is exactly the base-schema per-chunk
// rows, one cheap stream decode.
if let Some(lk) = entries.iter().find(|e| e.module_name == LOOKUP_MODULE) {
file.seek(SeekFrom::Start(lk.index_offset))?;
let mut bytes = vec![0u8; lk.index_len as usize];
file.read_exact(&mut bytes)?;
return decode_base_rows(&bytes);
}
// Fallback: read every NON-reserved data sub-index and concatenate its rows.
let mut paths = Vec::new();
let mut locs = Vec::new();
for e in entries {
if is_reserved_module(&e.module_name) {
continue;
}
file.seek(SeekFrom::Start(e.index_offset))?;
let mut bytes = vec![0u8; e.index_len as usize];
file.read_exact(&mut bytes)?;
let (mut p, mut l) = decode_base_rows(&bytes)?;
paths.append(&mut p);
locs.append(&mut l);
}
Ok((paths, locs))
}
// ── inject-assert tests (the "tests inject values, not just no-crash" LAW) ──
// Every test here seals a real archive through a `CompressCtx`, so all of them
// need the codec — see the note on `archive::tests`.
#[cfg(all(test, feature = "openzl"))]
mod tests {
use super::*;
use crate::codec::CompressCtx;
use crate::meta::{BlobMeta, ChunkMeta};
use crate::{ArrowIpcSink, ZnippyArchive, ZnippyReader};
use std::time::{SystemTime, UNIX_EPOCH};
fn unique_dir(tag: &str) -> std::path::PathBuf {
let ns = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
let d = std::env::temp_dir().join(format!("znippy_append_{tag}_{ns}_{:?}", std::thread::current().id()));
std::fs::create_dir_all(&d).unwrap();
d
}
/// **The append path takes the skip decision, and it did not used to.**
///
/// Until 2026-08-03 `write_files_into_sink` ran `CompressCtx::compress` over
/// every byte of every appended file and kept the frame only when it came
/// out smaller. For already-compressed input that is the entire codec cost
/// paid to learn nothing — `compress_dir` has consulted `SkipPolicy` since
/// it existed, and this path silently did not.
///
/// The observable is chosen so it cannot be faked. A *highly compressible*
/// payload is appended under a `.pack` name, which the extension table
/// declares already-compressed. If the policy is honoured the blob is stored
/// RAW and `blob_bytes_added` equals the input length; if the codec runs, the
/// frame is far smaller and the number collapses. Asserting on CPU time
/// would have been the honest measure of the bug but is not a test; this is
/// the same decision made visible in a byte count.
///
/// Seen RED by restoring the unconditional `ctx.compress(bytes)`:
/// `blob_bytes_added` drops to a few hundred bytes and the assertion fires.
#[test]
fn an_append_honours_the_skip_policy_instead_of_compressing_everything() {
let dir = unique_dir("skip_policy");
let archive = dir.join("a.znippy");
create_archive(&archive, &[("seed.txt".into(), b"seed".to_vec())], 3).unwrap();
// 256 KiB of one byte: the codec would crush this to almost nothing.
let squishy = vec![b'A'; 256 * 1024];
let name = format!("pack-{}.pack", "0f".repeat(20));
let report =
append_files(&archive, &[(name.clone(), squishy.clone())], 3).unwrap();
assert_eq!(
report.blob_bytes_added,
squishy.len() as u64,
"a `.pack` entry must be stored RAW. {} bytes were written for a {}-byte input, so \
the codec ran over a file the extension table already said was compressed",
report.blob_bytes_added,
squishy.len()
);
// …and it still reads back byte-exact, which is what stored-raw has to mean.
assert_eq!(crate::get_file(&archive, &name).unwrap(), squishy);
// The MIRROR, so the test above is not merely asserting that nothing is
// ever compressed: the identical bytes under an ordinary name DO go
// through the codec.
let report2 =
append_files(&archive, &[("plain.txt".into(), squishy.clone())], 3).unwrap();
assert!(
report2.blob_bytes_added < squishy.len() as u64 / 10,
"an ordinary name must still be compressed; {} bytes for {}",
report2.blob_bytes_added,
squishy.len()
);
assert_eq!(crate::get_file(&archive, "plain.txt").unwrap(), squishy);
// And an explicit batch-level claim overrules the name entirely.
let report3 = append_files_with_policy(
&archive,
&[("also-plain.txt".into(), squishy.clone())],
3,
crate::SkipPolicy::already_compressed(),
)
.unwrap();
assert_eq!(
report3.blob_bytes_added,
squishy.len() as u64,
"`already_compressed()` must store raw whatever the name says"
);
std::fs::remove_dir_all(&dir).ok();
}
/// Deterministic synthetic rows — distinct, lexicographically-spread paths
/// (same flavour as the bench's `synth_blobs`, so the sort/fst do real work).
fn synth(n: usize, salt: u64) -> Vec<(String, Vec<u8>)> {
(0..n)
.map(|i| {
let g = (i.wrapping_mul(2_654_435_761) ^ salt as usize) % 1000;
let p = format!("repo/grp{g:03}/file{:08}_{salt}.bin", i);
let body = format!("payload {i} salt {salt} {}\n", "z".repeat(8 + (i % 40)));
(p, body.into_bytes())
})
.collect()
}
/// Write a fresh sealed archive with `sink` (codec-compressed blobs + one
/// base-schema data sub-index + the seal). Returns the sealed length. Generic
/// over a closure so we can drive BOTH ArrowIpcSink (A) and the clone (B)
/// through the *identical* fresh path and compare bytes.
fn write_fresh<S: ArchiveMetaSink + 'static>(
path: &Path,
files: &[(String, Vec<u8>)],
make_sink: impl FnOnce(Arc<File>, u64) -> S,
) -> u64 {
let file = Arc::new(File::create(path).unwrap());
let mut ctx = CompressCtx::new(3).unwrap();
let mut blobs = Vec::new();
let mut paths = Vec::new();
let mut cursor = 0u64;
for (fi, (rel, bytes)) in files.iter().enumerate() {
let checksum = *blake3::hash(bytes).as_bytes();
let frame = ctx.compress(bytes).unwrap();
let (on_disk, compressed): (&[u8], bool) =
if frame.len() < bytes.len() { (&frame, true) } else { (bytes, false) };
file.write_all_at(on_disk, cursor).unwrap();
let blob_offset = cursor;
cursor += on_disk.len() as u64;
paths.push(rel.clone());
blobs.push(BlobMeta {
blob_offset,
blob_size: on_disk.len() as u64,
chunk_meta: ChunkMeta {
fdata_offset: 0,
file_index: fi as u64,
chunk_seq: 0,
checksum,
compressed,
uncompressed_size: bytes.len() as u64,
compressed_size: on_disk.len() as u64,
},
});
}
let resolver = { let p = paths.clone(); move |fi: u64| p[fi as usize].clone() };
let batch = crate::build_metadata_batch(&blobs, resolver, &[], &[]).unwrap();
// Same DATA sub-index schema `write_files_into_sink` seals with, so this
// helper stays the byte-for-byte reference for `create_archive`.
let schema = data_subindex_schema();
let mut sink = make_sink(file.clone(), cursor);
sink.push_subindex(
schema.as_ref(),
&[batch],
GroupKey { pkg_type: 0, repo: String::new(), module_name: String::new() },
)
.unwrap();
Box::new(sink).finish().unwrap()
}
/// PARITY: the clone's fresh write+seal must produce a BYTE-IDENTICAL archive
/// to the original `ArrowIpcSink` over the same rows. This is the structural
/// proof that the clone is a zero-cost superset on the normal path (the
/// throughput parity number lives in the bench; this asserts correctness).
#[test]
fn clone_fresh_path_is_byte_identical_to_original() {
let dir = unique_dir("parity");
let files = synth(2_000, 1);
let a = dir.join("a.znippy");
let b = dir.join("b.znippy");
let len_a = write_fresh(&a, &files, ArrowIpcSink::new);
let len_b = write_fresh(&b, &files, ArrowIpcSinkAppend::new);
assert_eq!(len_a, len_b, "clone seal produced a different total length");
let bytes_a = std::fs::read(&a).unwrap();
let bytes_b = std::fs::read(&b).unwrap();
assert_eq!(
bytes_a, bytes_b,
"clone's fresh write path is NOT byte-identical to ArrowIpcSink — parity broken"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// RESUME: native append → reopen with the ORDINARY arrow-ipc reader → both
/// the original AND the appended files read back byte-exact, and the index
/// lists exactly old+new. Inject real bytes, assert real bytes out.
#[test]
fn native_append_roundtrips_old_and_new_files() {
let dir = unique_dir("resume");
let archive = dir.join("store.znippy");
let orig = synth(1_500, 7);
write_fresh(&archive, &orig, ArrowIpcSink::new);
let added = synth(300, 99);
let report = append_files(&archive, &added, 3).unwrap();
assert_eq!(report.rows_before, orig.len() as u64, "must recover all original rows");
assert_eq!(report.rows_added, added.len() as u64);
assert!(report.blob_bytes_added > 0, "append must write new blob bytes");
assert!(
report.sealed_total_bytes > report.blob_append_offset,
"re-sealed file must be larger than the old blob region"
);
// Reopen with the plain reader (no append awareness) and verify EVERY
// file — original and appended — comes back byte-exact.
let ar = ZnippyArchive::open(&archive).unwrap();
let mut listed = ar.list_files().unwrap();
listed.sort();
let mut expected: Vec<String> =
orig.iter().chain(added.iter()).map(|(p, _)| p.clone()).collect();
expected.sort();
assert_eq!(listed, expected, "index must list exactly old+new files after append");
for (p, bytes) in orig.iter().chain(added.iter()) {
let got = ar.extract_file(p).unwrap();
assert_eq!(&got, bytes, "byte mismatch after append for {p}");
}
// Random-access lookup of an appended file via the rebuilt trie+lookup.
let probe = &added[123].0;
let chunks = crate::locate_file(&archive, probe).unwrap();
assert!(!chunks.is_empty(), "appended file must be locatable via the re-sealed lookup");
let _ = std::fs::remove_dir_all(&dir);
}
/// BOOTSTRAP: `create_archive` seeds a fresh archive that (a) is byte-identical
/// to the proven fresh write path, (b) reads back byte-exact via the ordinary
/// reader, and (c) can then be grown by `append_files` — the create→append flow
/// holger's writable `put` relies on. Inject real bytes, assert real bytes out.
#[test]
fn create_archive_seeds_then_grows() {
let dir = unique_dir("create");
let seed = synth(40, 5);
// (a) create_archive == write_fresh(_, ArrowIpcSinkAppend::new) byte-for-byte.
let made = dir.join("made.znippy");
let report = create_archive(&made, &seed, 3).unwrap();
assert_eq!(report.rows_before, 0, "fresh archive has no prior rows");
assert_eq!(report.rows_added, seed.len() as u64);
let ref_path = dir.join("ref.znippy");
write_fresh(&ref_path, &seed, ArrowIpcSinkAppend::new);
assert_eq!(
std::fs::read(&made).unwrap(),
std::fs::read(&ref_path).unwrap(),
"create_archive must be byte-identical to the proven fresh write path"
);
// (b) seeded files read back byte-exact via the plain reader.
let ar = ZnippyArchive::open(&made).unwrap();
for (p, bytes) in &seed {
assert_eq!(&ar.extract_file(p).unwrap(), bytes, "seed byte mismatch for {p}");
}
// (c) append_files grows the bootstrapped archive; old+new read back exact.
let added = synth(15, 88);
let rep2 = append_files(&made, &added, 3).unwrap();
assert_eq!(rep2.rows_before, seed.len() as u64, "append must recover seeded rows");
assert_eq!(rep2.rows_added, added.len() as u64);
let ar2 = ZnippyArchive::open(&made).unwrap();
for (p, bytes) in seed.iter().chain(added.iter()) {
assert_eq!(&ar2.extract_file(p).unwrap(), bytes, "byte mismatch after grow for {p}");
}
let _ = std::fs::remove_dir_all(&dir);
}
/// ZERO-COPY / no-filesystem: `create_archive_to_vec` builds a valid archive
/// from in-memory bytes with NO staging tree and NO named output file (it seals
/// into an anonymous fd). Assert the returned bytes are byte-identical to
/// `create_archive`'s file output AND read back byte-exact through the reader.
#[test]
fn create_archive_to_vec_is_filesystem_free_and_round_trips() {
let files = synth(24, 7);
let (bytes, report) = create_archive_to_vec(&files, 3).unwrap();
assert_eq!(report.rows_before, 0, "fresh in-memory archive has no prior rows");
assert_eq!(report.rows_added, files.len() as u64);
assert_eq!(bytes.len() as u64, report.sealed_total_bytes, "vec len == sealed size");
let dir = unique_dir("tovec");
// (1) byte-identical to the proven path (`create_archive` → file).
let ref_path = dir.join("ref.znippy");
create_archive(&ref_path, &files, 3).unwrap();
assert_eq!(
bytes,
std::fs::read(&ref_path).unwrap(),
"in-memory archive must be byte-identical to create_archive's file output"
);
// (2) the in-memory bytes ARE a real archive: persist + read back byte-exact.
let p = dir.join("from_mem.znippy");
std::fs::write(&p, &bytes).unwrap();
let ar = ZnippyArchive::open(&p).unwrap();
for (name, content) in &files {
assert_eq!(&ar.extract_file(name).unwrap(), content, "byte mismatch for {name}");
}
let _ = std::fs::remove_dir_all(&dir);
}
/// Read an archive's recorded on-disk format version **exactly the way a
/// reader does** — manifest, then the Arrow schema metadata of the FIRST
/// non-reserved sub-index. This is byte-for-byte the same walk as
/// `znippy_common::read_znippy_index`'s `check_format_version` call and as
/// holger's independent `traits::recorded_format_version` pin, so what this
/// helper returns is what those two see.
fn recorded_format_version(path: &Path) -> Option<String> {
use std::io::{Read, Seek, SeekFrom};
let entries = crate::index::read_znippy_manifest(path).ok()?;
let mut file = File::open(path).ok()?;
for e in &entries {
if is_reserved_module(&e.module_name) {
continue;
}
file.seek(SeekFrom::Start(e.index_offset)).ok()?;
let mut bytes = vec![0u8; e.index_len as usize];
file.read_exact(&mut bytes).ok()?;
let reader = crate::ipc_guard::stream_reader(bytes).ok()?;
return reader
.schema()
.metadata()
.get(crate::index::FORMAT_VERSION_KEY)
.cloned();
}
None
}
/// FORMAT VERSION: every archive the append/create path writes must RECORD
/// its on-disk format version — on the fresh create, on the in-memory create,
/// and still after a re-seal by `append_files` (which rebuilds the whole
/// metadata tail, so a stamp that only survives the fresh path is no stamp).
///
/// This is what makes a reader-side version pin mean anything for the
/// archives this path produces — a writable holger repo and `cargo publish`
/// both write through here. Sealing the data sub-index with the bare
/// `lookup_schema()` (no schema metadata) recorded NO version at all, and the
/// pin silently degraded to "undetermined → read as before" for exactly those
/// archives.
#[test]
fn every_appended_archive_records_the_format_version() {
let dir = unique_dir("fmtver");
let want = crate::index::ZNIPPY_FORMAT_VERSION.to_string();
// (a) fresh create.
let made = dir.join("made.znippy");
create_archive(&made, &synth(12, 3), 3).unwrap();
assert_eq!(
recorded_format_version(&made).as_deref(),
Some(want.as_str()),
"create_archive must stamp the on-disk format version"
);
// (b) after a re-seal: `append_files` rebuilds the metadata tail from
// scratch, so the stamp has to be re-emitted, not merely inherited.
append_files(&made, &synth(7, 91), 3).unwrap();
assert_eq!(
recorded_format_version(&made).as_deref(),
Some(want.as_str()),
"append_files must re-stamp the format version on the re-sealed archive"
);
// (c) the in-memory create path seals through the same code.
let (bytes, _) = create_archive_to_vec(&synth(9, 4), 3).unwrap();
let mem = dir.join("mem.znippy");
std::fs::write(&mem, &bytes).unwrap();
assert_eq!(
recorded_format_version(&mem).as_deref(),
Some(want.as_str()),
"create_archive_to_vec must stamp the on-disk format version"
);
// The stamp did not cost readability: rows still read back byte-exact.
let ar = ZnippyArchive::open(&mem).unwrap();
for (p, body) in &synth(9, 4) {
assert_eq!(&ar.extract_file(p).unwrap(), body, "byte mismatch for {p}");
}
let _ = std::fs::remove_dir_all(&dir);
}
/// METADATA SEARCH, end to end through a real archive:
/// (a) an archive sealed WITHOUT metadata reads back as `NoMetadata` — the
/// backward-compatible case, since that is exactly what every archive
/// written before this module looks like;
/// (b) sealing metadata makes "which entries carry key X" answerable, and
/// the answer resolves to real bytes via the ordinary reader;
/// (c) a re-seal by `append_files` CARRIES the metadata forward and merges
/// new rows into it — without that, every append would silently erase it;
/// (d) the metadata rows never leak into the data index or the lookup.
#[test]
fn metadata_is_searchable_survives_a_reseal_and_absence_is_reported_as_absence() {
use crate::meta_index::{ArchiveMeta, MetaSearch, MetaTable, MetaValue, read_archive_meta};
let dir = unique_dir("meta");
let files = synth(30, 2);
// (a) BACKWARD COMPAT: no metadata sealed → NoMetadata, not "found nothing".
let plain = dir.join("plain.znippy");
create_archive(&plain, &files, 3).unwrap();
let m = read_archive_meta(&plain).unwrap();
assert_eq!(m, ArchiveMeta::NoMetadata, "an archive with no meta section must say so");
assert!(!m.is_searchable());
assert_eq!(m.find_by_key("build-thing"), MetaSearch::NoMetadata);
assert!(
m.find_by_key("build-thing").hits().is_none(),
"absence must NOT present itself as an empty result set"
);
// A present-but-EMPTY index is the other state, and it is distinguishable.
let empty = dir.join("empty.znippy");
create_archive_with_meta(&empty, &files, 3, Some(MetaTable::new())).unwrap();
let me = read_archive_meta(&empty).unwrap();
assert!(me.is_searchable(), "a sealed empty index WAS searched");
assert!(me.index().is_some_and(|i| i.is_empty()));
assert_eq!(me.find_by_key("build-thing"), MetaSearch::Hits(&[]));
// (b) SEARCHABLE: two entries carry a build-thing, one does not.
let wasm = b"\0asm\x01\0\0\0".to_vec();
let (p0, p1, p2) = (files[0].0.clone(), files[1].0.clone(), files[2].0.clone());
let mut t = MetaTable::new();
t.insert(p0.clone(), "build-thing", MetaValue::Bytes(wasm.clone()))
.insert(p0.clone(), "build-thing.abi", "wasi-p2")
.insert(p1.clone(), "build-thing", MetaValue::Bytes(wasm.clone()))
.insert(p2.clone(), "coverage", 0.5f64)
.insert_archive("producer", "znippy");
let ar = dir.join("meta.znippy");
create_archive_with_meta(&ar, &files, 3, Some(t)).unwrap();
let m = read_archive_meta(&ar).unwrap();
let idx = m.index().expect("sealed index is present");
assert_eq!(idx.len(), 5);
let hits = m.find_by_key("build-thing").hits().unwrap();
assert_eq!(hits.len(), 2, "exactly the two entries that carry one");
let mut got: Vec<&str> = hits.iter().filter_map(|h| h.path()).collect();
got.sort();
let mut want = vec![p0.as_str(), p1.as_str()];
want.sort();
assert_eq!(got, want, "the search names the right ENTRIES");
assert_eq!(hits[0].value.as_bytes(), Some(&wasm[..]), "and the right VALUE");
assert_eq!(idx.archive_value("producer").and_then(MetaValue::as_str), Some("znippy"));
assert_eq!(
idx.find_by_prefix("build-thing").len(),
3,
"prefix sweeps build-thing + build-thing.abi"
);
// The hit resolves to real bytes without extracting anything else.
let reader = ZnippyArchive::open(&ar).unwrap();
let want_bytes = &files.iter().find(|(p, _)| *p == p0).unwrap().1;
assert_eq!(&reader.extract_file(hits[0].path().unwrap()).unwrap(), want_bytes);
// (c) A RE-SEAL keeps it, and merges.
let added = synth(6, 77);
let mut more = MetaTable::new();
more.insert(added[0].0.clone(), "build-thing", MetaValue::Bytes(wasm.clone()));
append_files_with_meta(&ar, &added, 3, Some(more)).unwrap();
let m2 = read_archive_meta(&ar).unwrap();
let hits2 = m2.find_by_key("build-thing").hits().unwrap();
assert_eq!(hits2.len(), 3, "the append merged, it did not replace");
assert_eq!(
m2.index().unwrap().archive_value("producer").and_then(MetaValue::as_str),
Some("znippy"),
"the archive-level row survived the re-seal"
);
// A plain `append_files` (no meta argument) must also preserve it.
append_files(&ar, &synth(3, 91), 3).unwrap();
assert_eq!(
read_archive_meta(&ar).unwrap().find_by_key("build-thing").hits().unwrap().len(),
3,
"an append that says nothing about metadata must not erase it"
);
// (d) The metadata rows are NOT file rows: the data index and the lookup
// see only the real entries.
let ar2 = ZnippyArchive::open(&ar).unwrap();
let n_files = files.len() + added.len() + 3;
assert_eq!(
ar2.file_count(),
n_files,
"metadata rows leaked into the data index"
);
let lookup_bytes = read_reserved_section_bytes(&ar, LOOKUP_MODULE).unwrap().unwrap();
assert_eq!(
decode_base_rows(&lookup_bytes).unwrap().0.len(),
n_files,
"metadata rows leaked into the random-access lookup"
);
for (p, bytes) in files.iter().chain(added.iter()) {
assert_eq!(&ar2.extract_file(p).unwrap(), bytes, "byte mismatch for {p}");
}
let _ = std::fs::remove_dir_all(&dir);
}
/// Cached `ArchiveReader` must return byte-for-byte the SAME chunks and
/// per-file metadata as the re-reading free functions — for a present file,
/// an absent file, and a prefix window. Proves idea (B) is a pure read-side
/// cache with zero behavioural drift from `locate_file` /
/// `get_files_meta_with_prefix`.
/// **An append kept the blobs and threw away the ref log.**
///
/// MEASURED 2026-08-04 on gunnar: an object-carrying push took the archive
/// from 144 928 to 146 396 bytes and `__gunnar_refs__` was **gone**.
/// `open_existing` truncates the whole metadata tail and re-supplies no
/// reserved section it was not handed, and nothing hands it one — so every
/// push after the first erased the ref history, which is what blocks serving
/// refs from the archive for most of a repository's life.
///
/// The distinction the append path could not express: `__gunnar_graph__` /
/// `__gunnar_reach__` / `__gunnar_oid__` are DERIVED from the objects and
/// must go when the objects change; `__gunnar_refs__` / `__gunnar_secrets__`
/// are INDEPENDENT logs and nothing else can reproduce them. Both halves are
/// asserted here — carrying everything would be the opposite bug and would
/// leave a stale commit graph behind.
///
/// Seen RED by dropping the `write_carried_reserved()` call from `finish`:
/// `__gunnar_refs__` reads back as `None` and the first assertion fires.
#[test]
fn an_append_carries_the_ref_log_forward_and_drops_the_derived_sections() {
use crate::index::{
GUNNAR_GRAPH_MODULE, GUNNAR_REFS_MODULE, GUNNAR_SECRETS_MODULE,
};
use crate::meta_sink::{ReservedSection, ReservedSectionBuilder};
let dir = unique_dir("carried_reserved");
let archive = dir.join("a.znippy");
let refs_bytes = b"refs/heads/main 0123456789abcdef -- push 1".to_vec();
let secrets_bytes = b"\x00ciphertext-only, never plaintext".to_vec();
let graph_bytes = b"a commit graph derived from the objects".to_vec();
// Seal an archive carrying one independent log, one secrets log and one
// DERIVED section, through the same builder gunnar's cold tier uses.
{
let f = Arc::new(File::create(&archive).unwrap());
let mut cursor = 0u64;
let mut blobs = Vec::new();
for (i, (name, bytes)) in [("obj/a.bin", b"first".to_vec())].iter().enumerate() {
use std::os::unix::fs::FileExt;
f.write_all_at(bytes, cursor).unwrap();
blobs.push(BlobMeta {
blob_offset: cursor,
blob_size: bytes.len() as u64,
chunk_meta: ChunkMeta {
fdata_offset: 0,
file_index: i as u64,
chunk_seq: 0,
checksum: *blake3::hash(bytes).as_bytes(),
compressed: false,
uncompressed_size: bytes.len() as u64,
compressed_size: bytes.len() as u64,
},
});
cursor += bytes.len() as u64;
let _ = name;
}
let batch = crate::index::build_metadata_batch(
&blobs,
|_fi: u64| "obj/a.bin".to_string(),
&[],
&[],
)
.unwrap();
// The sink starts AFTER the blob region, or it writes its index over
// the bytes it is indexing.
let mut sink = ArrowIpcSink::new(Arc::clone(&f), cursor);
let (r, s, g) = (refs_bytes.clone(), secrets_bytes.clone(), graph_bytes.clone());
let builder: ReservedSectionBuilder = Box::new(move |_lookup| {
Ok(vec![
ReservedSection::raw(GUNNAR_REFS_MODULE, r),
ReservedSection::raw(GUNNAR_SECRETS_MODULE, s),
ReservedSection::raw(GUNNAR_GRAPH_MODULE, g),
])
});
sink = sink.with_reserved_builder(builder);
sink.push_subindex(
crate::index::lookup_schema().as_ref(),
&[batch],
GroupKey { pkg_type: 0, repo: String::new(), module_name: String::new() },
)
.unwrap();
Box::new(sink).finish().unwrap();
}
// Sanity: all three are there before the append, or this proves nothing.
for m in [GUNNAR_REFS_MODULE, GUNNAR_SECRETS_MODULE, GUNNAR_GRAPH_MODULE] {
assert!(
read_reserved_section_bytes(&archive, m).unwrap().is_some(),
"{m} must be present BEFORE the append"
);
}
// An object-carrying push.
append_files(&archive, &[("obj/b.bin".to_string(), b"second".to_vec())], 3).unwrap();
// The independent logs survive, byte for byte.
assert_eq!(
read_reserved_section_bytes(&archive, GUNNAR_REFS_MODULE).unwrap(),
Some(refs_bytes),
"an object-carrying push must not erase the ref log — this is the 2026-08-04 bug"
);
assert_eq!(
read_reserved_section_bytes(&archive, GUNNAR_SECRETS_MODULE).unwrap(),
Some(secrets_bytes),
);
// The DERIVED section is gone, because the objects changed under it.
assert_eq!(
read_reserved_section_bytes(&archive, GUNNAR_GRAPH_MODULE).unwrap(),
None,
"a commit graph derived from the old object set must NOT be carried forward"
);
// And the archive is still an archive: both entries read back.
let reader = ZnippyArchive::open(&archive).unwrap();
assert_eq!(reader.extract_file("obj/a.bin").unwrap(), b"first".to_vec());
assert_eq!(reader.extract_file("obj/b.bin").unwrap(), b"second".to_vec());
let _ = std::fs::remove_dir_all(&dir);
}
/// **The writer, end to end: a superseded entry becomes a delta and still
/// reads back byte-exact.**
///
/// Two "generations" that share most of their bytes. After
/// `supersede_as_delta`, generation N is a delta chunk against N+1 and N+1 is
/// untouched — which is the direction that matters, because N+1 is the one a
/// clone copies entries out of.
#[test]
fn a_superseded_entry_becomes_a_delta_and_reads_back_exactly() {
let dir = unique_dir("supersede");
let archive = dir.join("a.znippy");
// Incompressible base bytes, so the saving measured is the DELTA's and
// not the codec's — the whole point of the exercise.
let mut st = 0x243f_6a88_85a3_08d3u64;
let gen_n: Vec<u8> = (0..400_000u32)
.map(|_| {
st = st.wrapping_add(0x9e37_79b9_7f4a_7c15);
let mut z = st;
z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
(z ^ (z >> 27)) as u8
})
.collect();
// N+1 keeps N's bytes and appends a little, as a later pack does.
let mut gen_n1 = gen_n.clone();
gen_n1.extend_from_slice(&gen_n[..20_000]);
create_archive(
&archive,
&[
("pack-N.pack".to_string(), gen_n.clone()),
("pack-N1.pack".to_string(), gen_n1.clone()),
],
3,
)
.unwrap();
let before = std::fs::metadata(&archive).unwrap().len();
let out = supersede_as_delta(&archive, "pack-N.pack", "pack-N1.pack", 0, 3).unwrap();
let (stored, delta_bytes) = match out {
SupersedeOutcome::Delta { stored_bytes, delta_bytes, chain_depth } => {
assert_eq!(chain_depth, 1);
(stored_bytes, delta_bytes)
}
other => panic!("expected a delta, got {other:?}"),
};
assert_eq!(stored, gen_n.len() as u64);
assert!(
delta_bytes * 20 < stored,
"N is a near-prefix of N+1, so the delta must be a small fraction of it; \
got {delta_bytes} against {stored}"
);
// THE ASSERTION THAT MATTERS: both entries still read back byte-exact,
// and the delta'd one through the delta path.
let ar = ZnippyArchive::open(&archive).unwrap();
assert_eq!(ar.extract_file("pack-N.pack").unwrap(), gen_n, "the superseded generation");
assert_eq!(ar.extract_file("pack-N1.pack").unwrap(), gen_n1, "the live generation");
assert_eq!(ar.extract_file_verified("pack-N.pack").unwrap(), gen_n);
assert_eq!(ar.file_size("pack-N.pack"), Some(gen_n.len() as u64));
// The live generation is still a WHOLE entry. If it were not, `P-001`'s
// copy path would be reading through a delta chain, which is the one
// outcome this direction exists to prevent.
assert!(
read_delta_map(&archive)
.unwrap()
.iter()
.all(|(p, _, _)| p == "pack-N.pack"),
"only the superseded generation may be a delta"
);
let _ = before;
let _ = std::fs::remove_dir_all(&dir);
}
/// The two refusals, both stated rather than silent.
#[test]
fn the_writer_refuses_a_pointless_delta_and_an_over_long_chain() {
let dir = unique_dir("supersede_refuse");
let archive = dir.join("a.znippy");
let mut st = 0x1234_5678_9abc_def0u64;
let noise = |n: usize, st: &mut u64| -> Vec<u8> {
(0..n)
.map(|_| {
*st = st.wrapping_add(0x9e37_79b9_7f4a_7c15);
let mut z = *st;
z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
(z ^ (z >> 27)) as u8
})
.collect()
};
let a = noise(60_000, &mut st);
let b = noise(60_000, &mut st); // unrelated: no delta can win
create_archive(
&archive,
&[("a.pack".to_string(), a.clone()), ("b.pack".to_string(), b.clone())],
3,
)
.unwrap();
match supersede_as_delta(&archive, "a.pack", "b.pack", 0, 3).unwrap() {
SupersedeOutcome::NotSmaller { .. } => {}
other => panic!("unrelated bytes must not produce a delta, got {other:?}"),
}
// Untouched, and still readable.
let ar = ZnippyArchive::open(&archive).unwrap();
assert_eq!(ar.extract_file("a.pack").unwrap(), a);
assert!(read_delta_map(&archive).unwrap().is_empty());
// And the chain bound, which is a WRITER policy: it declines rather than
// building a chain the reader would later refuse.
assert_eq!(
supersede_as_delta(&archive, "a.pack", "b.pack", MAX_GENERATION_CHAIN, 3).unwrap(),
SupersedeOutcome::ChainTooLong
);
assert!(supersede_as_delta(&archive, "a.pack", "a.pack", 0, 3).is_err());
let _ = std::fs::remove_dir_all(&dir);
}
/// **A chain of generations, and every one of them still exact.**
///
/// Four generations, each superseded against its successor as it arrives, so
/// the oldest sits at depth 3 and the newest is whole. This is the shape a
/// repository `gc`'d repeatedly produces, and the property is that reading
/// generation 0 walks three links and still returns the bytes that were
/// stored.
#[test]
fn a_generation_chain_reads_every_generation_back_exactly() {
let dir = unique_dir("genchain");
let archive = dir.join("g.znippy");
let mut st = 0x0f0f_0f0f_dead_beefu64;
let mut gens: Vec<Vec<u8>> = Vec::new();
let mut cur: Vec<u8> = (0..120_000u32)
.map(|_| {
st = st.wrapping_add(0x9e37_79b9_7f4a_7c15);
let mut z = st;
z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
(z ^ (z >> 27)) as u8
})
.collect();
gens.push(cur.clone());
for g in 1..4 {
cur.extend_from_slice(format!("generation {g} tail ").repeat(200).as_bytes());
gens.push(cur.clone());
}
let files: Vec<(String, Vec<u8>)> = gens
.iter()
.enumerate()
.map(|(i, b)| (format!("pack-{i}.pack"), b.clone()))
.collect();
create_archive(&archive, &files, 3).unwrap();
// Supersede each older generation against the one after it, oldest last,
// so each base is whole at the moment it is used.
for i in (0..3).rev() {
let out = supersede_as_delta(
&archive,
&format!("pack-{i}.pack"),
&format!("pack-{}.pack", i + 1),
3 - 1 - i,
3,
)
.unwrap();
assert!(matches!(out, SupersedeOutcome::Delta { .. }), "gen {i}: {out:?}");
}
let ar = ZnippyArchive::open(&archive).unwrap();
for (i, want) in gens.iter().enumerate() {
assert_eq!(
&ar.extract_file(&format!("pack-{i}.pack")).unwrap(),
want,
"generation {i} at chain depth {}",
3 - i
);
}
// The newest is whole: nothing in the map names it.
assert!(
read_delta_map(&archive)
.unwrap()
.iter()
.all(|(p, _, _)| p != "pack-3.pack"),
"the live generation must stay whole"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **What the writer costs and what it saves, on a REAL generation chain.**
///
/// `#[ignore]`d: it wants real `gc` generations on disk. `ZNIPPY_CHAIN_DIR`
/// names a directory of `pack-<i>.pack`, `i` ascending, oldest first.
#[test]
#[ignore]
fn perf_real_generation_chain() {
let Ok(d) = std::env::var("ZNIPPY_CHAIN_DIR") else { return };
let dir = unique_dir("realchain");
let archive = dir.join("g.znippy");
let mut files: Vec<(String, Vec<u8>)> = Vec::new();
for i in 0.. {
let p = std::path::Path::new(&d).join(format!("pack-{i}.pack"));
if !p.is_file() {
break;
}
files.push((format!("pack-{i}.pack"), std::fs::read(&p).unwrap()));
}
let n = files.len();
let raw: u64 = files.iter().map(|(_, b)| b.len() as u64).sum();
let t0 = std::time::Instant::now();
create_archive(&archive, &files, 3).unwrap();
let seal_ms = t0.elapsed().as_secs_f64() * 1e3;
let sealed = std::fs::metadata(&archive).unwrap().len();
// Supersede oldest-last, so each base is whole when it is used.
println!("gen,stored_b,delta_b,ratio_x,supersede_ms");
let mut encode_ms_total = 0.0;
let mut delta_total = 0u64;
for i in (0..n - 1).rev() {
let t = std::time::Instant::now();
let out = supersede_as_delta(
&archive,
&format!("pack-{i}.pack"),
&format!("pack-{}.pack", i + 1),
n - 2 - i,
3,
)
.unwrap();
let ms = t.elapsed().as_secs_f64() * 1e3;
encode_ms_total += ms;
match out {
SupersedeOutcome::Delta { stored_bytes, delta_bytes, .. } => {
delta_total += delta_bytes;
println!(
"{i},{stored_bytes},{delta_bytes},{:.2},{ms:.0}",
stored_bytes as f64 / delta_bytes as f64
);
}
other => println!("{i},-,-,-,{ms:.0} ({other:?})"),
}
}
// A rewrite reclaims the superseded blobs. Measured by rebuilding the
// archive from what it now holds, which is what a compaction would do.
let ar = ZnippyArchive::open(&archive).unwrap();
let mut live: Vec<(String, Vec<u8>)> = Vec::new();
let t = std::time::Instant::now();
for i in 0..n {
let name = format!("pack-{i}.pack");
let got = ar.extract_file(&name).unwrap();
assert_eq!(got, files[i].1, "generation {i} did not read back exactly");
live.push((name, got));
}
let read_all_ms = t.elapsed().as_secs_f64() * 1e3;
// Per-generation read cost, by chain depth.
println!("gen,depth,read_ms");
for i in 0..n {
let name = format!("pack-{i}.pack");
let t = std::time::Instant::now();
for _ in 0..5 {
let _ = ar.extract_file(&name).unwrap();
}
println!("{i},{},{:.2}", n - 1 - i, t.elapsed().as_secs_f64() * 1e3 / 5.0);
}
let compacted = dir.join("c.znippy");
let mut packed: Vec<(String, Vec<u8>)> = Vec::new();
for i in 0..n {
packed.push(live[i].clone());
}
// The compacted size is what the format would cost steady-state: the live
// generation whole plus one delta per superseded one, and nothing else.
let overhead = sealed - raw;
let steady = files[n - 1].1.len() as u64 + delta_total + overhead;
println!(
"SUMMARY generations={n} raw={raw} sealed={sealed} live_whole={} deltas={delta_total} \
steady_state={steady} saving_x={:.2} seal_ms={seal_ms:.0} \
supersede_ms_total={encode_ms_total:.0} read_all_ms={read_all_ms:.0}",
files[n - 1].1.len(),
raw as f64 / steady as f64
);
let _ = (compacted, packed);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn archive_reader_matches_free_functions() {
let dir = unique_dir("reader");
let archive = dir.join("store.znippy");
let files = synth(600, 13);
write_fresh(&archive, &files, ArrowIpcSink::new);
let reader = crate::ArchiveReader::open(&archive).unwrap();
assert_eq!(reader.row_count(), files.len(), "one chunk per synth file");
// Present files: cached locate == free-function locate.
for (p, _) in files.iter().step_by(37) {
let cached = reader.locate(p);
let free = crate::locate_file(&archive, p).unwrap();
assert!(!cached.is_empty(), "cached reader failed to locate {p}");
assert_eq!(cached, free, "cached vs free locate diverged for {p}");
}
// Absent file: both return empty.
let missing = "repo/does/not/exist.bin";
assert!(reader.locate(missing).is_empty());
assert!(crate::locate_file(&archive, missing).unwrap().is_empty());
// Whole-archive metadata parity.
assert_eq!(
reader.files_meta(),
crate::get_all_files_meta(&archive).unwrap(),
"cached files_meta diverged from get_all_files_meta"
);
// Prefix window parity (synth spreads paths across repo/grpNNN/).
for prefix in ["repo/grp001/", "repo/grp0", "repo/", ""] {
assert_eq!(
reader.files_meta_with_prefix(prefix),
crate::get_files_meta_with_prefix(&archive, prefix).unwrap(),
"cached prefix meta diverged for {prefix:?}"
);
}
let _ = std::fs::remove_dir_all(&dir);
}
/// **A supersede does not shrink the file, and this is what does.**
///
/// The first assertion is the fact the writer's own numbers do not state:
/// after `supersede_as_delta` the archive is no smaller — the superseded
/// blob is still in it, unreferenced. The second is that `compact_archive`
/// removes it and that every entry, delta chunks included, reads back exactly
/// the same bytes at the same depth afterwards.
///
/// Seen RED three ways, each restored:
///
/// * `sink.delta_map = Vec::new()` instead of carrying it — the compacted
/// archive hands back generation 0's delta instruction stream (11 bytes)
/// as its content and the byte comparison fires.
/// * copying `loc` unchanged instead of moving `blob_offset` — every entry
/// reads garbage from the wrong offset.
/// * skipping the rename — the size assertion fires, because the original
/// file is still the one on disk.
#[test]
fn compaction_reclaims_the_superseded_blob_and_changes_no_entry() {
let dir = unique_dir("compact");
let archive = dir.join("c.znippy");
let mut st = 0x5151_2323_abcd_ef01u64;
let base: Vec<u8> = (0..600_000u32)
.map(|_| {
st = st.wrapping_add(0x9e37_79b9_7f4a_7c15);
let mut z = st;
z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
(z ^ (z >> 27)) as u8
})
.collect();
let mut gens: Vec<Vec<u8>> = vec![base.clone()];
for g in 1..4 {
let mut next = gens[g - 1].clone();
next.extend_from_slice(format!("generation {g} tail ").repeat(300).as_bytes());
gens.push(next);
}
let files: Vec<(String, Vec<u8>)> = gens
.iter()
.enumerate()
.map(|(i, b)| (format!("pack-{i}.pack"), b.clone()))
.collect();
create_archive(&archive, &files, 3).unwrap();
let raw: u64 = gens.iter().map(|g| g.len() as u64).sum();
let whole = std::fs::metadata(&archive).unwrap().len();
for i in (0..3).rev() {
let out = supersede_as_delta(
&archive,
&format!("pack-{i}.pack"),
&format!("pack-{}.pack", i + 1),
3 - 1 - i,
3,
)
.unwrap();
assert!(matches!(out, SupersedeOutcome::Delta { .. }), "gen {i}: {out:?}");
}
// The finding, as an assertion. Three of four generations are now
// deltas of a few hundred bytes each, and the file is not smaller.
let after_supersede = std::fs::metadata(&archive).unwrap().len();
assert!(
after_supersede >= whole,
"supersede shrank the file ({whole} -> {after_supersede}); if that is now true, this \
test and everything built on the dead-payload finding wants revisiting"
);
let report = compact_archive(&archive).unwrap();
let compacted = std::fs::metadata(&archive).unwrap().len();
assert_eq!(report.bytes_after, compacted);
assert_eq!(report.rows, 4, "a compaction must not change the row count");
assert_eq!(report.delta_rows, 3, "the delta map must travel across it");
assert!(
compacted * 2 < raw,
"the compacted archive is {compacted} bytes for {raw} bytes of generations — the dead \
payload was not reclaimed"
);
// Every generation, at every depth, byte for byte.
let ar = ZnippyArchive::open(&archive).unwrap();
for (i, want) in gens.iter().enumerate() {
assert_eq!(
&ar.extract_file(&format!("pack-{i}.pack")).unwrap(),
want,
"generation {i} did not survive the compaction"
);
}
// The live generation is still whole, and the map still says so.
let map = read_delta_map(&archive).unwrap();
assert!(
map.iter().all(|(p, _, _)| p != "pack-3.pack"),
"compaction must not put the live generation behind a link: {map:?}"
);
// Idempotent: a second compaction of a compact archive changes nothing
// it should not, which is what a `gc` on a timer depends on.
let again = compact_archive(&archive).unwrap();
assert_eq!(again.rows, 4);
assert_eq!(again.delta_rows, 3);
let ar = ZnippyArchive::open(&archive).unwrap();
for (i, want) in gens.iter().enumerate() {
assert_eq!(&ar.extract_file(&format!("pack-{i}.pack")).unwrap(), want);
}
let _ = std::fs::remove_dir_all(&dir);
}
/// A compaction must carry the independent reserved logs too. gunnar's
/// `__gunnar_refs__` is the section this archive format has already lost once
/// (`25850cd`), and a rewrite is the same opportunity to lose it.
#[test]
fn compaction_carries_the_reserved_logs() {
use crate::index::GUNNAR_REFS_MODULE;
use crate::meta_sink::{ReservedSection, ReservedSectionBuilder};
let dir = unique_dir("compact_reserved");
let archive = dir.join("r.znippy");
let refs_bytes = b"refs/heads/main 0123456789abcdef".to_vec();
{
use std::os::unix::fs::FileExt;
let f = Arc::new(File::create(&archive).unwrap());
let payload = b"an entry".to_vec();
f.write_all_at(&payload, 0).unwrap();
let blobs = vec![BlobMeta {
blob_offset: 0,
blob_size: payload.len() as u64,
chunk_meta: ChunkMeta {
fdata_offset: 0,
file_index: 0,
chunk_seq: 0,
checksum: *blake3::hash(&payload).as_bytes(),
compressed: false,
uncompressed_size: payload.len() as u64,
compressed_size: payload.len() as u64,
},
}];
let batch =
crate::index::build_metadata_batch(&blobs, |_| "obj/a.bin".to_string(), &[], &[])
.unwrap();
let mut sink = ArrowIpcSink::new(Arc::clone(&f), payload.len() as u64);
let carried = refs_bytes.clone();
let builder: ReservedSectionBuilder =
Box::new(move |_lookup| Ok(vec![ReservedSection::raw(GUNNAR_REFS_MODULE, carried)]));
sink = sink.with_reserved_builder(builder);
sink.push_subindex(
crate::index::lookup_schema().as_ref(),
&[batch],
GroupKey {
pkg_type: 0,
repo: String::new(),
module_name: String::new(),
},
)
.unwrap();
Box::new(sink).finish().unwrap();
}
assert_eq!(
read_reserved_section_bytes(&archive, GUNNAR_REFS_MODULE).unwrap(),
Some(refs_bytes.clone())
);
compact_archive(&archive).unwrap();
assert_eq!(
read_reserved_section_bytes(&archive, GUNNAR_REFS_MODULE).unwrap(),
Some(refs_bytes),
"the compaction dropped __gunnar_refs__"
);
let _ = std::fs::remove_dir_all(&dir);
}
}
/// Decode an Arrow-IPC sub-index stream of the base schema into parallel
/// `(paths, locs)` vectors (mirrors `index::decode_lookup`, kept here so the
/// original stays untouched).
pub(crate) fn decode_base_rows(bytes: &[u8]) -> Result<(Vec<String>, Vec<ChunkLoc>)> {
let reader = crate::ipc_guard::stream_reader(bytes)
.map_err(|e| anyhow!("append: lookup ipc reader: {e}"))?;
let mut paths = Vec::new();
let mut locs = Vec::new();
for batch in reader {
let batch = batch.map_err(|e| anyhow!("append: lookup batch decode: {e}"))?;
let get = |n: &str| batch.column_by_name(n)
.ok_or_else(|| anyhow!("append: lookup missing column {n}"));
let p = get("relative_path")?.as_any().downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("relative_path type"))?;
let seq = get("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()
.ok_or_else(|| anyhow!("chunk_seq type"))?;
let fdata = get("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("fdata_offset type"))?;
let comp = get("compressed")?.as_any().downcast_ref::<BooleanArray>()
.ok_or_else(|| anyhow!("compressed type"))?;
let usz = get("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("uncompressed_size type"))?;
let boff = get("blob_offset")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("blob_offset type"))?;
let bsz = get("blob_size")?.as_any().downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("blob_size type"))?;
let ck = get("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()
.ok_or_else(|| anyhow!("checksum type"))?;
for i in 0..batch.num_rows() {
let mut c = [0u8; 32];
c.copy_from_slice(ck.value(i));
paths.push(p.value(i).to_string());
locs.push(ChunkLoc {
chunk_seq: seq.value(i),
fdata_offset: fdata.value(i),
blob_offset: boff.value(i),
blob_size: bsz.value(i),
uncompressed_size: usz.value(i),
compressed: comp.value(i),
checksum: c,
});
}
}
Ok((paths, locs))
}