znippy-common/src/archive.rs
//! ZnippyArchive — trait and implementation for reading znippy archives.
//!
//! Provides selective file extraction by path (serve individual artifacts
//! on demand from a single .znippy archive).
use std::any::Any;
use std::collections::HashMap;
use std::fs::File;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use anyhow::{anyhow, Result};
use arrow::record_batch::RecordBatch;
use arrow_array::{BooleanArray, FixedSizeBinaryArray, StringArray, UInt32Array, UInt64Array};
use crate::codec;
use crate::index::{read_reserved_section_bytes, read_znippy_index, ZNIPPY_DELTA_MODULE};
use crate::views::{
CondaSpec, CondaView, CoordSpec, CoordView, DebSpec, DebView, GemSpec, GemView, MavenSpec,
MavenView, NpmSpec, NpmView, PythonSpec, PythonView, RpmSpec, RpmView, RustSpec, RustView,
VIEW_SLOTS,
};
/// Trait for reading from a znippy archive.
pub trait ZnippyReader: Send + Sync {
fn list_files(&self) -> Result<Vec<String>>;
fn extract_file(&self, relative_path: &str) -> Result<Vec<u8>>;
fn contains(&self, relative_path: &str) -> bool;
fn file_size(&self, relative_path: &str) -> Option<u64>;
/// Batch extract several files, **on every core**.
///
/// The default used to be `paths.iter().map(|p| self.extract_file(p))` — a
/// serial loop under the honest comment "calls extract_file sequentially".
/// MEASURED on oden (32 cores, 2026-09-20) over a 544 MB / 37 824-file
/// vendored-crate archive, page-cache warm, all workers in one process:
///
/// ```text
/// workers 1 818 MB/s cores 1.00 ← what the serial map was
/// workers 2 1615 MB/s cores 1.96
/// workers 4 2981 MB/s cores 3.89
/// workers 8 5441 MB/s cores 7.60
/// workers 32 12724 MB/s cores 18.00
/// ```
///
/// One core busy and thirty-one idle is the shape ROOT LAW #0 names by
/// hand, and nothing about the work was ever serial: `extract_file` reads
/// through positioned `pread` on a shared `Arc<File>` (see
/// [`ReconstructCtx`]), takes `&self`, and the trait is already
/// `Send + Sync`, so the units share nothing mutable. The only thing
/// holding them to one core was the `map`.
///
/// [`gatling_for_each_balanced`][gfeb] is the engine `decompress.rs`
/// already drives this same archive format with. It is used here rather
/// than plain `gatling_for_each` for two properties this call needs:
///
/// * **heaviest-first (LPT)**, weighted by [`Self::file_size`] — a vendored
/// crate set is all skew (one 8 MB `.rlib` among ten thousand 2 kB `.rs`
/// files), and unweighted, one worker ends up holding the monster while
/// its siblings have drained the small end and idle;
/// * **results in the original `0..n` index order**, so this is
/// byte-for-byte what the serial `map` returned — same values, same
/// errors, same positions. `extract_files_is_the_serial_map` asserts it
/// over a deliberately skewed fixture with an absent path in the middle.
///
/// `batch = 8`: a claim is one atomic `fetch_add`, and the cheap tail of a
/// crate set is thousands of 2 kB files where a per-unit hand-off costs
/// more than the read. Eight amortizes it without coarsening the schedule
/// for the heavy head, which is claimed first anyway.
///
/// # What this did NOT fix, and the number that proves it
///
/// The ceiling above ~8 workers is **the allocator, not the engine**:
/// `extract_file` returns an owned `Vec<u8>` per file, so a 37 824-file
/// batch is 37 824 fresh allocations, most of them over glibc's 128 kB
/// `mmap` threshold — one `mmap`/`munmap` pair each, and `munmap` costs a
/// cross-core TLB shootdown that gets worse with every worker added.
/// Measured the same run with `MALLOC_MMAP_THRESHOLD_=1G`: 16 workers went
/// 4491 → 6963 MB/s and 64 workers 1587 → 4842 MB/s, purely from not
/// returning the pages. That is this crate's remaining "zero allocation —
/// preallocate and reuse" debt and it cannot be paid here: it needs an
/// `extract_file_into(&self, path, &mut Vec<u8>)` sibling so a caller can
/// hand the same buffer back, which is a new method on a published trait
/// and therefore the owner's call, not a cleanup.
///
/// An override is still free to do better — a storage model that can
/// coalesce adjacent blobs into one `pread` should.
///
/// [gfeb]: znippy_zoomies::gatling_forkjoin::gatling_for_each_balanced
fn extract_files(&self, paths: &[&str]) -> Vec<Result<Vec<u8>>> {
znippy_zoomies::gatling_forkjoin::gatling_for_each_balanced(
paths.len(),
/* n_workers: 0 = one per core */ 0,
/* batch */ 8,
// The weight is the file's declared uncompressed size. A path the
// index does not hold weighs 0: it is about to be a cheap `Err`,
// and scheduling it last is correct.
|i| self.file_size(paths[i]).unwrap_or(0),
|i| self.extract_file(paths[i]),
)
}
}
/// What an entry reconstructs *from*: the shared archive fd, the real file
/// length every bounds check is made against, and — for storage models whose
/// entries reference other entries — a way to resolve one.
///
/// Passed by reference into [`Entry::reconstruct`] so that the storage model
/// owns the reconstruction and [`ZnippyArchive`] owns only the file.
pub struct ReconstructCtx<'a> {
archive: &'a Arc<File>,
/// Real byte length of the archive, cached at open. The index is untrusted,
/// so every `blob_offset + blob_size` is checked against this BEFORE any
/// allocation.
archive_len: u64,
/// Resolve another entry's bytes by path. `None` means "no resolver
/// available", which is what a chunked entry is given because it never
/// needs one. A delta chunk resolves its base through this.
resolve: Option<&'a dyn BaseResolve>,
/// How many base resolutions deep this reconstruction already is.
///
/// **The reader's own bound, and it is not the writer's.**
/// [`MAX_DELTA_CHAIN`] is a policy a *writer* applies to its own output; it
/// binds nothing about an index that arrives from elsewhere. A hostile or
/// corrupt index can name a base that names it back, and without this the
/// read recurses until the stack ends — a crash, not an `Err`. The per-entry
/// delta model had exactly this hole; it is closed here because the
/// recursion is now in one place.
depth: usize,
}
impl<'a> ReconstructCtx<'a> {
/// The archive fd, read via positioned I/O so reconstruction is safe to run
/// concurrently from many threads.
pub fn archive(&self) -> &Arc<File> {
self.archive
}
/// Real byte length of the archive file.
pub fn archive_len(&self) -> u64 {
self.archive_len
}
/// Resolve another entry's bytes, for a chunk that references one. `Err`
/// when no resolver was supplied, and `Err` — never a stack overflow — when
/// the chain is longer than [`MAX_RECONSTRUCT_DEPTH`].
pub fn resolve_base(&self, path: &str, verify: bool) -> Result<Vec<u8>> {
if self.depth >= MAX_RECONSTRUCT_DEPTH {
return Err(anyhow!(
"base chain for {} is deeper than {} — refusing to recurse further \
(a cyclic or over-long index must be an error, not a crash)",
path,
MAX_RECONSTRUCT_DEPTH
));
}
self.resolve
.ok_or_else(|| anyhow!("entry needs a base but this archive supplied no resolver"))?
.resolve(path, verify, self.depth + 1)
}
}
/// The reader's hard bound on how many base resolutions one `extract_file` may
/// perform, whatever the index claims.
///
/// **64**, deliberately far above the writer's [`MAX_DELTA_CHAIN`] of 8: this is
/// not a policy about what is worth storing, it is the line between an `Err` and
/// a stack overflow on an index this process did not write. A legitimate archive
/// never comes near it; a cycle hits it on the 64th link.
pub const MAX_RECONSTRUCT_DEPTH: usize = 64;
/// Resolve one entry's fully-reconstructed bytes by path.
///
/// Implemented by [`ZnippyArchive`]. Kept a separate trait rather than a method
/// on the archive so that a storage model depends on the *capability* and not on
/// the concrete archive type.
pub trait BaseResolve: Send + Sync {
/// `depth` is how many resolutions deep the caller already is. It is passed
/// rather than tracked by the resolver because a resolver is shared across
/// concurrent reads and a counter on it would be a shared mutable — the
/// depth belongs to one reconstruction, not to the archive.
fn resolve(&self, path: &str, verify: bool, depth: usize) -> Result<Vec<u8>>;
}
/// **Where one chunk's bytes come from.**
///
/// A chunk has always meant "bytes at this offset in the blob region". That is
/// [`ChunkSource::Stored`].
///
/// [`ChunkSource::Delta`] means "bytes from applying this diff to that entry".
/// The recursion lives in the CHUNK rather than in the entry, so reconstruction
/// stays one loop and a single file can be **part chunk and part delta**. A
/// per-entry seam cannot express that; see the note on [`Entry`].
enum ChunkSource {
/// The stored blob **is** the chunk's bytes (after decompression, if
/// `compressed`). Every znippy archive written before this existed holds
/// only these.
Stored,
/// The stored blob is a delta instruction stream (the format
/// [`apply_delta`] reads). The chunk's bytes are the result of applying it
/// to `base_path`'s fully reconstructed bytes, resolved through
/// [`BaseResolve`].
///
/// The base is named by path, so it may itself be chunked, delta'd, or a
/// mixture; this chunk never inspects it.
Delta { base_path: String },
}
struct ChunkInfo {
blob_offset: u64,
blob_size: u64,
fdata_offset: u64,
compressed: bool,
/// The index row's `uncompressed_size`: what this chunk's stored blob
/// decompresses to. Untrusted like every index value, but it is what a
/// compressed chunk's OpenZL frame is held to: a frame whose header claims
/// more is refused before its output buffer exists (see [`frame_limit`]).
/// Only read for a compressed chunk.
uncompressed_size: u64,
/// **blake3 over the bytes this chunk CONTRIBUTES to the file** (Design
/// law 3). All-zero if the index predates the checksum column (older
/// archives), in which case verification is skipped.
///
/// For [`ChunkSource::Stored`] those are the stored bytes after
/// decompression, which is exactly what this column has always meant — the
/// constant `bc191568…` pins that it is unchanged.
///
/// For [`ChunkSource::Delta`] those are the bytes the delta PRODUCES, not
/// the instruction stream that produces them. That distinction is the whole
/// integrity argument, and it is the same one the per-entry delta model made
/// with a separate `result_checksum`: a chain of individually-correct deltas
/// can still reconstruct wrong bytes, because the result also depends on the
/// base and **no delta row says what the base should have contained**.
/// Hashing the output closes that, and doing it per chunk rather than per
/// entry means a mixed entry is covered too, which `result_checksum` could
/// not be.
///
/// The cost is that a delta chunk's checksum must be checked on the **fast**
/// read as well — see [`Entry::reconstruct`] — where a stored chunk's is
/// checked only under `verify`. That asymmetry is deliberate: it buys the
/// hot artifact-serving path nothing to re-hash bytes that came straight off
/// disk, and it is the only thing standing between a delta chunk and silent
/// corruption.
checksum: [u8; 32],
/// The index row's `chunk_seq`. Carried only so the delta map
/// ([`ZNIPPY_DELTA_MODULE`]) can name one chunk of one entry — nothing in
/// reconstruction uses it, which places the chunks by `fdata_offset`.
chunk_seq: u32,
/// Which kind of stored unit this is. See [`ChunkSource`].
source: ChunkSource,
}
/// **One archive entry: a tiling of chunks, each placed at its own
/// `fdata_offset`.**
///
/// # Why this is a struct and not a trait
///
/// It *was* a trait — `EntryReader`, with two implementations that differed in
/// how they COMBINE: chunks concatenate, deltas apply. That difference is what
/// forced two of them, and it put the seam on the read side.
///
/// The seam belongs on the **write** side instead. A splitter decides how to cut
/// a file into stored units (see [`Splitter`]); a unit it may emit is a reference
/// to another entry. Reconstruction then stays exactly one loop — the loop below,
/// which is the pre-trait `extract_inner` code — and the recursion lives in
/// [`ChunkSource`].
///
/// What that buys, and it is the reason for the swap: **a single file can be part
/// chunk and part delta.** A large file with one changed region stores the changed
/// region as a delta and the rest as ordinary chunks. `EntryReader` was
/// all-or-nothing per entry and could not say that.
///
/// What it leaves alone: the read path, which is the half that is measured and
/// proven. `chunked_reconstruction_is_byte_identical_to_the_pre_trait_implementation`
/// pins that with one blake3 over a deterministic corpus, and that constant has
/// now survived both the trait's arrival and its removal.
pub struct Entry {
uncompressed_size: u64,
chunks: Vec<ChunkInfo>,
}
impl Entry {
/// Size of the reconstructed entry in bytes, as the index declares it.
/// Untrusted: never size an allocation from it.
pub fn uncompressed_size(&self) -> u64 {
self.uncompressed_size
}
/// Reconstruct the entry's bytes. `path` is carried for error messages only.
/// With `verify`, each chunk's bytes are blake3-checked against what the
/// index recorded for it (Design law 3).
pub fn reconstruct(
&self,
path: &str,
ctx: &ReconstructCtx<'_>,
verify: bool,
) -> Result<Vec<u8>> {
// Never pre-sized from `uncompressed_size` ALONE: that value comes verbatim
// from the untrusted index, so a `Vec::with_capacity` on it is a
// malformed-index-driven allocation that aborts the process rather than
// returning an `Err` (DoS). Bounded by the archive's REAL length it is safe
// (no index can make this reserve more than the file the reader opened), and
// it is exact for every stored entry: grown by doubling instead, a 64 MiB
// crate allocated 231 MiB and peaked at 117 MiB live (holger HOL-PERF-3).
let mut result: Vec<u8> = Vec::with_capacity(self.uncompressed_size.min(ctx.archive_len()) as usize);
let mut blob = Vec::new(); // reused across chunks
let mut decomp = Vec::new(); // reused across compressed chunks
// Bases resolved so far in THIS entry, keyed by path.
//
// This is the one cost delta-as-a-chunk introduces that the per-entry
// model did not have: a mixed entry with K delta chunks against the same
// base would resolve that base K times, where the per-entry model
// resolved it once. One local map removes it, and it is local on purpose
// — a cache living on the archive would be shared mutable state across
// concurrent reads, which is what `extract_file`'s thread-safety rests on
// not having.
//
// It does NOT flatten a CHAIN: A→B→C still resolves B, which resolves C.
// That recursion is inherent and costs what the per-entry model cost.
let mut bases: HashMap<String, Vec<u8>> = HashMap::new();
for chunk in &self.chunks {
// Bounds-check the untrusted (index-declared) blob extent against the
// real file length BEFORE allocating — a malformed index must not be
// able to drive a multi-GB zero-fill (DoS). Matches get_file.
if chunk.blob_size > 0 {
let in_bounds = chunk
.blob_offset
.checked_add(chunk.blob_size)
.is_some_and(|end| end <= ctx.archive_len());
if !in_bounds {
return Err(anyhow!(
"blob for {} out of bounds (offset={}, size={}, archive_len={})",
path,
chunk.blob_offset,
chunk.blob_size,
ctx.archive_len()
));
}
}
// A stored, uncompressed chunk that continues the file is read straight
// into the result: no scratch copy (holger HOL-PERF-3). The same checks
// as below: its blake3 when verifying, its place at `fdata_offset`.
if !chunk.compressed && matches!(chunk.source, ChunkSource::Stored) && chunk.fdata_offset as usize == result.len() {
let start = result.len();
result.resize(start + chunk.blob_size as usize, 0);
ctx.archive().read_exact_at(&mut result[start..], chunk.blob_offset)?;
if verify && chunk.checksum != [0u8; 32] && blake3::hash(&result[start..]).as_bytes()[..] != chunk.checksum[..] {
return Err(anyhow!("checksum mismatch for {} at fdata_offset {}", path, chunk.fdata_offset));
}
continue;
}
blob.resize(chunk.blob_size as usize, 0);
// Positioned read — no shared seek, safe under concurrent calls.
ctx.archive().read_exact_at(&mut blob, chunk.blob_offset)?;
// What the stored blob MEANS is the chunk's business, not the
// entry's. The loop around this match is the pre-trait
// reconstruction code and does not know a delta exists.
//
// The uncompressed stored bytes first — the instruction stream for a
// delta chunk, the payload itself for a stored one.
let raw: &[u8] = if chunk.compressed {
// A verified read is the untrusted one (Holger's bundle verifier):
// OpenZL's own allocations are budgeted from the declared size too.
// The fast read is the trusted, hot one and pays nothing extra.
if verify {
codec::decompress_into_budgeted(&blob, &mut decomp, frame_limit(chunk))
} else {
codec::decompress_into_max(&blob, &mut decomp, frame_limit(chunk))
}
.map_err(|e| anyhow!("{path} at fdata_offset {}: {e}", chunk.fdata_offset))?;
&decomp
} else {
&blob
};
// Then what those bytes MEAN. `applied` only exists on the delta arm
// so a stored chunk still copies nothing extra.
let applied: Vec<u8>;
let (bytes, must_check): (&[u8], bool) = match &chunk.source {
ChunkSource::Stored => (raw, false),
ChunkSource::Delta { base_path } => {
// Resolved through the same dispatch, so the base may be
// chunked, delta'd or mixed. `verify` propagates: a verified
// read of a delta chunk is only meaningful if its base was
// verified too.
if !bases.contains_key(base_path) {
let b = ctx.resolve_base(base_path, verify)?;
bases.insert(base_path.clone(), b);
}
let base = &bases[base_path];
applied = apply_delta(base, raw)?;
// TRUE, not `verify`. See `ChunkInfo::checksum`: this is the
// only thing that can tell a correct delta applied to the
// WRONG base from a correct one, and the per-entry model
// checked its equivalent on the fast read for the same
// reason.
(&applied, true)
}
};
if (verify || must_check) && chunk.checksum != [0u8; 32] {
let computed = blake3::hash(bytes);
if computed.as_bytes()[..] != chunk.checksum[..] {
// One hash over the OUTPUT catches strictly more than a hash
// over a delta's instruction stream would: a corrupt stream
// and a wrong base both land here. What it cannot do is say
// WHICH, so the delta arm names the base it used — that is
// the diagnostic the per-row hash used to give.
return match &chunk.source {
ChunkSource::Stored => Err(anyhow!(
"checksum mismatch for {} at fdata_offset {}",
path,
chunk.fdata_offset
)),
ChunkSource::Delta { base_path } => Err(anyhow!(
"delta chunk of {} at fdata_offset {} produced bytes that do not \
match its result checksum (base {})",
path,
chunk.fdata_offset,
base_path
)),
};
}
}
// Place each chunk at its declared `fdata_offset` (as `reassemble_file`
// does) instead of blindly concatenating. Concatenation meant that two
// index rows for the same path — both at fdata_offset 0 — produced a
// buffer of twice the real length holding both copies back to back,
// with no error and even `extract_file_verified` passing, since each
// chunk's blake3 is individually correct. `start <= result.len()` also
// caps the buffer at real verified bytes, so an invented offset cannot
// drive the allocation.
let start = chunk.fdata_offset as usize;
if start > result.len() {
return Err(anyhow!(
"chunk of {} leaves a gap at fdata_offset {} (file reaches {})",
path,
start,
result.len()
));
}
let end = start + bytes.len();
if end > result.len() {
result.resize(end, 0);
}
result[start..end].copy_from_slice(bytes);
}
Ok(result)
}
}
/// Slack a delta chunk's instruction stream may have over the bytes it
/// produces: the two size varints, and one opcode byte per 127 inserted bytes.
const DELTA_FRAME_SLACK: u64 = 64 * 1024;
/// **The most bytes a compressed chunk's OpenZL frame may decode to.**
///
/// The frame header's decompressed size used to size the output buffer on its
/// own, so a few KiB of frame behind an index row declaring 12 bytes bought a
/// zero-filled buffer of whatever the header said (lane gatwasm, 2026-10-31).
/// A stored chunk's frame decodes to exactly the row's `uncompressed_size`; a
/// delta chunk's frame holds the instruction stream, which may run a little
/// over the bytes it produces. The check is one comparison per chunk, on the
/// fast and the verified read alike.
///
/// OpenZL's internal stream buffers, which it sizes from the frame's graph, are
/// bounded only on the verified read, by `codec::decompress_into_budgeted`
/// (see `tests/openzl_internal_alloc.rs`).
fn frame_limit(chunk: &ChunkInfo) -> usize {
let limit = match chunk.source {
ChunkSource::Stored => chunk.uncompressed_size,
ChunkSource::Delta { .. } => chunk
.uncompressed_size
.saturating_add(chunk.uncompressed_size / 64)
.saturating_add(DELTA_FRAME_SLACK),
};
usize::try_from(limit).unwrap_or(usize::MAX)
}
/// Read a git-style delta varint (7 bits per byte, little-endian, high bit
/// continues). Returns the value and how many bytes it consumed.
fn delta_varint(buf: &[u8], at: &mut usize) -> Result<u64> {
let mut value: u64 = 0;
let mut shift = 0u32;
loop {
let byte = *buf
.get(*at)
.ok_or_else(|| anyhow!("delta header truncated at byte {}", at))?;
*at += 1;
if shift >= 64 {
return Err(anyhow!("delta size varint overflows u64"));
}
value |= u64::from(byte & 0x7f) << shift;
shift += 7;
if byte & 0x80 == 0 {
return Ok(value);
}
}
}
/// Apply one delta instruction stream to `base`.
///
/// The stream is git's delta encoding, chosen because it is fully specified,
/// compact, and the corpus that motivated this is git objects — but nothing here
/// is git-aware: it is a byte-level copy/insert format over an opaque base.
///
/// Layout: base-size varint, result-size varint, then instructions.
/// * high bit set → COPY, the low 7 bits select which of 4 offset and 3 size
/// bytes follow; a zero size means 0x10000.
/// * high bit clear, non-zero → INSERT that many literal bytes.
/// * `0x00` is not a valid instruction and is rejected rather than skipped.
///
/// Every offset and length is bounds-checked against the real base and the
/// declared result size before use, so a corrupt or hostile delta yields `Err`
/// and never an over-large allocation or an out-of-range read.
///
/// `pub` because it is the decoder half of the format whose encoder
/// ([`encode_delta_against`]) is already public. A public encoder with a private
/// decoder is an asymmetry, not an encapsulation.
pub fn apply_delta(base: &[u8], delta: &[u8]) -> Result<Vec<u8>> {
let mut at = 0usize;
let declared_base = delta_varint(delta, &mut at)?;
if declared_base != base.len() as u64 {
return Err(anyhow!(
"delta expects a base of {} bytes, resolved base is {}",
declared_base,
base.len()
));
}
let result_size = delta_varint(delta, &mut at)?;
// Cap the declared result against what the instructions could possibly
// produce, so `result_size` alone can never drive an allocation.
let mut out: Vec<u8> = Vec::new();
while at < delta.len() {
let op = delta[at];
at += 1;
if op & 0x80 != 0 {
let mut copy_off: u64 = 0;
let mut copy_len: u64 = 0;
for i in 0..4 {
if op & (1 << i) != 0 {
let b = *delta
.get(at)
.ok_or_else(|| anyhow!("delta copy offset truncated"))?;
at += 1;
copy_off |= u64::from(b) << (8 * i);
}
}
for i in 0..3 {
if op & (0x10 << i) != 0 {
let b = *delta
.get(at)
.ok_or_else(|| anyhow!("delta copy size truncated"))?;
at += 1;
copy_len |= u64::from(b) << (8 * i);
}
}
if copy_len == 0 {
copy_len = 0x1_0000;
}
let end = copy_off
.checked_add(copy_len)
.ok_or_else(|| anyhow!("delta copy range overflows"))?;
if end > base.len() as u64 {
return Err(anyhow!(
"delta copies [{}, {}) from a base of {} bytes",
copy_off,
end,
base.len()
));
}
out.extend_from_slice(&base[copy_off as usize..end as usize]);
} else if op != 0 {
let n = op as usize;
let end = at
.checked_add(n)
.ok_or_else(|| anyhow!("delta insert range overflows"))?;
if end > delta.len() {
return Err(anyhow!("delta insert of {} bytes runs past the stream", n));
}
out.extend_from_slice(&delta[at..end]);
at = end;
} else {
// 0x00 is unassigned. Skipping it silently is how a corrupt stream
// turns into plausible-looking output.
return Err(anyhow!("delta contains a 0x00 instruction"));
}
if out.len() as u64 > result_size {
return Err(anyhow!(
"delta produced {} bytes, more than the declared {}",
out.len(),
result_size
));
}
}
if out.len() as u64 != result_size {
return Err(anyhow!(
"delta produced {} bytes, declared {}",
out.len(),
result_size
));
}
Ok(out)
}
/// Maximum number of deltas in one chain before the writer refuses and stores
/// the version chunked instead.
///
/// **8.** This started at git's `pack.depth` of 50 and was tightened on review.
/// The two cases are not alike and the difference is the read path, not the
/// format: git resolves a delta chain inside one mmap'd packfile with its own
/// delta-base cache, whereas a znippy chain costs **one full entry
/// reconstruction per link** — a `pread` plus a codec frame decode each — and
/// reconstruction recurses through [`BaseResolve`] with no cache between links.
/// At depth 50 a single `extract_file` is up to 50 decompressions.
///
/// 8 bounds that at a cost the archival win absorbs easily. **I have no measured
/// evidence that a deeper cap is safe for this read path, so the tighter number
/// is taken rather than argued against.** Raising it needs a reconstruction-
/// latency measurement against chain depth, which does not exist yet.
pub const MAX_DELTA_CHAIN: usize = 8;
/// A delta must be smaller than this fraction of the base it is against, or the
/// version is stored chunked.
///
/// 0.7. A delta that saves less than 30% is not worth a chain link: it still
/// costs a whole extra reconstruction on every read of every later version, and
/// that cost is paid forever while the saving is paid once.
pub const DELTA_SIZE_ALPHA: f64 = 0.7;
/// Minimum window a copy must span to be worth emitting. Below this the copy
/// header costs more than the literal bytes it replaces.
const MIN_COPY: usize = 16;
/// **The encoder half of the delta format [`apply_delta`] reads.**
///
/// # Why this lives here and is not `gunnar-delta`
///
/// LAW 5 says reuse, do not twin, and both alternatives were checked before
/// writing this rather than after:
///
/// * **zstd `--patch-from` is not reachable.** znippy's codec exposes only
/// whole-buffer `compress`/`decompress`, and `openzl-sys-rs` (0.3.0, and still 0.4.0) —
/// the published crate that vendors the C sources — binds **no** `ZSTD_*`
/// symbol at all and no dictionary, prefix or patch-from entry point
/// (measured: `grep -c ZSTD_ src/bindings.rs` = 0). Taking that route means
/// adding the `zstd` crate, i.e. a second C dependency, which is the cost
/// `cold-tier-decision.md` spent a document avoiding.
/// * **`gunnar-delta` cannot be depended on from here.** It is unpublished and
/// lives in gunnar's tree; znippy depending on gunnar inverts the direction
/// the whole layering runs in.
///
/// So this is a second implementation of **the same wire format**, not a third
/// format. That is stated plainly rather than hidden: `gunnar-delta` remains the
/// tuned encoder on gunnar's side, this is the minimal one on znippy's, and
/// because the bytes are identical either side can read the other's output. The
/// round-trip test asserts exactly that inverse property against
/// [`apply_delta`], which is the decoder already in this file.
///
/// The match finder is a 16-byte-anchored hash index over the base — enough to
/// catch the "same file, edited" case historization is for, and deliberately not
/// a competitor to a tuned window search.
pub fn encode_delta_against(base: &[u8], target: &[u8]) -> Vec<u8> {
let mut out = Vec::new();
put_size_varint(&mut out, base.len() as u64);
put_size_varint(&mut out, target.len() as u64);
// Hash every MIN_COPY-aligned anchor in the base.
let mut index: HashMap<u64, Vec<usize>> = HashMap::new();
if base.len() >= MIN_COPY {
let mut i = 0usize;
while i + MIN_COPY <= base.len() {
index.entry(hash16(&base[i..i + MIN_COPY])).or_default().push(i);
i += MIN_COPY;
}
}
let mut literal_start = 0usize;
let mut at = 0usize;
while at < target.len() {
let mut best = (0usize, 0usize); // (base_off, len)
if at + MIN_COPY <= target.len() {
if let Some(cands) = index.get(&hash16(&target[at..at + MIN_COPY])) {
// Bounded candidate scan: historization sees few collisions and
// an unbounded scan is how an encoder becomes quadratic.
for &bo in cands.iter().take(8) {
if base.len() - bo < MIN_COPY || &base[bo..bo + MIN_COPY] != &target[at..at + MIN_COPY] {
continue;
}
let mut n = MIN_COPY;
while bo + n < base.len() && at + n < target.len() && base[bo + n] == target[at + n] {
n += 1;
}
if n > best.1 {
best = (bo, n);
}
}
}
}
if best.1 >= MIN_COPY {
flush_literal(&mut out, &target[literal_start..at]);
emit_copy(&mut out, best.0 as u64, best.1 as u64);
at += best.1;
literal_start = at;
} else {
at += 1;
}
}
flush_literal(&mut out, &target[literal_start..]);
out
}
fn hash16(b: &[u8]) -> u64 {
// FNV-1a over the anchor. Cheap and good enough to bucket candidates.
let mut h: u64 = 0xcbf2_9ce4_8422_2325;
for &x in b {
h ^= x as u64;
h = h.wrapping_mul(0x1000_0000_01b3);
}
h
}
fn put_size_varint(out: &mut Vec<u8>, mut v: u64) {
loop {
let mut b = (v & 0x7f) as u8;
v >>= 7;
if v != 0 {
b |= 0x80;
}
out.push(b);
if v == 0 {
return;
}
}
}
/// INSERT runs are capped at 0x7f, the largest an insert opcode can name.
fn flush_literal(out: &mut Vec<u8>, lit: &[u8]) {
for piece in lit.chunks(0x7f) {
out.push(piece.len() as u8);
out.extend_from_slice(piece);
}
}
/// Emit COPY ops, splitting at 0xffffff which is the largest size the three
/// size bytes can name.
fn emit_copy(out: &mut Vec<u8>, mut off: u64, mut len: u64) {
while len > 0 {
let take = len.min(0xff_ffff);
let mut op: u8 = 0x80;
let mut tail = Vec::new();
for i in 0..4 {
let b = ((off >> (8 * i)) & 0xff) as u8;
if b != 0 {
op |= 1 << i;
tail.push(b);
}
}
// Sizes are emitted in ascending byte order, matching the decoder.
let mut size_bytes = Vec::new();
for i in 0..3 {
let b = ((take >> (8 * i)) & 0xff) as u8;
if b != 0 {
op |= 0x10 << i;
size_bytes.push(b);
}
}
out.push(op);
out.extend_from_slice(&tail);
out.extend_from_slice(&size_bytes);
off += take;
len -= take;
}
}
/// What the writer decided to do with one version of a path.
///
/// The decision is made on **real bytes** — the delta is encoded and its
/// compressed size compared against the compressed chunked form — never on a
/// guess about how similar two versions look. A reader is therefore never
/// surprised: whichever branch wins is what the index records.
#[derive(Debug, PartialEq, Eq)]
pub enum VersionPlan {
/// Store as a delta against `base_path`, `chain_len` links deep.
Delta { base_path: String, chain_len: usize, delta_bytes: usize },
/// Store the whole version as chunks. Carries why, so the choice is
/// auditable rather than silent.
Chunked(ChunkedReason),
}
/// Why a version was not delta'd.
#[derive(Debug, PartialEq, Eq)]
pub enum ChunkedReason {
/// No previous version of this path in the archive — this IS the first row.
FirstVersion,
/// The chain would exceed [`MAX_DELTA_CHAIN`].
ChainTooLong,
/// The delta was not smaller than storing the bytes outright, or did not
/// clear [`DELTA_SIZE_ALPHA`] against the base.
DeltaNotSmaller,
}
/// Decide how to store `target` for `path`, given the previous version's bytes
/// and how deep its chain already is.
///
/// `compressed_len` is how the caller measures a candidate payload — the real
/// codec, so the comparison is between what would actually be written in each
/// branch and not between raw sizes.
pub fn plan_version(
path: &str,
previous: Option<(&str, &[u8], usize)>,
target: &[u8],
mut compressed_len: impl FnMut(&[u8]) -> usize,
) -> (VersionPlan, Option<Vec<u8>>) {
let (base_path, base_bytes, base_chain) = match previous {
None => return (VersionPlan::Chunked(ChunkedReason::FirstVersion), None),
Some(p) => p,
};
let _ = path;
if base_chain + 1 > MAX_DELTA_CHAIN {
return (VersionPlan::Chunked(ChunkedReason::ChainTooLong), None);
}
let delta = encode_delta_against(base_bytes, target);
// Two independent cutoffs, both on real bytes.
//
// 1. Against the BASE, the alpha rule: a delta that does not save at least
// (1 - alpha) of the base is not worth the permanent read cost of an
// extra chain link.
// 2. Against the TARGET, measured through the codec: a delta that compresses
// worse than simply storing the version is never a win, however well it
// scores on (1). Raw size alone would miss this.
if (delta.len() as f64) >= DELTA_SIZE_ALPHA * (base_bytes.len() as f64) {
return (VersionPlan::Chunked(ChunkedReason::DeltaNotSmaller), None);
}
if compressed_len(&delta) >= compressed_len(target) {
return (VersionPlan::Chunked(ChunkedReason::DeltaNotSmaller), None);
}
let n = delta.len();
(
VersionPlan::Delta { base_path: base_path.to_string(), chain_len: base_chain + 1, delta_bytes: n },
Some(delta),
)
}
// ───────────────────────────────────────────────────────────────────────────
// The seam, on the write side
// ───────────────────────────────────────────────────────────────────────────
/// **One stored unit: what a splitter emits, and what becomes one index row.**
///
/// `fdata_offset` is where this unit's bytes belong inside the reconstructed
/// file, which is what makes a MIXED entry expressible: the units of one entry
/// need not all be of the same kind, they only need to tile the file.
pub struct StoredUnit {
/// Offset of this unit's bytes within the reconstructed file.
pub fdata_offset: u64,
pub payload: UnitPayload,
}
/// What a stored unit holds. The two arms correspond exactly to the two
/// [`ChunkSource`] arms the reader knows about — this is the write side of the
/// same seam, which is the whole point of moving it here.
pub enum UnitPayload {
/// Store these bytes as an ordinary chunk.
Bytes(Vec<u8>),
/// Store this diff; the reader applies it to `base_path`'s bytes.
///
/// `expected` is the bytes the diff must produce. It is not stored as such —
/// its blake3 is, as the unit's checksum — and that is the property that
/// pins a delta chain's answer (see [`ChunkSource`]).
Delta {
base_path: String,
delta: Vec<u8>,
expected: Vec<u8>,
},
}
/// **How a file is cut into stored units.**
///
/// This is the seam. It was on the read side (`EntryReader`, one implementation
/// per storage model, dispatched per entry) and it is here now, because the
/// difference between the models is a *writing* decision: which units to emit.
/// The read side then has one loop and no trait at all.
///
/// A splitter may emit ordinary byte units, delta units, or **both for the same
/// file** — which is the capability the per-entry seam could not express and the
/// reason for the swap.
pub trait Splitter: Send + Sync {
/// Cut `bytes` into the units to store for `path`.
///
/// `previous` is the last stored version of this path, if any: its path, its
/// reconstructed bytes, and how deep its chain already is. A splitter that
/// never emits deltas ignores it.
fn split(
&self,
path: &str,
bytes: &[u8],
previous: Option<(&str, &[u8], usize)>,
compressed_len: &mut dyn FnMut(&[u8]) -> usize,
) -> Vec<StoredUnit>;
}
/// The splitter every znippy archive written to date was produced by: a fixed
/// tiling of `chunk_size` byte units, no deltas.
///
/// Kept as a named type rather than left implicit so that "what znippy does
/// today" is a value one can pass, compare against and test.
pub struct ChunkSplitter {
pub chunk_size: usize,
}
impl Splitter for ChunkSplitter {
fn split(
&self,
_path: &str,
bytes: &[u8],
_previous: Option<(&str, &[u8], usize)>,
_compressed_len: &mut dyn FnMut(&[u8]) -> usize,
) -> Vec<StoredUnit> {
if bytes.is_empty() {
return vec![StoredUnit { fdata_offset: 0, payload: UnitPayload::Bytes(Vec::new()) }];
}
bytes
.chunks(self.chunk_size.max(1))
.enumerate()
.map(|(i, piece)| StoredUnit {
fdata_offset: (i * self.chunk_size.max(1)) as u64,
payload: UnitPayload::Bytes(piece.to_vec()),
})
.collect()
}
}
/// The whole-entry delta splitter: exactly the decision [`plan_version`] makes,
/// expressed as units.
///
/// Emits either **one** delta unit covering the whole file, or the fallback
/// splitter's byte units. This is what the per-entry `EntryReader` model could
/// express, and it is here to show that the swap loses nothing: the same
/// archives are still writable.
pub struct DeltaSplitter {
pub fallback: ChunkSplitter,
}
impl Splitter for DeltaSplitter {
fn split(
&self,
path: &str,
bytes: &[u8],
previous: Option<(&str, &[u8], usize)>,
compressed_len: &mut dyn FnMut(&[u8]) -> usize,
) -> Vec<StoredUnit> {
let (plan, delta) = plan_version(path, previous, bytes, |b| compressed_len(b));
match (plan, delta) {
(VersionPlan::Delta { base_path, .. }, Some(delta)) => vec![StoredUnit {
fdata_offset: 0,
payload: UnitPayload::Delta { base_path, delta, expected: bytes.to_vec() },
}],
_ => self.fallback.split(path, bytes, previous, compressed_len),
}
}
}
/// **The splitter the per-entry seam could not have: one file, part chunk and
/// part delta.**
///
/// Both the base and the target are tiled at `chunk_size`. For each tile:
///
/// * bytes identical to the base's tile at the same offset → a **delta unit**,
/// which for an unchanged region is a single COPY instruction of a few bytes
/// whatever the tile's size;
/// * anything else → an ordinary **byte unit**.
///
/// So a large file with one edited region stores the edited region in full and
/// everything else as ~6 bytes per tile, and `EntryReader` had no way to say
/// that: its choice was per entry, so the whole file went one way or the other.
///
/// # What it costs, said plainly
///
/// Every delta unit names the same base, and reconstruction resolves that base
/// **once** — see the memo in [`Entry::reconstruct`]. Without that memo this
/// shape would be quadratic in the number of unchanged tiles, which is the one
/// new cost moving the seam introduces and the reason the memo is not optional.
///
/// A tile that differs is stored whole rather than delta'd against its
/// counterpart. That is the conservative choice: a per-tile delta would need its
/// own base-size bookkeeping for no saving on the case this exists for.
pub struct RegionDeltaSplitter {
pub chunk_size: usize,
}
impl Splitter for RegionDeltaSplitter {
fn split(
&self,
path: &str,
bytes: &[u8],
previous: Option<(&str, &[u8], usize)>,
compressed_len: &mut dyn FnMut(&[u8]) -> usize,
) -> Vec<StoredUnit> {
let size = self.chunk_size.max(1);
let fallback = ChunkSplitter { chunk_size: size };
let (base_path, base_bytes, base_chain) = match previous {
None => return fallback.split(path, bytes, previous, compressed_len),
Some(p) => p,
};
// The writer's own chain policy still applies: a delta unit adds a link
// exactly as a whole-entry delta did.
if base_chain + 1 > MAX_DELTA_CHAIN {
return fallback.split(path, bytes, previous, compressed_len);
}
let mut units = Vec::new();
let mut any_delta = false;
let mut off = 0usize;
while off < bytes.len() {
let end = (off + size).min(bytes.len());
let tile = &bytes[off..end];
let base_tile = base_bytes.get(off..end);
if base_tile == Some(tile) && !tile.is_empty() {
// Unchanged: one COPY of `tile.len()` from the base at `off`.
let mut delta = Vec::new();
put_size_varint(&mut delta, base_bytes.len() as u64);
put_size_varint(&mut delta, tile.len() as u64);
emit_copy(&mut delta, off as u64, tile.len() as u64);
any_delta = true;
units.push(StoredUnit {
fdata_offset: off as u64,
payload: UnitPayload::Delta {
base_path: base_path.to_string(),
delta,
expected: tile.to_vec(),
},
});
} else {
units.push(StoredUnit {
fdata_offset: off as u64,
payload: UnitPayload::Bytes(tile.to_vec()),
});
}
off = end;
}
// No tile matched, so this is the ordinary tiling with extra bookkeeping.
// Return the plain one rather than an equivalent-but-different encoding.
if !any_delta {
return fallback.split(path, bytes, previous, compressed_len);
}
units
}
}
/// One cached typed view: built once; `Ok(None)` = no sub-index of that type,
/// `Err` = the build failed (kept as text so every later call reports it too).
type ViewSlot = OnceLock<std::result::Result<Option<Box<dyn Any + Send + Sync>>, String>>;
/// The `Option` face of a view: a build failure is logged, never silently
/// turned into "no such type".
fn view_or_log<V>(r: Result<Option<&V>>) -> Option<&V> {
r.unwrap_or_else(|e| {
log::error!("{e:#}");
None
})
}
/// A znippy archive opened for random-access reads.
/// Loads only the Arrow IPC index on open; blobs are pread on demand. The
/// archive fd is shared (`Arc<File>`) and read via positioned I/O, so
/// `extract_file` is safe to call concurrently from many threads.
pub struct ZnippyArchive {
archive: Arc<File>,
/// Real byte length of the archive file, cached at open. Every `pread`
/// bounds-checks the (index-declared, therefore untrusted) `blob_offset +
/// blob_size` against this BEFORE allocating, so a corrupt/malicious index
/// can never force a giant zero-fill allocation (DoS) — mirrors the check
/// already in `decompress::get_file`.
archive_len: u64,
/// One [`Entry`] per path. Not boxed and not a trait object: an entry is a
/// list of chunks, and which KIND of chunk each one is lives in the chunk
/// (see [`ChunkSource`]) rather than in a per-entry implementation. That is
/// what lets one entry hold ordinary chunks and delta chunks side by side.
file_index: HashMap<String, Entry>,
/// Archive path — kept so the typed views can do the one-time filtered
/// sub-index read at view construction.
path: PathBuf,
/// The typed view cache, one slot per format ([`CoordSpec::SLOT`]). Built
/// once on first `as_*()` call and reused (the HARD perf contract: repeated
/// `as_maven()` is free). `None`
/// inside the `Ok` means "no sub-index of that type in this archive"; `Err`
/// is a build failure, cached so it is reported the same way every time.
views: [ViewSlot; VIEW_SLOTS],
}
impl ZnippyArchive {
pub fn open(path: &Path) -> Result<Self> {
let (_, batches) = read_znippy_index(path)?;
let mut file_index = Self::build_file_index(&batches)?;
// Absent section -> every chunk stays `Stored` and this read path is
// byte-for-byte the one `GOLDEN_CHUNKED_DIGEST` pins.
Self::apply_delta_map(path, &mut file_index)?;
let file = File::open(path)?;
let archive_len = file.metadata()?.len();
let archive = Arc::new(file);
Ok(Self {
archive,
archive_len,
file_index,
path: path.to_path_buf(),
views: Default::default(),
})
}
pub fn file_count(&self) -> usize {
self.file_index.len()
}
/// The typed view for format `S`, built once and cached. `Ok(None)` = the
/// archive has no sub-index of that type; `Err` = it has one that could not
/// be read.
pub fn try_view<S: CoordSpec>(&self) -> Result<Option<&CoordView<S>>> {
let slot = self.views[S::SLOT].get_or_init(|| {
CoordView::<S>::build(&self.path, Arc::clone(&self.archive))
.map(|v| (!v.is_empty()).then(|| Box::new(v) as Box<dyn Any + Send + Sync>))
.map_err(|e| format!("{e:#}"))
});
match slot {
Ok(v) => Ok(v.as_ref().and_then(|b| b.downcast_ref())),
Err(e) => Err(anyhow!("typed view (pkg_type {}) build failed: {e}", S::PKG_TYPE)),
}
}
/// Typed **rust/cargo** view (coords → crate); `None` if absent — or unreadable, which
/// [`Self::try_as_rust`] reports as the error it is.
pub fn as_rust(&self) -> Option<&RustView> {
view_or_log(self.try_view::<RustSpec>())
}
pub fn try_as_rust(&self) -> Result<Option<&RustView>> {
self.try_view::<RustSpec>()
}
/// Typed **maven** view (GAV[+classifier] → artifact); `None` if absent — or unreadable, which
/// [`Self::try_as_maven`] reports as the error it is.
pub fn as_maven(&self) -> Option<&MavenView> {
view_or_log(self.try_view::<MavenSpec>())
}
pub fn try_as_maven(&self) -> Result<Option<&MavenView>> {
self.try_view::<MavenSpec>()
}
/// Typed **python** view (name, version → wheel/sdist); `None` if absent — or unreadable, which
/// [`Self::try_as_python`] reports as the error it is.
pub fn as_python(&self) -> Option<&PythonView> {
view_or_log(self.try_view::<PythonSpec>())
}
pub fn try_as_python(&self) -> Result<Option<&PythonView>> {
self.try_view::<PythonSpec>()
}
/// Typed **npm** view (name incl. @scope, version → tarball); `None` if absent — or unreadable, which
/// [`Self::try_as_npm`] reports as the error it is.
pub fn as_npm(&self) -> Option<&NpmView> {
view_or_log(self.try_view::<NpmSpec>())
}
pub fn try_as_npm(&self) -> Result<Option<&NpmView>> {
self.try_view::<NpmSpec>()
}
/// Typed **gem** view (name, version[, platform] → gem); `None` if absent — or unreadable, which
/// [`Self::try_as_gem`] reports as the error it is.
pub fn as_gem(&self) -> Option<&GemView> {
view_or_log(self.try_view::<GemSpec>())
}
pub fn try_as_gem(&self) -> Result<Option<&GemView>> {
self.try_view::<GemSpec>()
}
/// Typed **conda** view (name, version[, build, subdir] → package); `None` if absent — or unreadable, which
/// [`Self::try_as_conda`] reports as the error it is.
pub fn as_conda(&self) -> Option<&CondaView> {
view_or_log(self.try_view::<CondaSpec>())
}
pub fn try_as_conda(&self) -> Result<Option<&CondaView>> {
self.try_view::<CondaSpec>()
}
/// Typed **rpm** view (NEVRA incl. epoch → rpm); `None` if absent — or unreadable, which
/// [`Self::try_as_rpm`] reports as the error it is.
pub fn as_rpm(&self) -> Option<&RpmView> {
view_or_log(self.try_view::<RpmSpec>())
}
pub fn try_as_rpm(&self) -> Result<Option<&RpmView>> {
self.try_view::<RpmSpec>()
}
/// Typed **deb** view (name, version, arch → deb + control); `None` if absent — or unreadable, which
/// [`Self::try_as_deb`] reports as the error it is.
pub fn as_deb(&self) -> Option<&DebView> {
view_or_log(self.try_view::<DebSpec>())
}
pub fn try_as_deb(&self) -> Result<Option<&DebView>> {
self.try_view::<DebSpec>()
}
fn build_file_index(batches: &[RecordBatch]) -> Result<HashMap<String, Entry>> {
let mut index: HashMap<String, Entry> = HashMap::new();
for batch in batches {
let paths = batch
.column_by_name("relative_path")
.ok_or_else(|| anyhow!("missing relative_path column"))?
.as_any()
.downcast_ref::<StringArray>()
.ok_or_else(|| anyhow!("relative_path not StringArray"))?;
let compressed_col = batch
.column_by_name("compressed")
.ok_or_else(|| anyhow!("missing compressed column"))?
.as_any()
.downcast_ref::<BooleanArray>()
.ok_or_else(|| anyhow!("compressed not BooleanArray"))?;
let sizes = batch
.column_by_name("uncompressed_size")
.ok_or_else(|| anyhow!("missing uncompressed_size column"))?
.as_any()
.downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("uncompressed_size not UInt64Array"))?;
let blob_offset_col = batch
.column_by_name("blob_offset")
.ok_or_else(|| anyhow!("missing blob_offset column"))?
.as_any()
.downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("blob_offset not UInt64Array"))?;
let blob_size_col = batch
.column_by_name("blob_size")
.ok_or_else(|| anyhow!("missing blob_size column"))?
.as_any()
.downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("blob_size not UInt64Array"))?;
let chunk_seq_col = batch
.column_by_name("chunk_seq")
.and_then(|c| c.as_any().downcast_ref::<UInt32Array>());
let fdata_offset_col = batch
.column_by_name("fdata_offset")
.ok_or_else(|| anyhow!("missing fdata_offset column"))?
.as_any()
.downcast_ref::<UInt64Array>()
.ok_or_else(|| anyhow!("fdata_offset not UInt64Array"))?;
// Optional: older archives may lack a valid 32-byte checksum column.
// When absent (or the wrong width), verification is simply skipped —
// the fast `extract_file` path never touched it, so this is additive.
let checksum_col = batch
.column_by_name("checksum")
.and_then(|c| c.as_any().downcast_ref::<FixedSizeBinaryArray>())
.filter(|c| c.value_length() == 32);
for row in 0..batch.num_rows() {
let path = paths.value(row).to_string();
let compressed = compressed_col.value(row);
let uncompressed_size = sizes.value(row);
let blob_offset = blob_offset_col.value(row);
let blob_size = blob_size_col.value(row);
let fdata_offset = fdata_offset_col.value(row);
let mut checksum = [0u8; 32];
if let Some(col) = checksum_col {
checksum.copy_from_slice(col.value(row));
}
let entry = index.entry(path).or_insert_with(|| Entry {
uncompressed_size: 0,
chunks: Vec::new(),
});
// A file's size is where its furthest chunk ENDS, not the sum of the
// row sizes. For the normal contiguous tiling these are equal, but
// summing made two index rows for the same path (which `append`
// used to produce) report double the real size — and, via
// `Vec::with_capacity` below, size an allocation off it.
entry.uncompressed_size = entry
.uncompressed_size
.max(fdata_offset.saturating_add(uncompressed_size));
entry.chunks.push(ChunkInfo {
blob_offset,
blob_size,
fdata_offset,
compressed,
uncompressed_size,
checksum,
chunk_seq: chunk_seq_col.map(|c| c.value(row)).unwrap_or(0),
source: ChunkSource::Stored,
});
}
}
for entry in index.values_mut() {
entry.chunks.sort_by_key(|c| c.fdata_offset);
}
Ok(index)
}
/// Turn the chunks named by [`ZNIPPY_DELTA_MODULE`] into delta chunks.
///
/// This is the whole cost of the format: one optional reserved section, read
/// once at open, joined onto `(relative_path, chunk_seq)`. An archive without
/// the section is untouched — which is every archive written before this
/// existed, and is why no [`ZNIPPY_FORMAT_VERSION`] bump is needed.
///
/// A row naming a chunk that does not exist is an **error**, not a silent
/// skip: it means the map and the index disagree, and the failure mode of
/// carrying on is serving a delta's instruction stream as if it were file
/// content.
fn apply_delta_map(path: &Path, index: &mut HashMap<String, Entry>) -> Result<()> {
let Some(bytes) = read_reserved_section_bytes(path, ZNIPPY_DELTA_MODULE)? else {
return Ok(());
};
let mut reader = crate::ipc_guard::stream_reader(bytes)
.map_err(|e| anyhow!("delta map: {e}"))?;
while let Some(batch) = reader.next() {
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 row in 0..batch.num_rows() {
let p = paths.value(row);
let seq = seqs.value(row);
let base = bases.value(row).to_string();
let entry = index
.get_mut(p)
.ok_or_else(|| anyhow!("delta map names entry {p}, which the index has not"))?;
let chunk = entry
.chunks
.iter_mut()
.find(|c| c.chunk_seq == seq)
.ok_or_else(|| anyhow!("delta map names chunk {seq} of {p}, which the index has not"))?;
chunk.source = ChunkSource::Delta { base_path: base };
}
}
Ok(())
}
/// Shared read loop for [`ZnippyReader::extract_file`] (fast, `verify=false`)
/// and [`Self::extract_file_verified`] (`verify=true`). Every chunk's
/// index-declared `blob_offset + blob_size` is bounds-checked against the
/// real archive length before the `resize`, so a corrupt/hostile index can
/// never force a giant allocation. With `verify` set, each chunk's
/// reconstructed bytes are blake3-checked against the per-chunk `checksum`.
fn extract_inner(&self, relative_path: &str, verify: bool) -> Result<Vec<u8>> {
self.extract_at_depth(relative_path, verify, 0)
}
/// `extract_inner` with the caller's recursion depth, which is what a delta
/// chunk's base resolution re-enters through.
fn extract_at_depth(&self, relative_path: &str, verify: bool, depth: usize) -> Result<Vec<u8>> {
let entry = self
.file_index
.get(relative_path)
.ok_or_else(|| anyhow!("file not found in archive: {}", relative_path))?;
entry.reconstruct(relative_path, &self.reconstruct_ctx(depth), verify)
}
/// The context every storage model reconstructs against: this archive's fd,
/// its real length, and this archive as the base resolver.
fn reconstruct_ctx(&self, depth: usize) -> ReconstructCtx<'_> {
ReconstructCtx {
archive: &self.archive,
archive_len: self.archive_len,
resolve: Some(self),
depth,
}
}
/// Random-access read of `relative_path` that **blake3-verifies** every chunk
/// against the per-chunk checksum in the index before returning (Design law 3).
///
/// This is the integrity-checked counterpart to the fast
/// [`ZnippyReader::extract_file`], which — by deliberate design, to keep the
/// artifact-serving hot path allocation-light — does NOT re-hash. Callers that
/// serve untrusted or long-lived archives (e.g. a registry) should prefer this.
/// Errors if any chunk's bytes do not match, or if the file is absent. Archives
/// written before the checksum column skip the hash silently (nothing to check).
///
/// It is also the read for an archive nobody vouches for: every compressed
/// chunk is decoded under an OpenZL allocation budget sized from the chunk's
/// declared size (`codec::decompress_into_budgeted`), so a hostile frame cannot
/// make OpenZL allocate past it.
pub fn extract_file_verified(&self, relative_path: &str) -> Result<Vec<u8>> {
self.extract_inner(relative_path, true)
}
}
/// The archive resolves a base by reconstructing that entry in full, through the
/// same dispatch as any other read — so a base may itself be chunked or delta'd
/// and neither model needs to know which.
impl BaseResolve for ZnippyArchive {
fn resolve(&self, path: &str, verify: bool, depth: usize) -> Result<Vec<u8>> {
self.extract_at_depth(path, verify, depth)
}
}
impl ZnippyReader for ZnippyArchive {
fn list_files(&self) -> Result<Vec<String>> {
Ok(self.file_index.keys().cloned().collect())
}
fn extract_file(&self, relative_path: &str) -> Result<Vec<u8>> {
self.extract_inner(relative_path, false)
}
fn contains(&self, relative_path: &str) -> bool {
self.file_index.contains_key(relative_path)
}
fn file_size(&self, relative_path: &str) -> Option<u64> {
self.file_index
.get(relative_path)
.map(|e| e.uncompressed_size())
}
}
// Every test here writes a real archive with compressed blobs and reads it back,
// so all of them need the codec. Without the `openzl` feature they are not
// "skipped for convenience" — the thing they exercise is genuinely absent, and
// the refusal itself is asserted in `codec::no_codec_tests` instead.
#[cfg(all(test, feature = "openzl"))]
mod tests {
use super::*;
use crate::codec::CompressCtx;
use crate::index::{build_metadata_batch, lookup_schema};
use crate::meta::{BlobMeta, ChunkMeta};
use crate::meta_sink::{ArchiveMetaSink, ArrowIpcSink, GroupKey};
use std::os::unix::fs::FileExt;
use std::time::{SystemTime, UNIX_EPOCH};
fn tmp(tag: &str) -> PathBuf {
let ns = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
let d = std::env::temp_dir().join(format!("znippy_archive_{tag}_{ns}"));
std::fs::create_dir_all(&d).unwrap();
d
}
/// Write a one-blob-per-file sealed archive. `size_override` lets a test
/// declare a `blob_size` in the index that differs from the bytes actually on
/// disk (used to synthesize a corrupt/hostile index for the bounds-check test).
fn write_archive(
path: &Path,
files: &[(String, Vec<u8>)],
size_override: Option<u64>,
) -> 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: size_override.unwrap_or(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 = build_metadata_batch(&blobs, resolver, &[], &[]).unwrap();
let schema = lookup_schema();
let mut sink = ArrowIpcSink::new(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()
}
/// Write an archive where each file is split into SEVERAL chunks at
/// successive `fdata_offset`s, and the rows are emitted **out of order**.
///
/// [`write_archive`] puts one chunk per file, all at `fdata_offset` 0, which
/// makes chunk ordering unobservable — a differential built only on it stays
/// green while chunks are reordered or dropped. This helper is what gives the
/// ordering something to be wrong about.
fn write_archive_multichunk(
path: &Path,
files: &[(String, Vec<u8>)],
chunk_size: usize,
) -> 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() {
paths.push(rel.clone());
let mut rows = Vec::new();
let mut off = 0usize;
let mut seq = 0u64;
// An empty file still needs one row, or it vanishes from the index.
while off < bytes.len() || (bytes.is_empty() && seq == 0) {
let end = bytes.len().min(off + chunk_size.max(1));
let piece = &bytes[off..end];
let checksum = *blake3::hash(piece).as_bytes();
let frame = ctx.compress(piece).unwrap();
let (on_disk, compressed): (&[u8], bool) =
if frame.len() < piece.len() { (&frame, true) } else { (piece, false) };
file.write_all_at(on_disk, cursor).unwrap();
rows.push(BlobMeta {
blob_offset: cursor,
blob_size: on_disk.len() as u64,
chunk_meta: ChunkMeta {
fdata_offset: off as u64,
file_index: fi as u64,
chunk_seq: seq as u32,
checksum,
compressed,
uncompressed_size: piece.len() as u64,
compressed_size: on_disk.len() as u64,
},
});
cursor += on_disk.len() as u64;
off = end;
seq += 1;
if bytes.is_empty() {
break;
}
}
// Emit LAST chunk first. If `build_file_index` stops sorting by
// `fdata_offset`, reconstruction places chunks in this order and the
// bytes come out wrong — which is exactly what the differential must
// be able to see.
rows.reverse();
blobs.extend(rows);
}
let resolver = { let p = paths.clone(); move |fi: u64| p[fi as usize].clone() };
let batch = build_metadata_batch(&blobs, resolver, &[], &[]).unwrap();
let schema = lookup_schema();
let mut sink = ArrowIpcSink::new(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()
}
const GOLDEN_CHUNKED_DIGEST: &str =
"bc191568a85e66bd67773a75ba4fbf1438762cc562bdb190befbced15fa3e8fd";
/// **The trait-extraction differential (LAW 2).**
///
/// `ChunkedEntry::reconstruct` is the pre-trait `extract_inner` loop moved
/// verbatim, so the risk is not in the loop — it is in the plumbing around
/// it: index building, boxing, dispatch, and the ordering of entries. A test
/// that only round-trips would pass on a refactor that silently reordered or
/// dropped chunks, because it would compare the output against itself.
///
/// So this pins ONE digest over a deterministic corpus, covering the shapes
/// that plumbing gets wrong: compressible and incompressible payloads, an
/// empty file, a single byte, and paths whose sort order differs from their
/// insertion order. The same constant is produced by the implementation on
/// `origin/master` before the trait existed; if a refactor changes a byte or
/// an order, this goes red and the identical-output claim is retracted.
///
/// The digest folds in the path, the declared size and the bytes, so a
/// mis-mapped entry (right bytes, wrong path) fails too.
fn corpus() -> Vec<(String, Vec<u8>)> {
let mut v: Vec<(String, Vec<u8>)> = Vec::new();
v.push(("z/last.txt".into(), b"zzz".to_vec()));
v.push(("a/empty.bin".into(), Vec::new()));
v.push(("m/one.bin".into(), vec![0x5a]));
// Highly compressible: exercises the `compressed = true` branch.
v.push(("c/runs.txt".into(), vec![b'q'; 9000]));
// Incompressible: exercises the stored-raw branch.
let mut s = 0x1234_5678_9abc_def0u64;
let noise: Vec<u8> = (0..7777u32)
.map(|_| {
s = s.wrapping_add(0x9e37_79b9_7f4a_7c15);
let mut z = s;
z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
(z ^ (z >> 31)) as u8
})
.collect();
v.push(("n/noise.bin".into(), noise));
v.push(("b/mixed.dat".into(), {
let mut b = vec![7u8; 300];
b.extend_from_slice(&[1, 2, 3, 4, 5]);
b.extend(std::iter::repeat(0xffu8).take(1200));
b
}));
v
}
fn corpus_digest(ar: &ZnippyArchive) -> String {
let mut names = ar.list_files().unwrap();
names.sort();
let mut h = blake3::Hasher::new();
for n in &names {
h.update(n.as_bytes());
h.update(&ar.file_size(n).unwrap_or(0).to_le_bytes());
h.update(&ar.extract_file(n).unwrap());
h.update(&ar.extract_file_verified(n).unwrap());
}
h.finalize().to_hex().to_string()
}
#[test]
fn chunked_reconstruction_is_byte_identical_to_the_pre_trait_implementation() {
let dir = tmp("differential");
let path = dir.join("a.znippy");
let files = corpus();
write_archive_multichunk(&path, &files, 1000);
let ar = ZnippyArchive::open(&path).unwrap();
// Every file comes back exactly as written — the property the digest
// then pins across implementations.
for (rel, bytes) in &files {
assert_eq!(&ar.extract_file(rel).unwrap(), bytes, "round-trip of {rel}");
}
assert_eq!(
corpus_digest(&ar),
GOLDEN_CHUNKED_DIGEST,
"chunked reconstruction changed. Either the refactor is not \
byte-identical, or the archive format changed and this constant \
must be re-derived from the previous commit ON PURPOSE."
);
let _ = std::fs::remove_dir_all(&dir);
}
// ---- delta storage model -------------------------------------------------
/// Encode a git-style varint (the size header form).
fn put_varint(out: &mut Vec<u8>, mut v: u64) {
loop {
let mut b = (v & 0x7f) as u8;
v >>= 7;
if v != 0 {
b |= 0x80;
}
out.push(b);
if v == 0 {
return;
}
}
}
/// Minimal delta encoder for tests: copy `base[..copy_len]`, then insert
/// `tail`. Enough to produce a real, valid instruction stream without
/// pulling in an encoder the read side does not need.
fn encode_delta(base_len: usize, copy_len: usize, tail: &[u8]) -> Vec<u8> {
let mut d = Vec::new();
put_varint(&mut d, base_len as u64);
put_varint(&mut d, (copy_len + tail.len()) as u64);
if copy_len > 0 {
// COPY with a 1-byte offset (0) and a 3-byte size.
d.push(0x80 | 0x01 | 0x10 | 0x20 | 0x40);
d.push(0); // offset byte 0
d.push((copy_len & 0xff) as u8);
d.push(((copy_len >> 8) & 0xff) as u8);
d.push(((copy_len >> 16) & 0xff) as u8);
}
for piece in tail.chunks(0x7f) {
d.push(piece.len() as u8);
d.extend_from_slice(piece);
}
d
}
/// Write an archive whose blob region holds the delta streams FIRST and the
/// base entry after them, then seals the index over the base alone.
///
/// The deltas must live inside the blob region rather than after the sealed
/// archive: znippy reads its index from the tail, so appending anything past
/// the seal makes the file unopenable ("v0.6 archives are not supported").
fn write_base_and_deltas(
path: &Path,
entries: &[(String, Vec<u8>)],
deltas: &[Vec<u8>],
) -> Vec<(u64, u64, [u8; 32])> {
let file = Arc::new(File::create(path).unwrap());
let mut cursor = 0u64;
let mut rows = Vec::new();
for d in deltas {
file.write_all_at(d, cursor).unwrap();
rows.push((cursor, d.len() as u64, *blake3::hash(d).as_bytes()));
cursor += d.len() as u64;
}
// Bases are stored RAW so their bytes are exactly what a delta expects.
let mut blobs = Vec::new();
let mut names = Vec::new();
for (fi, (rel, bytes)) in entries.iter().enumerate() {
let off = cursor;
file.write_all_at(bytes, off).unwrap();
cursor += bytes.len() as u64;
names.push(rel.clone());
blobs.push(BlobMeta {
blob_offset: off,
blob_size: bytes.len() as u64,
chunk_meta: ChunkMeta {
fdata_offset: 0,
file_index: fi 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,
},
});
}
let resolver = move |fi: u64| names[fi as usize].clone();
let batch = build_metadata_batch(&blobs, resolver, &[], &[]).unwrap();
let schema = lookup_schema();
let mut sink = ArrowIpcSink::new(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();
rows
}
/// A corrupt instruction stream is refused rather than turned into
/// plausible-looking bytes. Covers the three refusals that a lenient decoder
/// would silently paper over.
#[test]
fn corrupt_delta_streams_are_refused() {
let base = b"0123456789abcdef".to_vec();
// copy past the end of the base
let mut bad = Vec::new();
put_varint(&mut bad, base.len() as u64);
put_varint(&mut bad, 32);
bad.extend_from_slice(&[0x80 | 0x01 | 0x10, 0, 32]);
let e = apply_delta(&base, &bad).unwrap_err().to_string();
assert!(e.contains("from a base of"), "copy overrun: {e}");
// the unassigned 0x00 instruction
let mut zero = Vec::new();
put_varint(&mut zero, base.len() as u64);
put_varint(&mut zero, 1);
zero.push(0x00);
let e = apply_delta(&base, &zero).unwrap_err().to_string();
assert!(e.contains("0x00"), "zero opcode: {e}");
// a delta built for a base of another length
let d = encode_delta(base.len() + 1, 4, b"xy");
let e = apply_delta(&base, &d).unwrap_err().to_string();
assert!(e.contains("expects a base of"), "base size: {e}");
// result shorter than declared (truncated chain)
let mut short = Vec::new();
put_varint(&mut short, base.len() as u64);
put_varint(&mut short, 99);
short.push(2);
short.extend_from_slice(b"ab");
let e = apply_delta(&base, &short).unwrap_err().to_string();
assert!(e.contains("declared"), "short result: {e}");
}
/// **Encoder and decoder are inverses, on inputs shaped like real edits.**
///
/// This is the property that matters: `encode_delta_against` is a second
/// implementation of the format `apply_delta` reads, so if they ever
/// disagree the archive is unreadable. Asserted over edit shapes that
/// historization actually produces — append, prepend, middle insert, middle
/// delete, whole-file replace, identical, empty either side.
#[test]
fn encoder_and_decoder_are_inverses() {
let body: Vec<u8> = (0..40_000u32).map(|i| (i.wrapping_mul(2654435761) >> 13) as u8).collect();
let mut cases: Vec<(Vec<u8>, Vec<u8>)> = Vec::new();
cases.push((body.clone(), body.clone())); // identical
cases.push((body.clone(), { let mut v = body.clone(); v.extend_from_slice(b"appended tail bytes"); v }));
cases.push((body.clone(), { let mut v = b"prepended header".to_vec(); v.extend_from_slice(&body); v }));
cases.push((body.clone(), { let mut v = body[..15_000].to_vec(); v.extend_from_slice(b"INSERTED IN THE MIDDLE OF IT"); v.extend_from_slice(&body[15_000..]); v }));
cases.push((body.clone(), { let mut v = body[..10_000].to_vec(); v.extend_from_slice(&body[20_000..]); v }));
cases.push((body.clone(), (0..30_000u32).map(|i| (i.wrapping_mul(40503) >> 7) as u8).collect()));
cases.push((body.clone(), Vec::new())); // emptied
cases.push((Vec::new(), body.clone())); // created
cases.push((Vec::new(), Vec::new()));
cases.push((b"short".to_vec(), b"shorter".to_vec())); // below MIN_COPY
for (i, (base, target)) in cases.iter().enumerate() {
let d = encode_delta_against(base, target);
let got = apply_delta(base, &d)
.unwrap_or_else(|e| panic!("case {i}: decode failed: {e}"));
assert_eq!(&got, target, "case {i}: round-trip mismatch");
}
}
/// An append-shaped edit must actually be SMALL — an encoder that emits all
/// literals round-trips perfectly and is worthless, which is precisely the
/// kind of green a round-trip test alone cannot distinguish.
#[test]
fn a_small_edit_produces_a_small_delta() {
let body: Vec<u8> = (0..200_000u32).map(|i| (i.wrapping_mul(2654435761) >> 13) as u8).collect();
let mut edited = body.clone();
edited.extend_from_slice(b"one short appended line\n");
let d = encode_delta_against(&body, &edited);
assert_eq!(apply_delta(&body, &d).unwrap(), edited);
assert!(
d.len() < body.len() / 100,
"a 24-byte append to 200 kB must not cost {} B of delta",
d.len()
);
}
/// The three refusal reasons, each on real bytes.
#[test]
fn the_writer_refuses_a_delta_for_stated_reasons() {
let a: Vec<u8> = {
let mut st = 0x0123_4567_89ab_cdefu64;
(0..50_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 ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
(z ^ (z >> 31)) as u8
})
.collect()
};
let mut b = a.clone();
b.extend_from_slice(b"tail");
let clen = |x: &[u8]| CompressCtx::new(3).unwrap().compress(x).unwrap().len();
// First version of a path.
let (plan, d) = plan_version("p", None, &a, clen);
assert_eq!(plan, VersionPlan::Chunked(ChunkedReason::FirstVersion));
assert!(d.is_none());
// A real second version deltas, and the payload comes back.
let (plan, d) = plan_version("p", Some(("p", &a, 0)), &b, clen);
match plan {
VersionPlan::Delta { ref base_path, chain_len, delta_bytes } => {
assert_eq!(base_path, "p");
assert_eq!(chain_len, 1);
assert!(delta_bytes < a.len() / 50, "delta should be tiny, was {delta_bytes}");
}
other => panic!("expected a delta, got {other:?}"),
}
assert_eq!(apply_delta(&a, d.as_ref().unwrap()).unwrap(), b);
// Chain at the cap refuses.
let (plan, d) = plan_version("p", Some(("p", &a, MAX_DELTA_CHAIN)), &b, clen);
assert_eq!(plan, VersionPlan::Chunked(ChunkedReason::ChainTooLong));
assert!(d.is_none());
// Unrelated content, and genuinely high-entropy: a smooth arithmetic
// ramp is highly compressible, so a delta of literals over one can
// compress to LESS than the target and the refusal never fires. That is
// not a bug in the writer, it is a bug in the corpus, and it cost a red
// to notice. splitmix64 gives bytes neither the codec nor the match
// finder can do anything with.
let unrelated: Vec<u8> = {
let mut st = 0xdead_beef_cafe_1234u64;
(0..50_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 ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
(z ^ (z >> 31)) as u8
})
.collect()
};
let (plan, d) = plan_version("p", Some(("p", &a, 0)), &unrelated, clen);
assert_eq!(plan, VersionPlan::Chunked(ChunkedReason::DeltaNotSmaller));
assert!(d.is_none());
}
/// A verified read returns the exact bytes; a blob corrupted on disk is caught
/// by `extract_file_verified` (blake3 mismatch) while the fast `extract_file`
/// hands the corrupt bytes back unflagged — proving the verified path adds the
/// integrity check that the deliberately-fast path omits.
#[test]
fn extract_file_verified_catches_corruption_fast_path_does_not() {
let dir = tmp("verify");
let archive = dir.join("a.znippy");
// High-entropy (incompressible) payloads → stored raw, so flipping one
// byte changes the reconstructed bytes (checksum mismatch) without
// corrupting a compressed frame the decoder would then reject. A
// splitmix64 stream gives per-byte pseudo-random fill that zstd/OpenZL
// cannot shrink, so write_archive keeps it uncompressed.
let files: Vec<(String, Vec<u8>)> = (0..8)
.map(|i| {
let mut s = 0x9e37_79b9_7f4a_7c15u64 ^ (i as u64);
let body: Vec<u8> = (0..4096u32)
.map(|_| {
s = s.wrapping_add(0x9e37_79b9_7f4a_7c15);
let mut z = s;
z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
(z ^ (z >> 31)) as u8
})
.collect();
(format!("repo/f{i:03}.bin"), body)
})
.collect();
write_archive(&archive, &files, None);
// Clean archive: both paths agree and verification passes.
let ar = ZnippyArchive::open(&archive).unwrap();
for (p, bytes) in &files {
assert_eq!(&ar.extract_file(p).unwrap(), bytes);
assert_eq!(&ar.extract_file_verified(p).unwrap(), bytes, "clean verify for {p}");
}
drop(ar);
// Flip one byte inside the first blob region (offset 0).
{
let f = std::fs::OpenOptions::new().read(true).write(true).open(&archive).unwrap();
let mut b = [0u8; 1];
f.read_exact_at(&mut b, 0).unwrap();
b[0] ^= 0xFF;
f.write_all_at(&b, 0).unwrap();
}
let ar = ZnippyArchive::open(&archive).unwrap();
let target = &files[0].0;
// Fast path returns the corrupt bytes without complaint (by design).
assert_ne!(&ar.extract_file(target).unwrap(), &files[0].1);
// Verified path rejects them.
let err = ar.extract_file_verified(target).unwrap_err();
assert!(
err.to_string().contains("checksum mismatch"),
"expected checksum mismatch, got: {err}"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// Two index rows for the SAME `relative_path` (what `append` used to leave
/// behind, and what any hand-built/hostile index can carry) must not silently
/// produce a double-length buffer. `extract_inner` used to `extend_from_slice`
/// every chunk, so both copies came back concatenated at 2x the real length,
/// with `file_size()` reporting the summed size and even
/// `extract_file_verified` passing — each chunk's blake3 is individually
/// correct. Chunks must be PLACED at their `fdata_offset` instead.
#[test]
fn duplicate_rows_for_one_path_do_not_double_the_extracted_bytes() {
let dir = tmp("dup");
let archive = dir.join("dup.znippy");
let payload = b"one true copy of the payload".to_vec();
// Same path twice → two rows, both at fdata_offset 0.
let files = vec![
("repo/dup.bin".to_string(), payload.clone()),
("repo/dup.bin".to_string(), payload.clone()),
];
write_archive(&archive, &files, None);
let ar = ZnippyArchive::open(&archive).unwrap();
assert_eq!(
ar.file_size("repo/dup.bin"),
Some(payload.len() as u64),
"file_size must not be the SUM over duplicate rows"
);
assert_eq!(
ar.extract_file("repo/dup.bin").unwrap(),
payload,
"duplicate rows must not concatenate into a double-length buffer"
);
assert_eq!(ar.extract_file_verified("repo/dup.bin").unwrap(), payload);
let _ = std::fs::remove_dir_all(&dir);
}
/// `extract_files` went from a serial `map` to a gatling fork-join. The
/// speed is the point; **the results being identical is the contract**, and
/// it is the thing a parallel rewrite silently breaks. Two properties, both
/// asserted here against the serial map computed in this test:
///
/// 1. **index order** — `gatling_for_each_balanced` schedules
/// heaviest-first but returns in `0..n` order, so `out[i]` is still the
/// answer for `paths[i]`. The fixture is deliberately SKEWED (one big
/// file first, then small ones) so a schedule-ordered result would come
/// back visibly permuted rather than by luck identical.
/// 2. **errors stay in place** — a path the archive does not hold must be
/// an `Err` at its own index, not a dropped element that shifts every
/// later answer left by one.
///
/// Repeated, because a race that reorders or crosses buffers does not
/// necessarily show on the first run.
#[test]
fn extract_files_is_the_serial_map() {
let dir = tmp("batch");
let archive = dir.join("batch.znippy");
// Skewed on purpose: 512 KiB first, then 40 small ones. Heaviest-first
// therefore claims index 0 first, so any leak of schedule order into the
// result is visible immediately rather than by luck invisible.
let mut files: Vec<(String, Vec<u8>)> =
vec![("repo/big.bin".to_string(), vec![0xA5u8; 512 * 1024])];
for i in 0..40u8 {
files.push((format!("repo/small-{i}.bin"), vec![i; 64 + i as usize * 37]));
}
write_archive(&archive, &files, None);
let ar = ZnippyArchive::open(&archive).unwrap();
let mut paths: Vec<&str> = files.iter().map(|(p, _)| p.as_str()).collect();
// An absent path in the middle: its Err must not shift its neighbours.
paths.insert(7, "repo/not-in-this-archive.bin");
for run in 0..8 {
let parallel = ar.extract_files(&paths);
let serial: Vec<Result<Vec<u8>>> = paths.iter().map(|p| ar.extract_file(p)).collect();
assert_eq!(parallel.len(), serial.len(), "run {run}: length changed");
for (i, (par, ser)) in parallel.iter().zip(serial.iter()).enumerate() {
match (par, ser) {
(Ok(a), Ok(b)) => assert_eq!(
a, b,
"run {run}: extract_files[{i}] ({}) differs from the serial map",
paths[i]
),
(Err(_), Err(_)) => {}
_ => panic!(
"run {run}: extract_files[{i}] ({}) and the serial map disagree about \
whether it succeeded — an error was dropped or invented, which shifts \
every later answer",
paths[i]
),
}
}
}
let _ = std::fs::remove_dir_all(&dir);
}
/// A corrupt/hostile index that declares a multi-GB `blob_size` for a tiny
/// on-disk blob must yield a clean bounds Err — never a giant zero-fill
/// allocation (DoS). Guards the `checked_add`/`archive_len` check in the read
/// loop shared by both extract paths.
#[test]
fn extract_file_rejects_out_of_bounds_blob_without_huge_alloc() {
let dir = tmp("bounds");
let archive = dir.join("b.znippy");
let files = vec![("repo/small.bin".to_string(), b"hello znippy".to_vec())];
// Declare an 8 GiB blob_size while writing only a few real bytes.
write_archive(&archive, &files, Some(8 * 1024 * 1024 * 1024));
let ar = ZnippyArchive::open(&archive).unwrap();
let err = ar.extract_file("repo/small.bin").unwrap_err();
assert!(
err.to_string().contains("out of bounds"),
"expected out-of-bounds Err, got: {err}"
);
// The verified path funnels through the same guard.
assert!(ar.extract_file_verified("repo/small.bin").is_err());
let _ = std::fs::remove_dir_all(&dir);
}
// ---- delta AS A CHUNK ----------------------------------------------------
/// Build an entry out of stored/delta units, the way a splitter's output
/// would be written. Blob bytes are placed raw (never compressed) so the
/// deltas see exactly the bytes they were encoded against.
///
/// Returns the entry. The archive it references must already hold the base.
fn entry_from_units(units: Vec<StoredUnit>, blob_at: &mut dyn FnMut(&[u8]) -> (u64, u64)) -> Entry {
let mut chunks = Vec::new();
let mut size = 0u64;
for u in units {
let (produced, payload_bytes, source) = match u.payload {
UnitPayload::Bytes(b) => (b.clone(), b, ChunkSource::Stored),
UnitPayload::Delta { base_path, delta, expected } => {
(expected, delta, ChunkSource::Delta { base_path })
}
};
let (off, len) = blob_at(&payload_bytes);
size = size.max(u.fdata_offset + produced.len() as u64);
chunks.push(ChunkInfo {
blob_offset: off,
blob_size: len,
fdata_offset: u.fdata_offset,
compressed: false,
uncompressed_size: produced.len() as u64,
checksum: *blake3::hash(&produced).as_bytes(),
chunk_seq: 0,
source,
});
}
chunks.sort_by_key(|c| c.fdata_offset);
Entry { uncompressed_size: size, chunks }
}
/// A base plus a one-delta chunk reconstructs the intended bytes, and two
/// chained delta chunks apply in order — the property
/// `delta_chain_reconstructs_through_the_same_reader` asserted of the
/// per-entry model, now of a chunk.
#[test]
fn a_delta_chunk_reconstructs_through_the_one_reader() {
let dir = tmp("dchunk_ok");
let path = dir.join("d.znippy");
let base = b"the original object, stored once".to_vec();
let want1 = {
let mut v = base[..12].to_vec();
v.extend_from_slice(b" AND VERSION TWO");
v
};
let d1 = encode_delta(base.len(), 12, b" AND VERSION TWO");
let want2 = {
let mut v = want1[..5].to_vec();
v.extend_from_slice(b"third");
v
};
let d2 = encode_delta(want1.len(), 5, b"third");
let rows = write_base_and_deltas(
&path,
&[("obj/base.bin".to_string(), base.clone())],
&[d1.clone(), d2.clone()],
);
let ar = ZnippyArchive::open(&path).unwrap();
let mut at = |b: &[u8]| {
let (o, l, _) = *rows.iter().find(|(o, l, _)| {
let mut buf = vec![0u8; *l as usize];
ar.archive.read_exact_at(&mut buf, *o).unwrap();
buf == b
}).expect("payload not in blob region");
(o, l)
};
let e1 = entry_from_units(
vec![StoredUnit {
fdata_offset: 0,
payload: UnitPayload::Delta {
base_path: "obj/base.bin".into(),
delta: d1.clone(),
expected: want1.clone(),
},
}],
&mut at,
);
assert_eq!(e1.reconstruct("v1", &ar.reconstruct_ctx(0), false).unwrap(), want1);
assert_eq!(e1.reconstruct("v1", &ar.reconstruct_ctx(0), true).unwrap(), want1);
assert_eq!(e1.uncompressed_size(), want1.len() as u64);
let _ = (&d2, &want2);
let _ = std::fs::remove_dir_all(&dir);
}
/// **The integrity property, per chunk.**
///
/// The delta stream is intact and the base is a real, intact entry of exactly
/// the same LENGTH — but it is the wrong entry. The per-entry model needed a
/// separate `result_checksum` for this. A chunk hashed over the bytes it
/// PRODUCES catches it with the column it already had, on the fast read.
#[test]
fn a_correct_delta_chunk_against_the_wrong_base_is_caught_on_the_fast_read() {
let dir = tmp("dchunk_wrongbase");
let path = dir.join("d.znippy");
let base = b"the original object, stored once".to_vec();
let decoy: Vec<u8> = {
let mut d = b"A DIFFERENT FILE".to_vec();
d.resize(base.len(), b'!');
assert_eq!(d.len(), base.len(), "decoy must match the base length");
d
};
let want = {
let mut v = base[..12].to_vec();
v.extend_from_slice(b" AND VERSION TWO");
v
};
let d1 = encode_delta(base.len(), 12, b" AND VERSION TWO");
let rows = write_base_and_deltas(
&path,
&[
("obj/base.bin".to_string(), base.clone()),
("obj/decoy.bin".to_string(), decoy.clone()),
],
&[d1.clone()],
);
let ar = ZnippyArchive::open(&path).unwrap();
let (off, len, _) = rows[0];
let unit = |base_path: &str| {
Entry {
uncompressed_size: want.len() as u64,
chunks: vec![ChunkInfo {
blob_offset: off,
blob_size: len,
fdata_offset: 0,
compressed: false,
uncompressed_size: 0,
checksum: *blake3::hash(&want).as_bytes(),
chunk_seq: 0,
source: ChunkSource::Delta { base_path: base_path.to_string() },
}],
}
};
// Right base: reconstructs.
assert_eq!(
unit("obj/base.bin").reconstruct("v", &ar.reconstruct_ctx(0), false).unwrap(),
want
);
// Wrong base, same length, intact delta — `verify` is FALSE.
let err = unit("obj/decoy.bin")
.reconstruct("v", &ar.reconstruct_ctx(0), false)
.unwrap_err()
.to_string();
assert!(
err.contains("result checksum") && err.contains("obj/decoy.bin"),
"the wrong base is the same LENGTH, so only the output hash can catch it, \
and the message must name the base it used — got: {err}"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// A base corrupted IN PLACE to exactly the same byte length. Every bounds
/// check passes, the base-size varint matches, the delta stream is intact.
/// Only the output hash can see it, and it does so without `verify`.
#[test]
fn a_corrupted_base_of_identical_length_is_caught_by_the_chunk_checksum() {
let dir = tmp("dchunk_corruptbase");
let path = dir.join("cb.znippy");
let base: Vec<u8> = (0..8192u32).map(|i| (i.wrapping_mul(2654435761) >> 11) as u8).collect();
let mut target = base.clone();
target.extend_from_slice(b"\nthe second version\n");
let d1 = encode_delta_against(&base, &target);
let rows = write_base_and_deltas(
&path,
&[("v/base.bin".to_string(), base.clone())],
&[d1.clone()],
);
let (off, len, _) = rows[0];
let entry = || Entry {
uncompressed_size: target.len() as u64,
chunks: vec![ChunkInfo {
blob_offset: off,
blob_size: len,
fdata_offset: 0,
compressed: false,
uncompressed_size: 0,
checksum: *blake3::hash(&target).as_bytes(),
chunk_seq: 0,
source: ChunkSource::Delta { base_path: "v/base.bin".into() },
}],
};
{
let ar = ZnippyArchive::open(&path).unwrap();
assert_eq!(entry().reconstruct("v", &ar.reconstruct_ctx(0), false).unwrap(), target);
}
let base_off = rows.iter().map(|(o, s, _)| o + s).max().unwrap();
let f = File::options().read(true).write(true).open(&path).unwrap();
let mut b = [0u8; 1];
f.read_exact_at(&mut b, base_off).unwrap();
f.write_all_at(&[b[0] ^ 0xff], base_off).unwrap();
f.sync_all().unwrap();
let ar = ZnippyArchive::open(&path).unwrap();
assert_eq!(ar.file_size("v/base.bin"), Some(base.len() as u64));
let err = entry()
.reconstruct("v", &ar.reconstruct_ctx(0), false)
.unwrap_err()
.to_string();
assert!(err.contains("result checksum"), "got: {err}");
let _ = std::fs::remove_dir_all(&dir);
}
/// **The thing the per-entry seam could not express: one file, part chunk,
/// part delta.**
///
/// A 4-tile file whose second tile is edited. `RegionDeltaSplitter` emits
/// three delta units and one byte unit, the entry reconstructs exactly, and
/// the stored payload for the unchanged 3/4 of the file is a handful of
/// bytes. `EntryReader` had one choice per ENTRY and no way to say this.
#[test]
fn one_entry_can_be_part_chunk_and_part_delta() {
let dir = tmp("mixed");
let path = dir.join("m.znippy");
let tile = 4096usize;
let base: Vec<u8> = (0..(tile * 4) as u32)
.map(|i| (i.wrapping_mul(2654435761) >> 9) as u8)
.collect();
let mut target = base.clone();
for b in target[tile..tile * 2].iter_mut() {
*b ^= 0x5a;
}
let split = RegionDeltaSplitter { chunk_size: tile };
let mut clen = |b: &[u8]| b.len();
let units = split.split("v2", &target, Some(("v/base.bin", &base, 0)), &mut clen);
assert_eq!(units.len(), 4, "four tiles");
let deltas = units
.iter()
.filter(|u| matches!(u.payload, UnitPayload::Delta { .. }))
.count();
let stored = units
.iter()
.filter(|u| matches!(u.payload, UnitPayload::Bytes(_)))
.count();
assert_eq!((deltas, stored), (3, 1), "three unchanged tiles, one edited");
// The saving is the point: the unchanged three quarters cost a few bytes.
let delta_bytes: usize = units
.iter()
.filter_map(|u| match &u.payload {
UnitPayload::Delta { delta, .. } => Some(delta.len()),
_ => None,
})
.sum();
assert!(
delta_bytes < 64,
"three whole-tile COPY instructions should be tens of bytes, not {delta_bytes}"
);
// Write the payloads into the blob region and read the entry back.
let payloads: Vec<Vec<u8>> = units
.iter()
.map(|u| match &u.payload {
UnitPayload::Bytes(b) => b.clone(),
UnitPayload::Delta { delta, .. } => delta.clone(),
})
.collect();
let rows = write_base_and_deltas(&path, &[("v/base.bin".to_string(), base.clone())], &payloads);
let mut i = 0usize;
let mut at = |_b: &[u8]| {
let (o, l, _) = rows[i];
i += 1;
(o, l)
};
let entry = entry_from_units(units, &mut at);
let ar = ZnippyArchive::open(&path).unwrap();
assert_eq!(
entry.reconstruct("v2", &ar.reconstruct_ctx(0), false).unwrap(),
target,
"a mixed entry must reconstruct exactly"
);
assert_eq!(entry.uncompressed_size(), target.len() as u64);
let _ = std::fs::remove_dir_all(&dir);
}
/// **A base that names itself is an `Err`, not a stack overflow.**
///
/// The writer's `MAX_DELTA_CHAIN` binds nothing about an index that arrives
/// from elsewhere. `MAX_RECONSTRUCT_DEPTH` is the reader's own bound and this
/// is the guard for it; the per-entry model had the same hole and no such
/// guard.
#[test]
fn a_self_referential_base_errors_instead_of_recursing_forever() {
let dir = tmp("cycle");
let path = dir.join("c.znippy");
let base = b"a base that will be replaced by a cycle".to_vec();
let d = encode_delta(base.len(), 4, b"xy");
let rows = write_base_and_deltas(&path, &[("v/base.bin".to_string(), base.clone())], &[d]);
let ar = ZnippyArchive::open(&path).unwrap();
// Splice an entry that deltas against ITSELF into the opened index.
let mut ar = ar;
let (off, len, _) = rows[0];
ar.file_index.insert(
"v/loop.bin".to_string(),
Entry {
uncompressed_size: 6,
chunks: vec![ChunkInfo {
blob_offset: off,
blob_size: len,
fdata_offset: 0,
compressed: false,
uncompressed_size: 0,
checksum: [0u8; 32],
chunk_seq: 0,
source: ChunkSource::Delta { base_path: "v/loop.bin".into() },
}],
},
);
let err = ar.extract_file("v/loop.bin").unwrap_err().to_string();
assert!(
err.contains("deeper than"),
"a cycle must be refused by the depth bound, got: {err}"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **The memo, asserted by counting — not by a timing.**
///
/// A mixed entry with four delta chunks all naming one base must resolve that
/// base ONCE. Without the memo it is four, and the shape is quadratic in the
/// number of unchanged tiles — which would make `RegionDeltaSplitter`'s output
/// slower to read the more of it is unchanged, i.e. exactly backwards.
///
/// This is the one new cost delta-as-a-chunk introduces over the per-entry
/// model, so it is counted rather than assumed.
#[test]
fn a_mixed_entry_resolves_each_base_exactly_once() {
use std::sync::atomic::{AtomicUsize, Ordering};
struct Counting {
bytes: Vec<u8>,
calls: AtomicUsize,
}
impl BaseResolve for Counting {
fn resolve(&self, _path: &str, _verify: bool, _depth: usize) -> Result<Vec<u8>> {
self.calls.fetch_add(1, Ordering::SeqCst);
Ok(self.bytes.clone())
}
}
let dir = tmp("memo");
let path = dir.join("m.znippy");
let tile = 1024usize;
let base: Vec<u8> = (0..(tile * 4) as u32)
.map(|i| (i.wrapping_mul(2654435761) >> 9) as u8)
.collect();
// Four delta chunks, one per tile, all against the same base path.
let mut payloads = Vec::new();
for i in 0..4 {
let off = i * tile;
let mut d = Vec::new();
put_size_varint(&mut d, base.len() as u64);
put_size_varint(&mut d, tile as u64);
emit_copy(&mut d, off as u64, tile as u64);
payloads.push(d);
}
let rows = write_base_and_deltas(&path, &[("v/base.bin".to_string(), base.clone())], &payloads);
let file = Arc::new(File::open(&path).unwrap());
let archive_len = file.metadata().unwrap().len();
let chunks: Vec<ChunkInfo> = rows
.iter()
.enumerate()
.map(|(i, (off, len, _))| ChunkInfo {
blob_offset: *off,
blob_size: *len,
fdata_offset: (i * tile) as u64,
compressed: false,
uncompressed_size: 0,
checksum: *blake3::hash(&base[i * tile..(i + 1) * tile]).as_bytes(),
chunk_seq: 0,
source: ChunkSource::Delta { base_path: "v/base.bin".into() },
})
.collect();
let entry = Entry { uncompressed_size: base.len() as u64, chunks };
let counting = Counting { bytes: base.clone(), calls: AtomicUsize::new(0) };
let ctx = ReconstructCtx {
archive: &file,
archive_len,
resolve: Some(&counting),
depth: 0,
};
assert_eq!(entry.reconstruct("v2", &ctx, false).unwrap(), base);
assert_eq!(
counting.calls.load(Ordering::SeqCst),
1,
"four delta chunks naming ONE base must resolve it once"
);
let _ = std::fs::remove_dir_all(&dir);
}
/// **What a delta between two `gc` generations is worth, with znippy's OWN
/// encoder.** `#[ignore]`d: a measurement against real packs on disk.
///
/// `ZNIPPY_GEN_BASE` and `ZNIPPY_GEN_TARGETS` (comma-separated) name pack
/// files. Reports the delta size, the ratio, and the encode/decode cost.
#[test]
#[ignore]
fn perf_generation_delta() {
let base_p = match std::env::var("ZNIPPY_GEN_BASE") {
Ok(v) => v,
Err(_) => return,
};
let base = std::fs::read(&base_p).unwrap();
println!("target,base_b,target_b,delta_b,ratio_x,encode_ms,decode_ms,ok");
for t in std::env::var("ZNIPPY_GEN_TARGETS").unwrap().split(',') {
let target = std::fs::read(t).unwrap();
let t0 = std::time::Instant::now();
let d = encode_delta_against(&base, &target);
let enc = t0.elapsed().as_secs_f64() * 1e3;
let t1 = std::time::Instant::now();
let back = apply_delta(&base, &d).unwrap();
let dec = t1.elapsed().as_secs_f64() * 1e3;
let name = std::path::Path::new(t).file_name().unwrap().to_string_lossy();
println!(
"{name},{},{},{},{:.2},{enc:.1},{dec:.1},{}",
base.len(),
target.len(),
d.len(),
target.len() as f64 / d.len() as f64,
back == target
);
}
}
/// **The chunk-chain depth curve.** `#[ignore]`d: it is a measurement, not a
/// verdict. Two shapes are reported because they are not the same cost:
///
/// * a CHAIN — entry N deltas against entry N-1 — which recurses once per
/// link exactly as the per-entry model did;
/// * a FAN — one entry with K delta chunks all against the same base —
/// which resolves that base ONCE thanks to the memo in `reconstruct`.
#[test]
#[ignore]
fn perf_chunk_chain_depth() {
let dir = tmp("cdepth");
let base: Vec<u8> = (0..256_000u32)
.map(|i| (i.wrapping_mul(2654435761) >> 13) as u8)
.collect();
println!("shape,depth_or_k,reconstruct_us,bytes");
for depth in [1usize, 2, 4, 8, 16, 32, 50] {
let path = dir.join(format!("chain{depth}.znippy"));
let mut versions = vec![base.clone()];
for i in 0..depth {
let mut v = versions[i].clone();
v.extend_from_slice(format!("\nedit number {i} appended here\n").as_bytes());
versions.push(v);
}
let deltas: Vec<Vec<u8>> = (0..depth)
.map(|i| encode_delta_against(&versions[i], &versions[i + 1]))
.collect();
let rows = write_base_and_deltas(
&path,
&[("v/0.bin".to_string(), base.clone())],
&deltas,
);
// Each link is its own ENTRY with one delta chunk against the one below.
let mut ar = ZnippyArchive::open(&path).unwrap();
for (i, (off, len, _)) in rows.iter().enumerate() {
ar.file_index.insert(
format!("v/{}.bin", i + 1),
Entry {
uncompressed_size: versions[i + 1].len() as u64,
chunks: vec![ChunkInfo {
blob_offset: *off,
blob_size: *len,
fdata_offset: 0,
compressed: false,
uncompressed_size: 0,
checksum: *blake3::hash(&versions[i + 1]).as_bytes(),
chunk_seq: 0,
source: ChunkSource::Delta { base_path: format!("v/{i}.bin") },
}],
},
);
}
let tip = format!("v/{depth}.bin");
assert_eq!(&ar.extract_file(&tip).unwrap(), versions.last().unwrap());
let n = 20;
let t0 = std::time::Instant::now();
for _ in 0..n {
let _ = ar.extract_file(&tip).unwrap();
}
let us = t0.elapsed().as_secs_f64() * 1e6 / n as f64;
println!("chain,{depth},{us:.1},{}", versions.last().unwrap().len());
}
// FAN: one entry, K delta chunks, all against one base.
let tile = 4096usize;
for k in [1usize, 2, 4, 8, 16, 32, 50] {
let path = dir.join(format!("fan{k}.znippy"));
let target: Vec<u8> = base[..tile * k].to_vec();
let mut payloads = Vec::new();
let mut units = Vec::new();
for i in 0..k {
let off = i * tile;
let mut d = Vec::new();
put_size_varint(&mut d, base.len() as u64);
put_size_varint(&mut d, tile as u64);
emit_copy(&mut d, off as u64, tile as u64);
payloads.push(d.clone());
units.push(StoredUnit {
fdata_offset: off as u64,
payload: UnitPayload::Delta {
base_path: "v/base.bin".into(),
delta: d,
expected: base[off..off + tile].to_vec(),
},
});
}
let rows =
write_base_and_deltas(&path, &[("v/base.bin".to_string(), base.clone())], &payloads);
let mut idx = 0usize;
let mut at = |_b: &[u8]| {
let (o, l, _) = rows[idx];
idx += 1;
(o, l)
};
let entry = entry_from_units(units, &mut at);
let ar = ZnippyArchive::open(&path).unwrap();
assert_eq!(entry.reconstruct("fan", &ar.reconstruct_ctx(0), false).unwrap(), target);
let n = 20;
let t0 = std::time::Instant::now();
for _ in 0..n {
let _ = entry.reconstruct("fan", &ar.reconstruct_ctx(0), false).unwrap();
}
let us = t0.elapsed().as_secs_f64() * 1e6 / n as f64;
println!("fan,{k},{us:.1},{}", target.len());
}
let _ = std::fs::remove_dir_all(&dir);
}
// ── HOL-PERF-3: a whole-body read allocates the body once ─────────────────
/// Bytes allocated on THIS thread (the test binary's allocator counts per thread,
/// so tests running beside this one do not disturb it).
mod counting {
use std::alloc::{GlobalAlloc, Layout, System};
use std::cell::Cell;
thread_local! { pub static ALLOCATED: Cell<u64> = const { Cell::new(0) }; }
pub struct Counting;
unsafe impl GlobalAlloc for Counting {
unsafe fn alloc(&self, l: Layout) -> *mut u8 {
let _ = ALLOCATED.try_with(|c| c.set(c.get() + l.size() as u64));
unsafe { System.alloc(l) }
}
unsafe fn dealloc(&self, p: *mut u8, l: Layout) {
unsafe { System.dealloc(p, l) }
}
unsafe fn realloc(&self, p: *mut u8, l: Layout, new: usize) -> *mut u8 {
let _ = ALLOCATED.try_with(|c| c.set(c.get() + new as u64));
unsafe { System.realloc(p, l, new) }
}
}
pub fn allocated() -> u64 {
ALLOCATED.with(|c| c.get())
}
}
#[global_allocator]
static COUNTING: counting::Counting = counting::Counting;
/// **A whole-body read allocates the body once** (holger HOL-PERF-3). `reconstruct`
/// grew its result from `Vec::new()` and read every stored chunk into a scratch
/// buffer first: holger measured a 64 MiB crate allocating 231 MiB and peaking at
/// 117 MiB live per download. The result is now reserved (bounded by the archive's
/// real length, never by the untrusted index alone) and a stored chunk is read
/// straight into it. A 10.5 MiB incompressible file in 1 MiB chunks: at most its own
/// size plus one chunk.
#[test]
fn a_whole_body_read_allocates_the_body_once() {
let dir = tmp("reserve");
let path = dir.join("a.znippy");
let len = 10 * 1024 * 1024 + 512 * 1024;
let mut x: u64 = 0x9E37_79B9_7F4A_7C15;
let body: Vec<u8> = (0..len)
.map(|_| {
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
x as u8
})
.collect();
write_archive_multichunk(&path, &[("big.bin".to_string(), body.clone())], 1024 * 1024);
let archive = ZnippyArchive::open(&path).unwrap();
for verify in [false, true] {
let before = counting::allocated();
let got = if verify { archive.extract_file_verified("big.bin").unwrap() } else { archive.extract_file("big.bin").unwrap() };
let took = counting::allocated() - before;
assert!(got == body, "the bytes round-trip (verify {verify})");
assert!(
took <= len as u64 + 1024 * 1024 + 64 * 1024,
"a {len} byte read allocated {took} bytes (verify {verify}): {:.2}x its size",
took as f64 / len as f64
);
}
let _ = std::fs::remove_dir_all(&dir);
}
// ── gatwasm: a frame may not decode to more than its chunk declares ───────
/// Write a one-entry archive whose single chunk stores `blob` verbatim, with
/// the index declaring `declared` bytes and `checksum` for it. Lets a test put
/// a hostile OpenZL frame behind an honest-looking index row.
fn write_raw_chunk_archive(path: &Path, rel: &str, blob: &[u8], declared: u64, checksum: [u8; 32]) {
let file = Arc::new(File::create(path).unwrap());
file.write_all_at(blob, 0).unwrap();
let blobs = vec![BlobMeta {
blob_offset: 0,
blob_size: blob.len() as u64,
chunk_meta: ChunkMeta {
fdata_offset: 0,
file_index: 0,
chunk_seq: 0,
checksum,
compressed: true,
uncompressed_size: declared,
compressed_size: blob.len() as u64,
},
}];
let name = rel.to_string();
let batch = build_metadata_batch(&blobs, move |_| name.clone(), &[], &[]).unwrap();
let schema = lookup_schema();
let mut sink = ArrowIpcSink::new(file.clone(), blob.len() as u64);
sink.push_subindex(
schema.as_ref(),
&[batch],
GroupKey { pkg_type: 0, repo: String::new(), module_name: String::new() },
)
.unwrap();
Box::new(sink).finish().unwrap();
}
/// **A frame whose header claims more than its chunk declares is refused
/// before its output buffer exists.** An 8 KiB OpenZL frame of 64 MiB of zeros
/// sits behind an index row that declares 12 bytes. The reader used to size
/// its output from the frame header alone, so the row's 12 bytes bought a
/// 64 MiB zero-filled allocation (and a 40-byte frame claiming terabytes
/// bought an abort). Both read paths must refuse it by name and allocate
/// next to nothing.
#[test]
fn a_frame_larger_than_its_chunk_is_refused_without_allocating_it() {
let dir = tmp("framecap");
let path = dir.join("f.znippy");
let claim = 64usize << 20;
let frame = CompressCtx::new(3).unwrap().compress(&vec![0u8; claim]).unwrap();
assert!(frame.len() < 64 * 1024, "zeros compress to a small frame ({} bytes)", frame.len());
write_raw_chunk_archive(&path, "small.bin", &frame, 12, *blake3::hash(b"hello znippy").as_bytes());
let archive = ZnippyArchive::open(&path).unwrap();
for verify in [false, true] {
let before = counting::allocated();
let got = if verify { archive.extract_file_verified("small.bin") } else { archive.extract_file("small.bin") };
let took = counting::allocated() - before;
let err = got.expect_err("a frame that decodes past its chunk must be refused");
assert!(
err.to_string().contains("declares"),
"the refusal names the size lie (verify {verify}): {err}"
);
assert!(took < 1 << 20, "the refusal allocated {took} bytes (verify {verify}); the frame claims {claim}");
}
let _ = std::fs::remove_dir_all(&dir);
}
/// The heimdall fuzz find (lane zlwasm): a 36-byte frame whose header
/// declares 0 bytes and whose `constant` codec asks OpenZL for ~2^64 bytes.
/// Natively malloc says no and OpenZL reports an allocation error; it must
/// stay a refusal on both read paths, never a crash.
#[test]
fn the_constant_stream_fuzz_frame_is_refused() {
let dir = tmp("zlregress");
let path = dir.join("r.znippy");
let frame = include_bytes!("../tests/fixtures/openzl/regress-constant-stream-alloc.zl");
write_raw_chunk_archive(&path, "r.bin", frame, 0, [0u8; 32]);
let archive = ZnippyArchive::open(&path).unwrap();
assert!(archive.extract_file("r.bin").is_err());
assert!(archive.extract_file_verified("r.bin").is_err());
let _ = std::fs::remove_dir_all(&dir);
}
}