znippy_common/

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::ipc::reader::StreamReader;
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,
    /// **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>> {
        // Grown from verified bytes, never pre-sized from `uncompressed_size`:
        // 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).
        let mut result: Vec<u8> = Vec::new();
        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()
                    ));
                }
            }
            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 {
                codec::decompress_into(&blob, &mut decomp)?;
                &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)
    }
}

/// 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,
                    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 = StreamReader::try_new(std::io::Cursor::new(bytes), None)
            .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).
    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,
                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,
                    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,
                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,
                    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,
                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,
                            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);
    }
}