znippy_common/

meta_sink_append.rs

//! `ArrowIpcSinkAppend` — the **append/resume-capable v2 clone** of
//! [`ArrowIpcSink`](crate::ArrowIpcSink).
//!
//! ## Why a clone and not a refactor
//! The original [`ArrowIpcSink`] is the A baseline of the arrow-ipc write/seal
//! A/B audit — its hot path (`push_subindex` + `finish`) must stay byte- and
//! perf-identical. So this is a *near-identical copy* that adds the two new
//! capabilities **without touching the original**:
//!
//!  * **fresh path** — [`ArrowIpcSinkAppend::new`] + `push_subindex` + `finish`
//!    is a line-for-line clone of `ArrowIpcSink`. It writes the **same v0.7
//!    on-disk bytes** (same sub-index serialisation, same sorted lookup, same
//!    fst trie, same manifest, same `ZNPYMIDX` footer). The A/B parity test
//!    proves this clone is a zero-cost superset on the normal write path.
//!
//!  * **resume/append path** — [`ArrowIpcSinkAppend::open_existing`] reopens an
//!    already-sealed `.znippy`, recovers its existing data rows, repositions the
//!    cursor at the **end of the blob region** (truncating the old metadata
//!    tail), and lets the caller `push_subindex` more blobs' rows. `finish()`
//!    then re-seals the merged old+new row set — the **first-class native blob
//!    append** that the iceberg lifecycle test (#23) previously had to do by
//!    hand.
//!
//! ### The resume mechanism, in bytes
//! A sealed v0.7 archive is:
//! ```text
//! [ blob_0 … blob_N ][ data sub-idx(es) ][ lookup sub-idx ][ trie ][ manifest ][ ZNPYMIDX ][off]
//! ^0                 ^blob_end           (the whole metadata tail is rebuildable)
//! ```
//! To append we must:
//!   1. read the full manifest (incl. reserved lookup/trie entries),
//!   2. find `blob_end` = the lowest `index_offset` over all sub-index entries
//!      (where the blob region stops and the rebuildable tail begins),
//!   3. recover the existing **data** rows from the sorted lookup sub-index (one
//!      cheap read of the already-sorted reserved section — no per-sub-index
//!      re-scan), seeding the lookup accumulator,
//!   4. set the cursor to `blob_end` so the caller's new blob bytes + new data
//!      sub-index overwrite the old (now-stale) metadata tail,
//!   5. on `finish()`, re-sort the merged rows and re-emit lookup + trie +
//!      manifest + footer.
//!
//! The new blob bytes are written by the caller (the compress pipeline / a
//! library append entrypoint such as [`append_files`]) to the file at
//! `blob_end()` exactly as a fresh archive
//! writes blobs at offset 0; this sink owns only the metadata tail, identical to
//! the original's contract.

use std::fs::File;
use std::os::unix::fs::FileExt;
use std::path::Path;
use std::sync::Arc;

use anyhow::{Result, anyhow};
use arrow::array::{
    BooleanArray, BooleanBuilder, FixedSizeBinaryArray, FixedSizeBinaryBuilder, StringArray,
    StringBuilder, UInt32Array, UInt32Builder, UInt64Array, UInt64Builder,
};
use arrow::datatypes::Schema;
use arrow::ipc::writer::StreamWriter;
use arrow::record_batch::RecordBatch;

use crate::index::{
    ChunkLoc, LOOKUP_MODULE, MULTI_INDEX_MAGIC, ManifestEntry, RESERVED_PKG_TYPE, TRIE_MODULE,
    data_subindex_schema, is_reserved_module, lookup_schema, read_znippy_full_manifest,
    write_manifest_bytes,
};
use crate::index::{
    CARRIED_RESERVED_MODULES, META_MODULE, ZNIPPY_DELTA_MODULE, read_reserved_section_bytes,
};
use crate::meta_index::{
    MetaTable, build_meta_batch, decode_meta_section, meta_schema,
};
use crate::meta_sink::{ArchiveMetaSink, GroupKey};

/// Append/resume-capable v2 clone of [`ArrowIpcSink`](crate::ArrowIpcSink).
///
/// Field-for-field identical to the original; the only added surface is the
/// [`open_existing`](Self::open_existing) constructor and the [`blob_end`](Self::blob_end)
/// accessor. The fresh-write path is byte-identical to the original.
pub struct ArrowIpcSinkAppend {
    file: Arc<File>,
    cursor: u64,
    entries: Vec<ManifestEntry>,
    lookup_paths: Vec<String>,
    lookup_locs: Vec<ChunkLoc>,
    /// On resume, the recovered pre-existing data rows. They are re-emitted as a
    /// data sub-index in `finish()` (so the ordinary reader, which reads data
    /// sub-indexes, still sees them) AND merged into the rebuilt lookup. Empty
    /// for a fresh sink. Kept separate from `lookup_*` so they aren't
    /// double-counted before the re-emit.
    carried: Vec<(String, ChunkLoc)>,
    /// Searchable metadata to seal as the `META_MODULE` sub-index.
    ///
    /// `None` means "emit no section", which is what a reader later reports as
    /// `ArchiveMeta::NoMetadata`; `Some(empty)` means "emit a present, empty
    /// index". The two are different archives on disk and different answers on
    /// read, and this field is the only place that decision is made.
    ///
    /// On resume this is **seeded from the archive's existing section**, because
    /// `open_existing` truncates the whole metadata tail — without carrying it,
    /// every append would silently erase the metadata the archive already had.
    meta: Option<MetaTable>,
    /// Reserved sections carried forward verbatim across an append, as
    /// `(module_name, raw bytes)`.
    ///
    /// The same failure `meta` above fixes, for the reserved sections that are
    /// **independent logs** rather than derivations — see
    /// [`CARRIED_RESERVED_MODULES`]. MEASURED 2026-08-04: an object-carrying
    /// push dropped `__gunnar_refs__` (144 928 -> 146 396 bytes, section gone),
    /// because this sink re-supplies no reserved section it was not handed and
    /// nothing hands it one.
    ///
    /// Derived sections are deliberately NOT carried: when the objects change,
    /// `__gunnar_graph__` / `__gunnar_reach__` / `__gunnar_oid__` are wrong, and
    /// carrying a stale one forward is worse than dropping it.
    carried_reserved: Vec<(String, Vec<u8>)>,
    /// The delta map, as rows: `(relative_path, chunk_seq, base_path)`.
    ///
    /// Decoded rather than carried raw, because an append may ADD rows to it and
    /// the manifest cannot hold two sections under one module name. Same reason
    /// `meta` travels as a `MetaTable`.
    delta_map: Vec<(String, u32, String)>,
}

impl ArrowIpcSinkAppend {
    /// Fresh archive: identical to [`ArrowIpcSink::new`](crate::ArrowIpcSink::new).
    /// `blob_end_offset` is the byte offset just past the last blob.
    pub fn new(file: Arc<File>, blob_end_offset: u64) -> Self {
        Self {
            file,
            cursor: blob_end_offset,
            entries: Vec::new(),
            lookup_paths: Vec::new(),
            lookup_locs: Vec::new(),
            carried: Vec::new(),
            meta: None,
            carried_reserved: Vec::new(),
            delta_map: Vec::new(),
        }
    }

    /// Seal `meta` as this archive's searchable metadata sub-index, replacing
    /// anything carried from a resumed archive.
    pub fn with_meta(mut self, meta: MetaTable) -> Self {
        self.meta = Some(meta);
        self
    }

    /// Add rows to the metadata to be sealed, keeping whatever a resume carried.
    /// Creates the section if the archive had none.
    pub fn merge_meta(&mut self, rows: impl IntoIterator<Item = crate::meta_index::MetaEntry>) {
        self.meta.get_or_insert_with(MetaTable::new).extend(rows);
    }

    /// The metadata this sink will seal — `None` when it will emit no section.
    pub fn meta(&self) -> Option<&MetaTable> {
        self.meta.as_ref()
    }

    /// Reopen an **already-sealed** v0.7 `.znippy` for append/resume.
    ///
    /// Recovers the existing data rows from the sorted lookup sub-index, drops
    /// the (rebuildable) metadata tail by positioning the cursor at the end of
    /// the blob region, and returns a sink ready to accept more `push_subindex`
    /// calls. The caller appends its new blob bytes to the same file starting at
    /// [`blob_end`](Self::blob_end) before pushing the matching index rows.
    ///
    /// The file is opened read+write; nothing is mutated until `push_subindex` /
    /// `finish` overwrite the old tail.
    pub fn open_existing(path: &Path) -> Result<Self> {
        let file = Arc::new(
            std::fs::OpenOptions::new()
                .read(true)
                .write(true)
                .open(path)
                .map_err(|e| anyhow!("append: open {} for resume: {e}", path.display()))?,
        );

        // 1. Full manifest (incl. reserved lookup/trie entries).
        let (entries, _manifest_offset) = read_znippy_full_manifest(path)?;
        if entries.is_empty() {
            return Err(anyhow!("append: archive {} has an empty manifest", path.display()));
        }

        // 2. blob_end = lowest index_offset over all sub-index/reserved sections.
        //    Everything from there to EOF is the rebuildable metadata tail.
        let blob_end = entries
            .iter()
            .map(|e| e.index_offset)
            .min()
            .ok_or_else(|| anyhow!("append: no sections in manifest"))?;

        // 3. Recover existing DATA rows from the sorted lookup sub-index (one
        //    read of the already-sorted reserved section). Fall back to scanning
        //    the data sub-indexes if (unexpectedly) no lookup section is present.
        let (paths, locs) = recover_rows(path, &entries)?;
        let carried: Vec<(String, ChunkLoc)> = paths.into_iter().zip(locs).collect();

        // 4. Carry the searchable metadata section forward. The tail we are about
        //    to overwrite contains it, so a resume that did not recover it would
        //    quietly turn an archive WITH metadata into one without — and the
        //    reader would then honestly report `NoMetadata` about an archive that
        //    used to have some. Absent stays absent; present-but-empty stays
        //    present-but-empty.
        let meta = match read_reserved_section_bytes(path, META_MODULE)? {
            None => None,
            Some(bytes) => Some(decode_meta_section(&bytes)?.to_table()),
        };

        // 5. Carry the INDEPENDENT reserved sections forward verbatim. The tail
        //    about to be overwritten holds them, and nothing else in this process
        //    can reproduce them: they are per-push logs, not derivations of the
        //    blobs. Derived sections are left to be rebuilt.
        let mut carried_reserved = Vec::new();
        for module in CARRIED_RESERVED_MODULES {
            if let Some(bytes) = read_reserved_section_bytes(path, module)? {
                carried_reserved.push(((*module).to_string(), bytes));
            }
        }

        // 6. The delta map, decoded. Dropping it would turn every delta chunk in
        //    the archive back into a stored chunk on the next append, and the
        //    reader would then hand a delta's instruction stream to a caller as
        //    file content.
        let delta_map = read_delta_map(path)?;

        Ok(Self {
            file,
            cursor: blob_end,
            entries: Vec::new(), // rebuilt fresh by push_subindex + finish
            lookup_paths: Vec::new(),
            lookup_locs: Vec::new(),
            carried,
            meta,
            carried_reserved,
            delta_map,
        })
    }

    /// Byte offset where the blob region ends in a resumed archive — where the
    /// caller writes its newly-appended blob bytes (and where the first new data
    /// sub-index will be placed). For a fresh sink this is the `blob_end_offset`
    /// passed to [`new`](Self::new) until the first `push_subindex`.
    pub fn blob_end(&self) -> u64 {
        self.cursor
    }

    /// Number of pre-existing data rows recovered on resume (0 for a fresh sink).
    pub fn recovered_rows(&self) -> usize {
        self.carried.len()
    }

    /// Drop every carried (pre-existing) row whose `relative_path` is about to be
    /// re-written by this append, giving last-writer-wins **replace** semantics.
    /// Returns the number of rows dropped.
    ///
    /// Without this, appending a path the archive already contains left TWO row
    /// sets for it in the re-sealed index — the stale one and the new one — and
    /// nothing downstream treated that as an error: the reader that concatenates
    /// chunks returned both copies back to back at twice the real length, and the
    /// reader that places chunks at `fdata_offset` wrote both to offset 0, so the
    /// carried STALE copy (re-emitted last, in `finish`) won and `znippy get`
    /// silently handed back the old file. Every chunk's blake3 is individually
    /// correct in both cases, so even verified reads passed.
    fn drop_carried_paths(&mut self, replacing: &std::collections::HashSet<&str>) -> usize {
        if self.carried.is_empty() || replacing.is_empty() {
            return 0;
        }
        let before = self.carried.len();
        self.carried.retain(|(p, _)| !replacing.contains(p.as_str()));
        before - self.carried.len()
    }

    /// Re-emit the carried (recovered) rows as a single data sub-index so the
    /// ordinary reader — which reads data sub-indexes, not the lookup — still
    /// lists them after the re-seal. `push_subindex` also folds them into the
    /// rebuilt lookup accumulator. No-op for a fresh sink.
    fn emit_carried(&mut self) -> Result<()> {
        if self.carried.is_empty() {
            return Ok(());
        }
        let carried = std::mem::take(&mut self.carried);
        let (paths, locs): (Vec<String>, Vec<ChunkLoc>) = carried.into_iter().unzip();
        let batch = base_batch_from_rows(&paths, &locs)?;
        self.push_subindex(data_subindex_schema().as_ref(), &[batch], GroupKey {
            pkg_type: 0,
            repo: String::new(),
            module_name: String::new(),
        })
    }

    // ── below: a line-for-line clone of ArrowIpcSink's private machinery ──

    fn accumulate_lookup(&mut self, batch: &RecordBatch) {
        let cols = (|| {
            Some((
                batch.column_by_name("relative_path")?.as_any().downcast_ref::<StringArray>()?,
                batch.column_by_name("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()?,
                batch.column_by_name("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("compressed")?.as_any().downcast_ref::<BooleanArray>()?,
                batch.column_by_name("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("blob_offset")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("blob_size")?.as_any().downcast_ref::<UInt64Array>()?,
                batch.column_by_name("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()?,
            ))
        })();
        let Some((paths, chunk_seq, fdata, compressed, usz, blob_off, blob_sz, checksum)) = cols
        else { return; };
        for i in 0..batch.num_rows() {
            let mut ck = [0u8; 32];
            ck.copy_from_slice(checksum.value(i));
            self.lookup_paths.push(paths.value(i).to_string());
            self.lookup_locs.push(ChunkLoc {
                chunk_seq: chunk_seq.value(i),
                fdata_offset: fdata.value(i),
                blob_offset: blob_off.value(i),
                blob_size: blob_sz.value(i),
                uncompressed_size: usz.value(i),
                compressed: compressed.value(i),
                checksum: ck,
            });
        }
    }

    fn write_lookup_and_trie(&mut self) -> Result<()> {
        let n = self.lookup_paths.len();
        let mut order: Vec<usize> = (0..n).collect();
        order.sort_by(|&a, &b| {
            self.lookup_paths[a].cmp(&self.lookup_paths[b])
                .then(self.lookup_locs[a].chunk_seq.cmp(&self.lookup_locs[b].chunk_seq))
        });

        let schema = lookup_schema();
        let batch = base_batch_permuted(
            schema.clone(),
            &self.lookup_paths,
            &self.lookup_locs,
            &order,
        )?;
        self.push_subindex(&schema, &[batch], GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: LOOKUP_MODULE.to_string(),
        })?;

        let mut builder = fst::MapBuilder::memory();
        let mut prev: Option<&str> = None;
        for (sorted_idx, &orig) in order.iter().enumerate() {
            let p = self.lookup_paths[orig].as_str();
            if prev != Some(p) {
                builder.insert(p.as_bytes(), sorted_idx as u64)
                    .map_err(|e| anyhow!("trie insert: {e}"))?;
                prev = Some(p);
            }
        }
        let trie_bytes = builder.into_inner().map_err(|e| anyhow!("trie finish: {e}"))?;
        self.write_raw_section(&trie_bytes, GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: TRIE_MODULE.to_string(),
        })
    }

    /// Emit the searchable metadata sub-index, when there is one to emit.
    ///
    /// Reserved module, so the data readers skip it and an older znippy simply
    /// ignores the entry. `None` writes NOTHING — that absence is exactly what
    /// `ArchiveMeta::NoMetadata` reports, and it is why an archive sealed without
    /// metadata stays byte-identical to one sealed before this module existed.
    fn write_meta_subindex(&mut self) -> Result<()> {
        let Some(table) = self.meta.take() else {
            return Ok(());
        };
        let batch = build_meta_batch(&table)?;
        let schema = meta_schema();
        self.push_subindex(schema.as_ref(), &[batch], GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: META_MODULE.to_string(),
        })
    }

    /// Re-emit the reserved sections a resume carried, byte for byte.
    ///
    /// After `write_meta_subindex`, so a `__meta__` section this sink was handed
    /// wins over anything else; the carried list never contains `__meta__`,
    /// which travels as a decoded `MetaTable` instead.
    fn write_carried_reserved(&mut self) -> Result<()> {
        for (module, bytes) in std::mem::take(&mut self.carried_reserved) {
            self.write_raw_section(&bytes, GroupKey {
                pkg_type: RESERVED_PKG_TYPE,
                repo: String::new(),
                module_name: module,
            })?;
        }
        Ok(())
    }

    /// Seal the delta map, if there is one.
    fn write_delta_map(&mut self) -> Result<()> {
        if self.delta_map.is_empty() {
            return Ok(());
        }
        let rows = std::mem::take(&mut self.delta_map);
        let paths = StringArray::from(rows.iter().map(|r| r.0.as_str()).collect::<Vec<_>>());
        let seqs = UInt32Array::from(rows.iter().map(|r| r.1).collect::<Vec<_>>());
        let bases = StringArray::from(rows.iter().map(|r| r.2.as_str()).collect::<Vec<_>>());
        let schema = crate::index::delta_map_schema();
        let batch = RecordBatch::try_new(
            Arc::clone(&schema),
            vec![Arc::new(paths), Arc::new(seqs), Arc::new(bases)],
        )
        .map_err(|e| anyhow!("delta map batch: {e}"))?;
        self.push_subindex(schema.as_ref(), &[batch], GroupKey {
            pkg_type: RESERVED_PKG_TYPE,
            repo: String::new(),
            module_name: ZNIPPY_DELTA_MODULE.to_string(),
        })
    }

    /// The archive file, for a caller appending blob bytes at [`blob_end`](Self::blob_end).
    pub fn file(&self) -> &Arc<File> {
        &self.file
    }

    /// Move the blob cursor on after a caller wrote `n` bytes at `blob_end`.
    pub fn advance_blob_end(&mut self, n: u64) {
        self.cursor += n;
    }

    /// Replace every carried row of `path` with `locs`.
    ///
    /// Used to turn a stored entry into a delta entry in place: its old chunk
    /// rows go, one delta row arrives. Rows of other paths are untouched.
    pub fn replace_carried(&mut self, path: &str, locs: Vec<ChunkLoc>) {
        self.carried.retain(|(p, _)| p != path);
        for loc in locs {
            self.carried.push((path.to_string(), loc));
        }
    }

    /// Record that `(path, chunk_seq)` is a delta against `base`.
    pub fn push_delta_map_row(&mut self, path: String, chunk_seq: u32, base: String) {
        self.delta_map.retain(|(p, s, _)| !(p == &path && *s == chunk_seq));
        self.delta_map.push((path, chunk_seq, base));
    }

    fn write_raw_section(&mut self, bytes: &[u8], key: GroupKey) -> Result<()> {
        let start = self.cursor;
        self.file.write_all_at(bytes, start)?;
        self.cursor += bytes.len() as u64;
        self.entries.push(ManifestEntry {
            pkg_type: key.pkg_type,
            repo: key.repo,
            module_name: key.module_name,
            index_offset: start,
            index_len: bytes.len() as u64,
            row_count: 0,
        });
        Ok(())
    }
}

impl ArchiveMetaSink for ArrowIpcSinkAppend {
    fn push_subindex(
        &mut self,
        schema: &Schema,
        batches: &[RecordBatch],
        key: GroupKey,
    ) -> Result<()> {
        let sub_start = self.cursor;
        let mut sub_bytes: Vec<u8> = Vec::new();
        let mut sw = StreamWriter::try_new(&mut sub_bytes, schema)
            .map_err(|e| anyhow!("sub-index writer: {e}"))?;
        let mut row_count = 0u64;
        for batch in batches {
            row_count += batch.num_rows() as u64;
            sw.write(batch).map_err(|e| anyhow!("sub-index write: {e}"))?;
        }
        sw.finish().map_err(|e| anyhow!("sub-index finish: {e}"))?;

        // Accumulate base columns for the lookup layer from DATA sub-indexes only.
        // Widened from "not lookup, not trie" to "not reserved" when META_MODULE
        // arrived: the metadata sub-index is Arrow IPC and does come through here,
        // and its rows are key/value facts, not chunk locations — folding them
        // into the lookup would corrupt random access. Behaviour for every
        // pre-existing module is unchanged (sign sections are raw, never pushed).
        if !is_reserved_module(&key.module_name) {
            for batch in batches {
                self.accumulate_lookup(batch);
            }
        }

        let sub_len = sub_bytes.len() as u64;
        self.file.write_all_at(&sub_bytes, sub_start)?;
        self.cursor += sub_len;

        self.entries.push(ManifestEntry {
            pkg_type: key.pkg_type,
            repo: key.repo,
            module_name: key.module_name,
            index_offset: sub_start,
            index_len: sub_len,
            row_count,
        });
        Ok(())
    }

    fn finish(mut self: Box<Self>) -> Result<u64> {
        // Resume: re-emit recovered rows as a data sub-index (the ordinary reader
        // reads data sub-indexes, not the lookup). No-op on the fresh path.
        self.emit_carried()?;
        self.write_lookup_and_trie()?;
        self.write_meta_subindex()?;
        self.write_carried_reserved()?;
        self.write_delta_map()?;

        let manifest_offset = self.cursor;
        let manifest_bytes =
            write_manifest_bytes(&self.entries).map_err(|e| anyhow!("manifest: {e}"))?;
        self.file.write_all_at(&manifest_bytes, manifest_offset)?;

        let after = manifest_offset + manifest_bytes.len() as u64;
        self.file.write_all_at(&MULTI_INDEX_MAGIC, after)?;
        self.file.write_all_at(
            &manifest_offset.to_le_bytes(),
            after + MULTI_INDEX_MAGIC.len() as u64,
        )?;
        // Resume overwrites the old (longer-or-shorter) tail in place; if the new
        // tail is shorter than the old one, truncate so no stale footer lingers.
        let final_len = after + MULTI_INDEX_MAGIC.len() as u64 + 8;
        self.file.set_len(final_len)?;
        self.file.sync_all()?;

        Ok(final_len)
    }
}


/// Read the delta map out of an archive, as rows. Empty when the section is
/// absent, which is every archive that holds no delta chunk.
///
/// **Public because the chain is the caller's to bound.**
/// [`supersede_as_delta`] takes `chain_depth` — how many links already sit
/// behind the base — and refuses past [`MAX_GENERATION_CHAIN`]. A caller that
/// supersedes one generation at a time (which is what a repository `gc` does)
/// cannot supply that number from what it did in this process: the chain was
/// built by earlier runs and the archive is the only record of it. Without this
/// the caller must keep a sidecar counter, and a counter that drifts low walks
/// the chain past the cap the writer exists to enforce.
pub fn read_delta_map(path: &Path) -> Result<Vec<(String, u32, String)>> {
    use arrow::ipc::reader::StreamReader;
    let Some(bytes) = read_reserved_section_bytes(path, ZNIPPY_DELTA_MODULE)? else {
        return Ok(Vec::new());
    };
    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)
        .map_err(|e| anyhow!("delta map: {e}"))?;
    let mut out = Vec::new();
    for batch in reader {
        let batch = batch.map_err(|e| anyhow!("delta map batch: {e}"))?;
        let paths = batch
            .column_by_name("relative_path")
            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
            .ok_or_else(|| anyhow!("delta map: missing relative_path"))?;
        let seqs = batch
            .column_by_name("chunk_seq")
            .and_then(|c| c.as_any().downcast_ref::<UInt32Array>())
            .ok_or_else(|| anyhow!("delta map: missing chunk_seq"))?;
        let bases = batch
            .column_by_name("base_path")
            .and_then(|c| c.as_any().downcast_ref::<StringArray>())
            .ok_or_else(|| anyhow!("delta map: missing base_path"))?;
        for r in 0..batch.num_rows() {
            out.push((paths.value(r).to_string(), seqs.value(r), bases.value(r).to_string()));
        }
    }
    Ok(out)
}


/// **The writer: re-store a superseded entry as a delta against a live one.**
///
/// Given an archive holding generation N and generation N+1, this replaces N's
/// stored bytes with a delta against N+1 and records the reference in
/// [`ZNIPPY_DELTA_MODULE`]. N's old blob bytes become dead space in the file;
/// they are reclaimed by a rewrite, not by this call.
///
/// # Why the OLD one is delta'd against the NEW one, and never the other way
///
/// The obvious direction is forward — store N+1 as a delta against N — and it is
/// wrong for gunnar, for a measured reason that is not znippy's. The current
/// generation is what `P-001`'s `write_pack_copy` copies entries out of on every
/// clone and fetch; putting it behind a delta chain would move it onto the
/// entry-copy path, measured at 18x the bytes and 221x the CPU of the path that
/// avoids it.
///
/// Delta-ing backwards keeps the property that matters: **the live generation is
/// always a whole entry, at chain depth zero**, whatever the archive's history.
/// The k-th-oldest generation sits at depth k, and depth there is free because
/// nothing on the serve path reads a superseded generation — the live pack holds
/// every reachable object.
///
/// # The chain bound
///
/// A repository `gc`'d weekly reaches [`MAX_RECONSTRUCT_DEPTH`] in fifteen
/// months, and the reader would then refuse the oldest generation rather than
/// serve it slowly. So the writer bounds itself first: past
/// [`MAX_GENERATION_CHAIN`] it declines and returns
/// [`SupersedeOutcome::ChainTooLong`], and the caller keeps that generation
/// whole. A whole generation is a keyframe: the chain restarts from it and every
/// generation is reachable in at most `MAX_GENERATION_CHAIN` links, for ever.
/// That is a policy decided here rather than a limit discovered in production.
///
/// `chain_depth` is how many links the base already sits behind a whole entry —
/// the caller knows this because it wrote them. Zero for a base that is whole.
pub fn supersede_as_delta(
    archive: &Path,
    superseded: &str,
    base: &str,
    chain_depth: usize,
    compression_level: i32,
) -> Result<SupersedeOutcome> {
    if superseded == base {
        return Err(anyhow!("an entry cannot be a delta against itself: {superseded}"));
    }
    if chain_depth + 1 > MAX_GENERATION_CHAIN {
        return Ok(SupersedeOutcome::ChainTooLong);
    }

    let (old_bytes, base_bytes) = {
        let ar = crate::ZnippyArchive::open(archive)?;
        // VERIFIED reads: the bytes about to become a checksum and a delta base
        // must be the bytes the archive claims, or the delta is correct against
        // something nobody stored.
        (
            ar.extract_file_verified(superseded)?,
            ar.extract_file_verified(base)?,
        )
    };

    let delta = crate::archive::encode_delta_against(&base_bytes, &old_bytes);
    // The same two cutoffs `plan_version` applies, on real bytes: a delta that
    // does not clearly win is not worth a chain link that is paid on every later
    // read of this entry.
    if (delta.len() as f64) >= crate::archive::DELTA_SIZE_ALPHA * (old_bytes.len() as f64) {
        return Ok(SupersedeOutcome::NotSmaller {
            delta_bytes: delta.len() as u64,
            stored_bytes: old_bytes.len() as u64,
        });
    }

    let mut sink = ArrowIpcSinkAppend::open_existing(archive)?;
    let at = sink.blob_end();
    // A delta between two packs is not obviously compressible — pack bytes are
    // already deflated, and a diff of them mostly is too. So the codec is
    // consulted and its answer taken, exactly as the append path does, rather
    // than a rule being asserted either way (`precompressed.rs`).
    let mut ctx = crate::codec::CompressCtx::new(compression_level)?;
    let frame = ctx.compress(&delta).ok();
    let (on_disk, compressed): (&[u8], bool) = match frame.as_deref() {
        Some(f) if f.len() < delta.len() => (f, true),
        _ => (&delta, false),
    };
    sink.file().write_all_at(on_disk, at)?;
    sink.advance_blob_end(on_disk.len() as u64);

    // Drop every old row of the superseded entry, and put one delta row in.
    sink.replace_carried(superseded, vec![ChunkLoc {
        chunk_seq: 0,
        fdata_offset: 0,
        blob_offset: at,
        blob_size: on_disk.len() as u64,
        // What the entry reconstructs TO, which is what the reader sizes and
        // what the chunk's blake3 is over.
        uncompressed_size: old_bytes.len() as u64,
        compressed,
        checksum: *blake3::hash(&old_bytes).as_bytes(),
    }]);
    sink.push_delta_map_row(superseded.to_string(), 0, base.to_string());
    Box::new(sink).finish()?;

    Ok(SupersedeOutcome::Delta {
        stored_bytes: old_bytes.len() as u64,
        delta_bytes: on_disk.len() as u64,
        chain_depth: chain_depth + 1,
    })
}

/// What one [`compact_archive`] reclaimed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CompactReport {
    pub bytes_before: u64,
    pub bytes_after: u64,
    /// Live data rows carried across. A compaction never changes this.
    pub rows: u64,
    /// Delta-map rows carried across. Also never changed.
    pub delta_rows: u64,
}

/// **Rewrite a sealed archive keeping only the bytes something still points at.**
///
/// The other half of [`supersede_as_delta`], and without it that writer cannot
/// deliver a byte. Superseding an entry writes the delta and drops the old
/// entry's index rows, but the old *blob* stays where it was, unreferenced —
/// MEASURED on oden 2026-08-05: an 8 056 624-byte archive went to **8 057 731**
/// after a supersede replaced 4 000 000 bytes of live payload with 11. An archive
/// fed one generation at a time therefore grows by a whole generation each time,
/// which is exactly what storing them whole would have cost. The saving
/// `supersede_as_delta` measures is in the archive's LIVE bytes; this is what
/// makes it the archive's size.
///
/// # It copies, and that is the point
///
/// Every live chunk's on-disk bytes are copied **verbatim** to the new file — no
/// decode, no re-encode, no delta re-computation, and the `compressed` flag,
/// `uncompressed_size` and blake3 of each chunk are carried unchanged. A delta
/// chunk stays a delta chunk at the same depth, and `__znippy_delta__` travels as
/// its decoded rows exactly as an append carries it. So a compaction costs one
/// pass over the live bytes and cannot change what any entry reads back as.
///
/// The obvious alternative — read every entry out and write a fresh archive —
/// cannot work here at all, and that is worth recording because it is what a
/// caller reaches for first: rebuilding stores every entry WHOLE, and superseding
/// them again reproduces exactly the dead payload the rebuild was meant to
/// remove.
///
/// Staged beside the destination and renamed over it, so an interruption at any
/// byte leaves the original archive, dead payload and all, serving.
pub fn compact_archive(archive: &Path) -> Result<CompactReport> {
    let bytes_before = std::fs::metadata(archive)?.len();
    let src = ArrowIpcSinkAppend::open_existing(archive)?;

    let staged = {
        let unique = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let mut p = archive.as_os_str().to_owned();
        p.push(format!(".compact-{}-{unique}", std::process::id()));
        std::path::PathBuf::from(p)
    };
    let out = Arc::new(
        std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(true)
            .open(&staged)
            .map_err(|e| anyhow!("compact: staging {}: {e}", staged.display()))?,
    );

    let mut sink = ArrowIpcSinkAppend::new(Arc::clone(&out), 0);
    sink.meta = src.meta.clone();
    sink.carried_reserved = src.carried_reserved.clone();
    sink.delta_map = src.delta_map.clone();
    let delta_rows = sink.delta_map.len() as u64;

    let mut cursor = 0u64;
    // Buffer reused across chunks: a repository's generations are packs, and
    // allocating one per chunk is the kind of thing that turns a copy into a
    // profile.
    let mut buf: Vec<u8> = Vec::new();
    for (path, loc) in &src.carried {
        let n = loc.blob_size as usize;
        buf.clear();
        buf.resize(n, 0);
        src.file
            .read_exact_at(&mut buf, loc.blob_offset)
            .map_err(|e| anyhow!("compact: reading {path} at {}: {e}", loc.blob_offset))?;
        out.write_all_at(&buf, cursor)?;
        let mut moved = loc.clone();
        moved.blob_offset = cursor;
        cursor += loc.blob_size;
        sink.carried.push((path.clone(), moved));
    }
    let rows = sink.carried.len() as u64;
    sink.cursor = cursor;
    Box::new(sink).finish()?;

    out.sync_all()?;
    drop(out);
    drop(src);
    std::fs::rename(&staged, archive)?;
    if let Some(parent) = archive.parent() {
        if let Ok(f) = std::fs::File::open(parent) {
            let _ = f.sync_all();
        }
    }

    Ok(CompactReport {
        bytes_before,
        bytes_after: std::fs::metadata(archive)?.len(),
        rows,
        delta_rows,
    })
}

/// What [`supersede_as_delta`] did, and why. Never silent: a caller that asked
/// for a delta and got a whole entry has to be able to see which.
#[derive(Debug, PartialEq, Eq)]
pub enum SupersedeOutcome {
    /// Stored as a delta. `delta_bytes` is what is on disk now.
    Delta { stored_bytes: u64, delta_bytes: u64, chain_depth: usize },
    /// The delta did not clear [`crate::archive::DELTA_SIZE_ALPHA`]; the entry is
    /// untouched.
    NotSmaller { delta_bytes: u64, stored_bytes: u64 },
    /// The chain would exceed [`MAX_GENERATION_CHAIN`]; keep this one whole. It
    /// becomes the keyframe the next chain is built against.
    ChainTooLong,
}

/// How many generations may chain before one is kept whole as a keyframe.
///
/// **32**, half the reader's [`MAX_RECONSTRUCT_DEPTH`] of 64. The reader's bound
/// is about a hostile index and is a refusal; this is a writer policy about cost,
/// and it is set below the refusal so a legitimate archive never approaches it.
///
/// The cost it bounds is MEASURED on a real eight-generation `nornir` chain
/// (oden, 2026-08-05, ~7.3-8.3 MB packs): reading the live generation is
/// **12.8 ms** and every link behind it adds **18.2 ms** — 6.4 ms of delta
/// application and decompression, 11.8 ms of blake3 over the reconstructed
/// entry. Thirty-two links is therefore ~0.6 s to read the OLDEST generation in
/// a full chain, and the reader's bound of 64 is ~1.2 s.
///
/// That is a cold archival read that nothing on a serve path makes: the live
/// generation holds every reachable object and sits at depth 0, whole, for ever.
/// Against it, each link saves a whole generation — measured **5.26x** over the
/// chain as a whole, 62 350 306 bytes of packs down to 11 860 793.
///
/// **The per-link blake3 is kept, deliberately.** It is 56% of that cold read and
/// it is the only thing that can catch a delta applied to the wrong base — which
/// is exactly the mistake a generation chain invites, because every entry in it
/// is a pack of the same repository at a similar size, and a base swapped for
/// its neighbour would apply cleanly. Trading that for 80 ms on a read nobody
/// makes is the wrong way round.
pub const MAX_GENERATION_CHAIN: usize = 32;

/// One base-schema data batch from `(path, ChunkLoc)` rows, in the order given.
///
/// The single builder for base-schema index rows. Four call sites used to carry
/// a copy of this loop — `emit_carried`, `write_lookup_and_trie`, `ArrowIpcSink`'s
/// two — and a column appended in one order in one of them and another order in
/// the next is a silent index corruption no checksum catches, because every
/// individual chunk still hashes correctly (LAW 5, by construction).
/// `pub` rather than `pub(crate)` since 2026-08-08: `znippy-plugin-git`'s
/// generation-0 seal needs exactly these columns in exactly this order for the
/// verbatim packs it already holds on disk, and a second copy of this loop in
/// that crate is the drift this function was extracted to prevent (LAW 5).
pub fn base_batch_from_rows(paths: &[String], locs: &[ChunkLoc]) -> Result<RecordBatch> {
    let order: Vec<usize> = (0..paths.len()).collect();
    base_batch_permuted(data_subindex_schema(), paths, locs, &order)
}

/// [`base_batch_from_rows`] emitting rows in `order` under an explicit `schema` —
/// the sorted-lookup case, which is the same columns in a different order.
pub(crate) fn base_batch_permuted(
    schema: Arc<Schema>,
    paths: &[String],
    locs: &[ChunkLoc],
    order: &[usize],
) -> Result<RecordBatch> {
    let n = order.len();
    let mut path_b = StringBuilder::with_capacity(n, n * 16);
    let mut seq_b = UInt32Builder::with_capacity(n);
    let mut fdata_b = UInt64Builder::with_capacity(n);
    let mut comp_b = BooleanBuilder::with_capacity(n);
    let mut usz_b = UInt64Builder::with_capacity(n);
    let mut boff_b = UInt64Builder::with_capacity(n);
    let mut bsz_b = UInt64Builder::with_capacity(n);
    let mut ck_b = FixedSizeBinaryBuilder::with_capacity(n, 32);
    for &i in order {
        let loc = &locs[i];
        path_b.append_value(&paths[i]);
        seq_b.append_value(loc.chunk_seq);
        fdata_b.append_value(loc.fdata_offset);
        comp_b.append_value(loc.compressed);
        usz_b.append_value(loc.uncompressed_size);
        boff_b.append_value(loc.blob_offset);
        bsz_b.append_value(loc.blob_size);
        ck_b.append_value(loc.checksum).expect("checksum is 32 bytes");
    }
    Ok(RecordBatch::try_new(
        schema,
        vec![
            Arc::new(path_b.finish()),
            Arc::new(seq_b.finish()),
            Arc::new(fdata_b.finish()),
            Arc::new(comp_b.finish()),
            Arc::new(usz_b.finish()),
            Arc::new(boff_b.finish()),
            Arc::new(bsz_b.finish()),
            Arc::new(ck_b.finish()),
        ],
    )?)
}

/// Compress-or-store each file and write its blob at `cursor`, returning the
/// index rows and the new cursor. **No metadata is touched** — this is the blob
/// half alone, shared by the re-sealing [`append_files`] path and by the hot
/// journal path, so the two cannot drift on the skip decision, the blake3 domain
/// (original bytes) or the store-raw rule.
pub(crate) fn write_blobs(
    file: &File,
    cursor: u64,
    files: &[(String, Vec<u8>)],
    ctx: &mut crate::codec::CompressCtx,
    policy: crate::SkipPolicy,
) -> Result<(Vec<String>, Vec<ChunkLoc>, u64)> {
    let mut paths = Vec::with_capacity(files.len());
    let mut locs = Vec::with_capacity(files.len());
    let mut cursor = cursor;
    for (rel, bytes) in files {
        let checksum = *blake3::hash(bytes).as_bytes();
        // The skip decision, which the append path did not make at all until
        // 2026-08-03: every byte went to the codec and the frame was thrown away
        // whenever it came out no smaller. For already-compressed input — a
        // `.pack`, a `.jar`, a `.crate` — that is the entire codec cost paid to
        // learn what the file's name already said.
        //
        // MEASURED on gunnar's cold tier (oden, 2026-08-03): appending a
        // 34.1 MiB consolidated packfile cost 0.34 s of CPU compressing and
        // 0.03 s skipping — 11.3x — for byte-identical output.
        let skip = policy.skip_by_path(std::path::Path::new(rel.as_str()));
        let frame = if skip { Vec::new() } else { ctx.compress(bytes)? };
        let (on_disk, compressed): (&[u8], bool) = if !skip && frame.len() < bytes.len() {
            (&frame, true)
        } else {
            (bytes, false)
        };
        let blob_offset = cursor;
        file.write_all_at(on_disk, blob_offset)?;
        cursor += on_disk.len() as u64;
        paths.push(rel.clone());
        locs.push(ChunkLoc {
            chunk_seq: 0,
            fdata_offset: 0,
            blob_offset,
            blob_size: on_disk.len() as u64,
            uncompressed_size: bytes.len() as u64,
            compressed,
            checksum,
        });
    }
    Ok((paths, locs, cursor))
}

/// Outcome of a native [`append_files`] call.
#[derive(Debug, Clone)]
pub struct AppendReport {
    /// Data rows that existed in the archive before the append (recovered),
    /// including any that the append then replaced.
    pub rows_before: u64,
    /// Pre-existing rows dropped because the append re-wrote the same
    /// `relative_path` (replace semantics). Always 0 on the fresh-create path.
    pub rows_replaced: u64,
    /// New rows (one per appended file/chunk) written by the append.
    pub rows_added: u64,
    /// Byte offset where the appended blob region started (old blob_end).
    pub blob_append_offset: u64,
    /// Bytes of new blob payload appended (compressed/stored).
    pub blob_bytes_added: u64,
    /// Final size of the re-sealed archive.
    pub sealed_total_bytes: u64,
}

/// First-class **native blob append** — the caller-facing library primitive that
/// the iceberg lifecycle test (#23) previously had to perform by hand. (There is
/// no `compress --append` CLI verb today; this is the in-process entry point.)
///
/// Opens an existing sealed v0.7 `.znippy`, compresses each `(relative_path,
/// bytes)` in `new_files` with the znippy codec, appends the resulting blobs to
/// the same file past the existing blob region, then re-seals (merged old+new
/// lookup + trie + manifest + footer) via [`ArrowIpcSinkAppend`]. The original
/// blob bytes and existing rows are reused verbatim — nothing is recompressed.
///
/// **Replace semantics:** a `relative_path` in `new_files` that the archive
/// already contains REPLACES the existing entry — the pre-existing rows for that
/// path are dropped from the re-sealed index (counted in
/// [`AppendReport::rows_replaced`]) and its old blob bytes become unreferenced
/// dead payload. Appending the same path twice never leaves two live copies.
///
/// Mirrors the real compress path's per-blob accounting: blake3 over the
/// ORIGINAL bytes, store-raw when the codec frame is not smaller, one chunk per
/// file (`chunk_seq = 0`). `compression_level` is the codec level (e.g. 3).
pub fn append_files(
    archive: &Path,
    new_files: &[(String, Vec<u8>)],
    compression_level: i32,
) -> Result<AppendReport> {
    append_files_with_meta(archive, new_files, compression_level, None)
}

/// [`append_files`] with an explicit [`SkipPolicy`](crate::SkipPolicy).
///
/// The batch-level form, for a caller that already knows what it is appending —
/// `SkipPolicy::already_compressed()` stores every entry raw without inspecting
/// a byte, which is exact, free, and better informed than any probe. gunnar's
/// cold tier appends packfiles and uses it.
pub fn append_files_with_policy(
    archive: &Path,
    new_files: &[(String, Vec<u8>)],
    compression_level: i32,
    policy: crate::SkipPolicy,
) -> Result<AppendReport> {
    let sink = ArrowIpcSinkAppend::open_existing(archive)?;
    let rows_before = sink.recovered_rows() as u64;
    let blob_append_offset = sink.blob_end();
    write_files_into_sink(
        sink,
        new_files,
        compression_level,
        rows_before,
        blob_append_offset,
        policy,
    )
}

/// [`append_files`], additionally merging `meta` rows into the archive's
/// searchable metadata sub-index.
///
/// `None` leaves the metadata exactly as the archive had it — including having
/// none. `Some(rows)` merges into whatever was already there (the resume carries
/// the old section forward), creating the section if the archive had none.
pub fn append_files_with_meta(
    archive: &Path,
    new_files: &[(String, Vec<u8>)],
    compression_level: i32,
    meta: Option<MetaTable>,
) -> Result<AppendReport> {
    let mut sink = ArrowIpcSinkAppend::open_existing(archive)?;
    if let Some(table) = meta {
        sink.merge_meta(table.rows().to_vec());
    }
    let rows_before = sink.recovered_rows() as u64;
    let blob_append_offset = sink.blob_end();
    write_files_into_sink(
        sink,
        new_files,
        compression_level,
        rows_before,
        blob_append_offset,
        // Resolve per entry from its name, then from its bytes — the same
        // default `compress_dir` has. Deliberately NOT "always compress", which
        // is what this path did before and which is never the right answer for
        // an entry whose extension already says it is compressed.
        crate::SkipPolicy::resolve(),
    )
}

/// Create a fresh `.znippy` archive from in-memory `files` — the bootstrap inverse
/// of [`append_files`], which requires an already-sealed archive (it rejects an
/// empty manifest). Seed a new writable archive with this, then grow it with
/// [`append_files`]. Overwrites `archive` if it already exists.
pub fn create_archive(
    archive: &Path,
    files: &[(String, Vec<u8>)],
    compression_level: i32,
) -> Result<AppendReport> {
    create_archive_with_meta(archive, files, compression_level, None)
}

/// [`create_archive`], additionally sealing a searchable metadata sub-index.
///
/// `None` seals **no section** — the archive reads back as
/// `ArchiveMeta::NoMetadata` and is byte-identical to one from
/// [`create_archive`]. `Some(table)` seals the section even when the table is
/// empty, which reads back as a present-but-empty index: "searched, records
/// nothing", a different statement from "never had an index".
pub fn create_archive_with_meta(
    archive: &Path,
    files: &[(String, Vec<u8>)],
    compression_level: i32,
    meta: Option<MetaTable>,
) -> Result<AppendReport> {
    let blob_file = Arc::new(
        File::create(archive)
            .map_err(|e| anyhow!("create archive {}: {e}", archive.display()))?,
    );
    let mut sink = ArrowIpcSinkAppend::new(blob_file, 0);
    sink.meta = meta;
    write_files_into_sink(sink, files, compression_level, 0, 0, crate::SkipPolicy::resolve())
}

/// Build a complete `.znippy` archive **entirely in memory** from in-memory
/// `files` and return its bytes — no staging directory, no named output file. The
/// sink needs a positioned-write fd, so this seals into an **anonymous temp file**
/// (`O_TMPFILE` where the OS supports it → never linked into the filesystem
/// namespace), then reads the sealed bytes straight back out. The returned `Vec`
/// is byte-identical to what [`create_archive`] would write to a path.
///
/// Use this for zero-disk pipelines: build a release/airgap archive from product
/// bytes held in RAM and stream it across the gap without ever touching disk on
/// the build side. Pair with [`append_files`] (path) or hold the bytes and re-seal.
pub fn create_archive_to_vec(
    files: &[(String, Vec<u8>)],
    compression_level: i32,
) -> Result<(Vec<u8>, AppendReport)> {
    let anon = Arc::new(
        tempfile::tempfile().map_err(|e| anyhow!("anonymous archive fd: {e}"))?,
    );
    let sink = ArrowIpcSinkAppend::new(anon.clone(), 0);
    let report =
        write_files_into_sink(sink, files, compression_level, 0, 0, crate::SkipPolicy::resolve())?;
    // The Arc keeps the anonymous fd alive past `finish()`; read the sealed bytes.
    let mut bytes = vec![0u8; report.sealed_total_bytes as usize];
    anon.read_exact_at(&mut bytes, 0)
        .map_err(|e| anyhow!("read back anonymous archive: {e}"))?;
    Ok((bytes, report))
}

/// Shared core of [`append_files`] / [`create_archive`]: compress each file's bytes
/// (store-raw if not smaller), write the blobs at the sink's running blob cursor,
/// then push one base-schema data sub-index and seal. `sink` is either a fresh
/// [`ArrowIpcSinkAppend::new`] or an [`ArrowIpcSinkAppend::open_existing`].
fn write_files_into_sink(
    mut sink: ArrowIpcSinkAppend,
    new_files: &[(String, Vec<u8>)],
    compression_level: i32,
    rows_before: u64,
    blob_append_offset: u64,
    policy: crate::SkipPolicy,
) -> Result<AppendReport> {
    use crate::codec::CompressCtx;

    // Replace, don't duplicate: a path being (re-)written now supersedes whatever
    // rows the archive already held for it. The old blob bytes stay in the file as
    // dead payload — they are simply no longer referenced by any index row.
    let incoming: std::collections::HashSet<&str> =
        new_files.iter().map(|(rel, _)| rel.as_str()).collect();
    let rows_replaced = sink.drop_carried_paths(&incoming) as u64;

    // Append the new blob bytes to the file at the running blob cursor, mirroring
    // the compress pipeline (hash original bytes; store-raw if not smaller). One
    // writer, shared with the hot journal path.
    let blob_file = sink.file.clone();
    let mut ctx = CompressCtx::new(compression_level)?;
    let (paths, locs, cursor) =
        write_blobs(&blob_file, blob_append_offset, new_files, &mut ctx, policy)?;
    let blob_bytes_added = cursor - blob_append_offset;
    blob_file.sync_all()?;

    // Advance the sink's cursor past the freshly-written blob region so the new
    // data sub-index lands after the appended blobs (not over them).
    sink.cursor = cursor;

    // A DATA sub-index: seal it with the format-version-stamped schema. With the
    // bare `lookup_schema()` every archive this path writes — i.e. every archive
    // a writable holger repo or `cargo publish` produces — recorded no format
    // version, so the reader-side version pin had nothing to check.
    let batch = base_batch_from_rows(&paths, &locs)?;
    let schema = data_subindex_schema();
    let rows_added = batch.num_rows() as u64;
    sink.push_subindex(
        schema.as_ref(),
        &[batch],
        GroupKey { pkg_type: 0, repo: String::new(), module_name: String::new() },
    )?;

    let sealed_total_bytes = Box::new(sink).finish()?;

    Ok(AppendReport {
        rows_before,
        rows_replaced,
        rows_added,
        blob_append_offset,
        blob_bytes_added,
        sealed_total_bytes,
    })
}

/// Recover existing DATA rows for the resume accumulator. Reads the sorted
/// lookup reserved section (the cheapest source — already the exact per-chunk
/// rows, sorted) when present; otherwise concatenates the data sub-indexes.
pub(crate) fn recover_rows(
    path: &Path,
    entries: &[ManifestEntry],
) -> Result<(Vec<String>, Vec<ChunkLoc>)> {
    use std::io::{Read, Seek, SeekFrom};

    let mut file = File::open(path)?;

    // Prefer the sorted lookup section: it is exactly the base-schema per-chunk
    // rows, one cheap stream decode.
    if let Some(lk) = entries.iter().find(|e| e.module_name == LOOKUP_MODULE) {
        file.seek(SeekFrom::Start(lk.index_offset))?;
        let mut bytes = vec![0u8; lk.index_len as usize];
        file.read_exact(&mut bytes)?;
        return decode_base_rows(&bytes);
    }

    // Fallback: read every NON-reserved data sub-index and concatenate its rows.
    let mut paths = Vec::new();
    let mut locs = Vec::new();
    for e in entries {
        if is_reserved_module(&e.module_name) {
            continue;
        }
        file.seek(SeekFrom::Start(e.index_offset))?;
        let mut bytes = vec![0u8; e.index_len as usize];
        file.read_exact(&mut bytes)?;
        let (mut p, mut l) = decode_base_rows(&bytes)?;
        paths.append(&mut p);
        locs.append(&mut l);
    }
    Ok((paths, locs))
}

// ── inject-assert tests (the "tests inject values, not just no-crash" LAW) ──
// Every test here seals a real archive through a `CompressCtx`, so all of them
// need the codec — see the note on `archive::tests`.
#[cfg(all(test, feature = "openzl"))]
mod tests {
    use super::*;
    use crate::codec::CompressCtx;
    use crate::meta::{BlobMeta, ChunkMeta};
    use crate::{ArrowIpcSink, ZnippyArchive, ZnippyReader};
    use std::time::{SystemTime, UNIX_EPOCH};

    fn unique_dir(tag: &str) -> std::path::PathBuf {
        let ns = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
        let d = std::env::temp_dir().join(format!("znippy_append_{tag}_{ns}_{:?}", std::thread::current().id()));
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    /// **The append path takes the skip decision, and it did not used to.**
    ///
    /// Until 2026-08-03 `write_files_into_sink` ran `CompressCtx::compress` over
    /// every byte of every appended file and kept the frame only when it came
    /// out smaller. For already-compressed input that is the entire codec cost
    /// paid to learn nothing — `compress_dir` has consulted `SkipPolicy` since
    /// it existed, and this path silently did not.
    ///
    /// The observable is chosen so it cannot be faked. A *highly compressible*
    /// payload is appended under a `.pack` name, which the extension table
    /// declares already-compressed. If the policy is honoured the blob is stored
    /// RAW and `blob_bytes_added` equals the input length; if the codec runs, the
    /// frame is far smaller and the number collapses. Asserting on CPU time
    /// would have been the honest measure of the bug but is not a test; this is
    /// the same decision made visible in a byte count.
    ///
    /// Seen RED by restoring the unconditional `ctx.compress(bytes)`:
    /// `blob_bytes_added` drops to a few hundred bytes and the assertion fires.
    #[test]
    fn an_append_honours_the_skip_policy_instead_of_compressing_everything() {
        let dir = unique_dir("skip_policy");
        let archive = dir.join("a.znippy");
        create_archive(&archive, &[("seed.txt".into(), b"seed".to_vec())], 3).unwrap();

        // 256 KiB of one byte: the codec would crush this to almost nothing.
        let squishy = vec![b'A'; 256 * 1024];
        let name = format!("pack-{}.pack", "0f".repeat(20));
        let report =
            append_files(&archive, &[(name.clone(), squishy.clone())], 3).unwrap();

        assert_eq!(
            report.blob_bytes_added,
            squishy.len() as u64,
            "a `.pack` entry must be stored RAW. {} bytes were written for a {}-byte input, so \
             the codec ran over a file the extension table already said was compressed",
            report.blob_bytes_added,
            squishy.len()
        );
        // …and it still reads back byte-exact, which is what stored-raw has to mean.
        assert_eq!(crate::get_file(&archive, &name).unwrap(), squishy);

        // The MIRROR, so the test above is not merely asserting that nothing is
        // ever compressed: the identical bytes under an ordinary name DO go
        // through the codec.
        let report2 =
            append_files(&archive, &[("plain.txt".into(), squishy.clone())], 3).unwrap();
        assert!(
            report2.blob_bytes_added < squishy.len() as u64 / 10,
            "an ordinary name must still be compressed; {} bytes for {}",
            report2.blob_bytes_added,
            squishy.len()
        );
        assert_eq!(crate::get_file(&archive, "plain.txt").unwrap(), squishy);

        // And an explicit batch-level claim overrules the name entirely.
        let report3 = append_files_with_policy(
            &archive,
            &[("also-plain.txt".into(), squishy.clone())],
            3,
            crate::SkipPolicy::already_compressed(),
        )
        .unwrap();
        assert_eq!(
            report3.blob_bytes_added,
            squishy.len() as u64,
            "`already_compressed()` must store raw whatever the name says"
        );

        std::fs::remove_dir_all(&dir).ok();
    }

    /// Deterministic synthetic rows — distinct, lexicographically-spread paths
    /// (same flavour as the bench's `synth_blobs`, so the sort/fst do real work).
    fn synth(n: usize, salt: u64) -> Vec<(String, Vec<u8>)> {
        (0..n)
            .map(|i| {
                let g = (i.wrapping_mul(2_654_435_761) ^ salt as usize) % 1000;
                let p = format!("repo/grp{g:03}/file{:08}_{salt}.bin", i);
                let body = format!("payload {i} salt {salt} {}\n", "z".repeat(8 + (i % 40)));
                (p, body.into_bytes())
            })
            .collect()
    }

    /// Write a fresh sealed archive with `sink` (codec-compressed blobs + one
    /// base-schema data sub-index + the seal). Returns the sealed length. Generic
    /// over a closure so we can drive BOTH ArrowIpcSink (A) and the clone (B)
    /// through the *identical* fresh path and compare bytes.
    fn write_fresh<S: ArchiveMetaSink + 'static>(
        path: &Path,
        files: &[(String, Vec<u8>)],
        make_sink: impl FnOnce(Arc<File>, u64) -> S,
    ) -> u64 {
        let file = Arc::new(File::create(path).unwrap());
        let mut ctx = CompressCtx::new(3).unwrap();
        let mut blobs = Vec::new();
        let mut paths = Vec::new();
        let mut cursor = 0u64;
        for (fi, (rel, bytes)) in files.iter().enumerate() {
            let checksum = *blake3::hash(bytes).as_bytes();
            let frame = ctx.compress(bytes).unwrap();
            let (on_disk, compressed): (&[u8], bool) =
                if frame.len() < bytes.len() { (&frame, true) } else { (bytes, false) };
            file.write_all_at(on_disk, cursor).unwrap();
            let blob_offset = cursor;
            cursor += on_disk.len() as u64;
            paths.push(rel.clone());
            blobs.push(BlobMeta {
                blob_offset,
                blob_size: on_disk.len() as u64,
                chunk_meta: ChunkMeta {
                    fdata_offset: 0,
                    file_index: fi as u64,
                    chunk_seq: 0,
                    checksum,
                    compressed,
                    uncompressed_size: bytes.len() as u64,
                    compressed_size: on_disk.len() as u64,
                },
            });
        }
        let resolver = { let p = paths.clone(); move |fi: u64| p[fi as usize].clone() };
        let batch = crate::build_metadata_batch(&blobs, resolver, &[], &[]).unwrap();
        // Same DATA sub-index schema `write_files_into_sink` seals with, so this
        // helper stays the byte-for-byte reference for `create_archive`.
        let schema = data_subindex_schema();
        let mut sink = make_sink(file.clone(), cursor);
        sink.push_subindex(
            schema.as_ref(),
            &[batch],
            GroupKey { pkg_type: 0, repo: String::new(), module_name: String::new() },
        )
        .unwrap();
        Box::new(sink).finish().unwrap()
    }

    /// PARITY: the clone's fresh write+seal must produce a BYTE-IDENTICAL archive
    /// to the original `ArrowIpcSink` over the same rows. This is the structural
    /// proof that the clone is a zero-cost superset on the normal path (the
    /// throughput parity number lives in the bench; this asserts correctness).
    #[test]
    fn clone_fresh_path_is_byte_identical_to_original() {
        let dir = unique_dir("parity");
        let files = synth(2_000, 1);

        let a = dir.join("a.znippy");
        let b = dir.join("b.znippy");
        let len_a = write_fresh(&a, &files, ArrowIpcSink::new);
        let len_b = write_fresh(&b, &files, ArrowIpcSinkAppend::new);

        assert_eq!(len_a, len_b, "clone seal produced a different total length");
        let bytes_a = std::fs::read(&a).unwrap();
        let bytes_b = std::fs::read(&b).unwrap();
        assert_eq!(
            bytes_a, bytes_b,
            "clone's fresh write path is NOT byte-identical to ArrowIpcSink — parity broken"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// RESUME: native append → reopen with the ORDINARY arrow-ipc reader → both
    /// the original AND the appended files read back byte-exact, and the index
    /// lists exactly old+new. Inject real bytes, assert real bytes out.
    #[test]
    fn native_append_roundtrips_old_and_new_files() {
        let dir = unique_dir("resume");
        let archive = dir.join("store.znippy");
        let orig = synth(1_500, 7);
        write_fresh(&archive, &orig, ArrowIpcSink::new);

        let added = synth(300, 99);
        let report = append_files(&archive, &added, 3).unwrap();
        assert_eq!(report.rows_before, orig.len() as u64, "must recover all original rows");
        assert_eq!(report.rows_added, added.len() as u64);
        assert!(report.blob_bytes_added > 0, "append must write new blob bytes");
        assert!(
            report.sealed_total_bytes > report.blob_append_offset,
            "re-sealed file must be larger than the old blob region"
        );

        // Reopen with the plain reader (no append awareness) and verify EVERY
        // file — original and appended — comes back byte-exact.
        let ar = ZnippyArchive::open(&archive).unwrap();
        let mut listed = ar.list_files().unwrap();
        listed.sort();
        let mut expected: Vec<String> =
            orig.iter().chain(added.iter()).map(|(p, _)| p.clone()).collect();
        expected.sort();
        assert_eq!(listed, expected, "index must list exactly old+new files after append");

        for (p, bytes) in orig.iter().chain(added.iter()) {
            let got = ar.extract_file(p).unwrap();
            assert_eq!(&got, bytes, "byte mismatch after append for {p}");
        }

        // Random-access lookup of an appended file via the rebuilt trie+lookup.
        let probe = &added[123].0;
        let chunks = crate::locate_file(&archive, probe).unwrap();
        assert!(!chunks.is_empty(), "appended file must be locatable via the re-sealed lookup");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// BOOTSTRAP: `create_archive` seeds a fresh archive that (a) is byte-identical
    /// to the proven fresh write path, (b) reads back byte-exact via the ordinary
    /// reader, and (c) can then be grown by `append_files` — the create→append flow
    /// holger's writable `put` relies on. Inject real bytes, assert real bytes out.
    #[test]
    fn create_archive_seeds_then_grows() {
        let dir = unique_dir("create");
        let seed = synth(40, 5);

        // (a) create_archive == write_fresh(_, ArrowIpcSinkAppend::new) byte-for-byte.
        let made = dir.join("made.znippy");
        let report = create_archive(&made, &seed, 3).unwrap();
        assert_eq!(report.rows_before, 0, "fresh archive has no prior rows");
        assert_eq!(report.rows_added, seed.len() as u64);

        let ref_path = dir.join("ref.znippy");
        write_fresh(&ref_path, &seed, ArrowIpcSinkAppend::new);
        assert_eq!(
            std::fs::read(&made).unwrap(),
            std::fs::read(&ref_path).unwrap(),
            "create_archive must be byte-identical to the proven fresh write path"
        );

        // (b) seeded files read back byte-exact via the plain reader.
        let ar = ZnippyArchive::open(&made).unwrap();
        for (p, bytes) in &seed {
            assert_eq!(&ar.extract_file(p).unwrap(), bytes, "seed byte mismatch for {p}");
        }

        // (c) append_files grows the bootstrapped archive; old+new read back exact.
        let added = synth(15, 88);
        let rep2 = append_files(&made, &added, 3).unwrap();
        assert_eq!(rep2.rows_before, seed.len() as u64, "append must recover seeded rows");
        assert_eq!(rep2.rows_added, added.len() as u64);

        let ar2 = ZnippyArchive::open(&made).unwrap();
        for (p, bytes) in seed.iter().chain(added.iter()) {
            assert_eq!(&ar2.extract_file(p).unwrap(), bytes, "byte mismatch after grow for {p}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// ZERO-COPY / no-filesystem: `create_archive_to_vec` builds a valid archive
    /// from in-memory bytes with NO staging tree and NO named output file (it seals
    /// into an anonymous fd). Assert the returned bytes are byte-identical to
    /// `create_archive`'s file output AND read back byte-exact through the reader.
    #[test]
    fn create_archive_to_vec_is_filesystem_free_and_round_trips() {
        let files = synth(24, 7);
        let (bytes, report) = create_archive_to_vec(&files, 3).unwrap();
        assert_eq!(report.rows_before, 0, "fresh in-memory archive has no prior rows");
        assert_eq!(report.rows_added, files.len() as u64);
        assert_eq!(bytes.len() as u64, report.sealed_total_bytes, "vec len == sealed size");

        let dir = unique_dir("tovec");
        // (1) byte-identical to the proven path (`create_archive` → file).
        let ref_path = dir.join("ref.znippy");
        create_archive(&ref_path, &files, 3).unwrap();
        assert_eq!(
            bytes,
            std::fs::read(&ref_path).unwrap(),
            "in-memory archive must be byte-identical to create_archive's file output"
        );
        // (2) the in-memory bytes ARE a real archive: persist + read back byte-exact.
        let p = dir.join("from_mem.znippy");
        std::fs::write(&p, &bytes).unwrap();
        let ar = ZnippyArchive::open(&p).unwrap();
        for (name, content) in &files {
            assert_eq!(&ar.extract_file(name).unwrap(), content, "byte mismatch for {name}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Read an archive's recorded on-disk format version **exactly the way a
    /// reader does** — manifest, then the Arrow schema metadata of the FIRST
    /// non-reserved sub-index. This is byte-for-byte the same walk as
    /// `znippy_common::read_znippy_index`'s `check_format_version` call and as
    /// holger's independent `traits::recorded_format_version` pin, so what this
    /// helper returns is what those two see.
    fn recorded_format_version(path: &Path) -> Option<String> {
        use std::io::{Read, Seek, SeekFrom};

        use arrow::ipc::reader::StreamReader;

        let entries = crate::index::read_znippy_manifest(path).ok()?;
        let mut file = File::open(path).ok()?;
        for e in &entries {
            if is_reserved_module(&e.module_name) {
                continue;
            }
            file.seek(SeekFrom::Start(e.index_offset)).ok()?;
            let mut bytes = vec![0u8; e.index_len as usize];
            file.read_exact(&mut bytes).ok()?;
            let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None).ok()?;
            return reader
                .schema()
                .metadata()
                .get(crate::index::FORMAT_VERSION_KEY)
                .cloned();
        }
        None
    }

    /// FORMAT VERSION: every archive the append/create path writes must RECORD
    /// its on-disk format version — on the fresh create, on the in-memory create,
    /// and still after a re-seal by `append_files` (which rebuilds the whole
    /// metadata tail, so a stamp that only survives the fresh path is no stamp).
    ///
    /// This is what makes a reader-side version pin mean anything for the
    /// archives this path produces — a writable holger repo and `cargo publish`
    /// both write through here. Sealing the data sub-index with the bare
    /// `lookup_schema()` (no schema metadata) recorded NO version at all, and the
    /// pin silently degraded to "undetermined → read as before" for exactly those
    /// archives.
    #[test]
    fn every_appended_archive_records_the_format_version() {
        let dir = unique_dir("fmtver");
        let want = crate::index::ZNIPPY_FORMAT_VERSION.to_string();

        // (a) fresh create.
        let made = dir.join("made.znippy");
        create_archive(&made, &synth(12, 3), 3).unwrap();
        assert_eq!(
            recorded_format_version(&made).as_deref(),
            Some(want.as_str()),
            "create_archive must stamp the on-disk format version"
        );

        // (b) after a re-seal: `append_files` rebuilds the metadata tail from
        // scratch, so the stamp has to be re-emitted, not merely inherited.
        append_files(&made, &synth(7, 91), 3).unwrap();
        assert_eq!(
            recorded_format_version(&made).as_deref(),
            Some(want.as_str()),
            "append_files must re-stamp the format version on the re-sealed archive"
        );

        // (c) the in-memory create path seals through the same code.
        let (bytes, _) = create_archive_to_vec(&synth(9, 4), 3).unwrap();
        let mem = dir.join("mem.znippy");
        std::fs::write(&mem, &bytes).unwrap();
        assert_eq!(
            recorded_format_version(&mem).as_deref(),
            Some(want.as_str()),
            "create_archive_to_vec must stamp the on-disk format version"
        );

        // The stamp did not cost readability: rows still read back byte-exact.
        let ar = ZnippyArchive::open(&mem).unwrap();
        for (p, body) in &synth(9, 4) {
            assert_eq!(&ar.extract_file(p).unwrap(), body, "byte mismatch for {p}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// METADATA SEARCH, end to end through a real archive:
    ///  (a) an archive sealed WITHOUT metadata reads back as `NoMetadata` — the
    ///      backward-compatible case, since that is exactly what every archive
    ///      written before this module looks like;
    ///  (b) sealing metadata makes "which entries carry key X" answerable, and
    ///      the answer resolves to real bytes via the ordinary reader;
    ///  (c) a re-seal by `append_files` CARRIES the metadata forward and merges
    ///      new rows into it — without that, every append would silently erase it;
    ///  (d) the metadata rows never leak into the data index or the lookup.
    #[test]
    fn metadata_is_searchable_survives_a_reseal_and_absence_is_reported_as_absence() {
        use crate::meta_index::{ArchiveMeta, MetaSearch, MetaTable, MetaValue, read_archive_meta};

        let dir = unique_dir("meta");
        let files = synth(30, 2);

        // (a) BACKWARD COMPAT: no metadata sealed → NoMetadata, not "found nothing".
        let plain = dir.join("plain.znippy");
        create_archive(&plain, &files, 3).unwrap();
        let m = read_archive_meta(&plain).unwrap();
        assert_eq!(m, ArchiveMeta::NoMetadata, "an archive with no meta section must say so");
        assert!(!m.is_searchable());
        assert_eq!(m.find_by_key("build-thing"), MetaSearch::NoMetadata);
        assert!(
            m.find_by_key("build-thing").hits().is_none(),
            "absence must NOT present itself as an empty result set"
        );

        // A present-but-EMPTY index is the other state, and it is distinguishable.
        let empty = dir.join("empty.znippy");
        create_archive_with_meta(&empty, &files, 3, Some(MetaTable::new())).unwrap();
        let me = read_archive_meta(&empty).unwrap();
        assert!(me.is_searchable(), "a sealed empty index WAS searched");
        assert!(me.index().is_some_and(|i| i.is_empty()));
        assert_eq!(me.find_by_key("build-thing"), MetaSearch::Hits(&[]));

        // (b) SEARCHABLE: two entries carry a build-thing, one does not.
        let wasm = b"\0asm\x01\0\0\0".to_vec();
        let (p0, p1, p2) = (files[0].0.clone(), files[1].0.clone(), files[2].0.clone());
        let mut t = MetaTable::new();
        t.insert(p0.clone(), "build-thing", MetaValue::Bytes(wasm.clone()))
            .insert(p0.clone(), "build-thing.abi", "wasi-p2")
            .insert(p1.clone(), "build-thing", MetaValue::Bytes(wasm.clone()))
            .insert(p2.clone(), "coverage", 0.5f64)
            .insert_archive("producer", "znippy");

        let ar = dir.join("meta.znippy");
        create_archive_with_meta(&ar, &files, 3, Some(t)).unwrap();

        let m = read_archive_meta(&ar).unwrap();
        let idx = m.index().expect("sealed index is present");
        assert_eq!(idx.len(), 5);
        let hits = m.find_by_key("build-thing").hits().unwrap();
        assert_eq!(hits.len(), 2, "exactly the two entries that carry one");
        let mut got: Vec<&str> = hits.iter().filter_map(|h| h.path()).collect();
        got.sort();
        let mut want = vec![p0.as_str(), p1.as_str()];
        want.sort();
        assert_eq!(got, want, "the search names the right ENTRIES");
        assert_eq!(hits[0].value.as_bytes(), Some(&wasm[..]), "and the right VALUE");
        assert_eq!(idx.archive_value("producer").and_then(MetaValue::as_str), Some("znippy"));
        assert_eq!(
            idx.find_by_prefix("build-thing").len(),
            3,
            "prefix sweeps build-thing + build-thing.abi"
        );

        // The hit resolves to real bytes without extracting anything else.
        let reader = ZnippyArchive::open(&ar).unwrap();
        let want_bytes = &files.iter().find(|(p, _)| *p == p0).unwrap().1;
        assert_eq!(&reader.extract_file(hits[0].path().unwrap()).unwrap(), want_bytes);

        // (c) A RE-SEAL keeps it, and merges.
        let added = synth(6, 77);
        let mut more = MetaTable::new();
        more.insert(added[0].0.clone(), "build-thing", MetaValue::Bytes(wasm.clone()));
        append_files_with_meta(&ar, &added, 3, Some(more)).unwrap();

        let m2 = read_archive_meta(&ar).unwrap();
        let hits2 = m2.find_by_key("build-thing").hits().unwrap();
        assert_eq!(hits2.len(), 3, "the append merged, it did not replace");
        assert_eq!(
            m2.index().unwrap().archive_value("producer").and_then(MetaValue::as_str),
            Some("znippy"),
            "the archive-level row survived the re-seal"
        );

        // A plain `append_files` (no meta argument) must also preserve it.
        append_files(&ar, &synth(3, 91), 3).unwrap();
        assert_eq!(
            read_archive_meta(&ar).unwrap().find_by_key("build-thing").hits().unwrap().len(),
            3,
            "an append that says nothing about metadata must not erase it"
        );

        // (d) The metadata rows are NOT file rows: the data index and the lookup
        //     see only the real entries.
        let ar2 = ZnippyArchive::open(&ar).unwrap();
        let n_files = files.len() + added.len() + 3;
        assert_eq!(
            ar2.file_count(),
            n_files,
            "metadata rows leaked into the data index"
        );
        let lookup_bytes = read_reserved_section_bytes(&ar, LOOKUP_MODULE).unwrap().unwrap();
        assert_eq!(
            decode_base_rows(&lookup_bytes).unwrap().0.len(),
            n_files,
            "metadata rows leaked into the random-access lookup"
        );
        for (p, bytes) in files.iter().chain(added.iter()) {
            assert_eq!(&ar2.extract_file(p).unwrap(), bytes, "byte mismatch for {p}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Cached `ArchiveReader` must return byte-for-byte the SAME chunks and
    /// per-file metadata as the re-reading free functions — for a present file,
    /// an absent file, and a prefix window. Proves idea (B) is a pure read-side
    /// cache with zero behavioural drift from `locate_file` /
    /// `get_files_meta_with_prefix`.

    /// **An append kept the blobs and threw away the ref log.**
    ///
    /// MEASURED 2026-08-04 on gunnar: an object-carrying push took the archive
    /// from 144 928 to 146 396 bytes and `__gunnar_refs__` was **gone**.
    /// `open_existing` truncates the whole metadata tail and re-supplies no
    /// reserved section it was not handed, and nothing hands it one — so every
    /// push after the first erased the ref history, which is what blocks serving
    /// refs from the archive for most of a repository's life.
    ///
    /// The distinction the append path could not express: `__gunnar_graph__` /
    /// `__gunnar_reach__` / `__gunnar_oid__` are DERIVED from the objects and
    /// must go when the objects change; `__gunnar_refs__` / `__gunnar_secrets__`
    /// are INDEPENDENT logs and nothing else can reproduce them. Both halves are
    /// asserted here — carrying everything would be the opposite bug and would
    /// leave a stale commit graph behind.
    ///
    /// Seen RED by dropping the `write_carried_reserved()` call from `finish`:
    /// `__gunnar_refs__` reads back as `None` and the first assertion fires.
    #[test]
    fn an_append_carries_the_ref_log_forward_and_drops_the_derived_sections() {
        use crate::index::{
            GUNNAR_GRAPH_MODULE, GUNNAR_REFS_MODULE, GUNNAR_SECRETS_MODULE,
        };
        use crate::meta_sink::{ReservedSection, ReservedSectionBuilder};

        let dir = unique_dir("carried_reserved");
        let archive = dir.join("a.znippy");

        let refs_bytes = b"refs/heads/main 0123456789abcdef -- push 1".to_vec();
        let secrets_bytes = b"\x00ciphertext-only, never plaintext".to_vec();
        let graph_bytes = b"a commit graph derived from the objects".to_vec();

        // Seal an archive carrying one independent log, one secrets log and one
        // DERIVED section, through the same builder gunnar's cold tier uses.
        {
            let f = Arc::new(File::create(&archive).unwrap());
            let mut cursor = 0u64;
            let mut blobs = Vec::new();
            for (i, (name, bytes)) in [("obj/a.bin", b"first".to_vec())].iter().enumerate() {
                use std::os::unix::fs::FileExt;
                f.write_all_at(bytes, cursor).unwrap();
                blobs.push(BlobMeta {
                    blob_offset: cursor,
                    blob_size: bytes.len() as u64,
                    chunk_meta: ChunkMeta {
                        fdata_offset: 0,
                        file_index: i as u64,
                        chunk_seq: 0,
                        checksum: *blake3::hash(bytes).as_bytes(),
                        compressed: false,
                        uncompressed_size: bytes.len() as u64,
                        compressed_size: bytes.len() as u64,
                    },
                });
                cursor += bytes.len() as u64;
                let _ = name;
            }
            let batch = crate::index::build_metadata_batch(
                &blobs,
                |_fi: u64| "obj/a.bin".to_string(),
                &[],
                &[],
            )
            .unwrap();
            // The sink starts AFTER the blob region, or it writes its index over
            // the bytes it is indexing.
            let mut sink = ArrowIpcSink::new(Arc::clone(&f), cursor);
            let (r, s, g) = (refs_bytes.clone(), secrets_bytes.clone(), graph_bytes.clone());
            let builder: ReservedSectionBuilder = Box::new(move |_lookup| {
                Ok(vec![
                    ReservedSection::raw(GUNNAR_REFS_MODULE, r),
                    ReservedSection::raw(GUNNAR_SECRETS_MODULE, s),
                    ReservedSection::raw(GUNNAR_GRAPH_MODULE, g),
                ])
            });
            sink = sink.with_reserved_builder(builder);
            sink.push_subindex(
                crate::index::lookup_schema().as_ref(),
                &[batch],
                GroupKey { pkg_type: 0, repo: String::new(), module_name: String::new() },
            )
            .unwrap();
            Box::new(sink).finish().unwrap();
        }

        // Sanity: all three are there before the append, or this proves nothing.
        for m in [GUNNAR_REFS_MODULE, GUNNAR_SECRETS_MODULE, GUNNAR_GRAPH_MODULE] {
            assert!(
                read_reserved_section_bytes(&archive, m).unwrap().is_some(),
                "{m} must be present BEFORE the append"
            );
        }

        // An object-carrying push.
        append_files(&archive, &[("obj/b.bin".to_string(), b"second".to_vec())], 3).unwrap();

        // The independent logs survive, byte for byte.
        assert_eq!(
            read_reserved_section_bytes(&archive, GUNNAR_REFS_MODULE).unwrap(),
            Some(refs_bytes),
            "an object-carrying push must not erase the ref log — this is the 2026-08-04 bug"
        );
        assert_eq!(
            read_reserved_section_bytes(&archive, GUNNAR_SECRETS_MODULE).unwrap(),
            Some(secrets_bytes),
        );
        // The DERIVED section is gone, because the objects changed under it.
        assert_eq!(
            read_reserved_section_bytes(&archive, GUNNAR_GRAPH_MODULE).unwrap(),
            None,
            "a commit graph derived from the old object set must NOT be carried forward"
        );

        // And the archive is still an archive: both entries read back.
        let reader = ZnippyArchive::open(&archive).unwrap();
        assert_eq!(reader.extract_file("obj/a.bin").unwrap(), b"first".to_vec());
        assert_eq!(reader.extract_file("obj/b.bin").unwrap(), b"second".to_vec());

        let _ = std::fs::remove_dir_all(&dir);
    }


    /// **The writer, end to end: a superseded entry becomes a delta and still
    /// reads back byte-exact.**
    ///
    /// Two "generations" that share most of their bytes. After
    /// `supersede_as_delta`, generation N is a delta chunk against N+1 and N+1 is
    /// untouched — which is the direction that matters, because N+1 is the one a
    /// clone copies entries out of.
    #[test]
    fn a_superseded_entry_becomes_a_delta_and_reads_back_exactly() {
        let dir = unique_dir("supersede");
        let archive = dir.join("a.znippy");

        // Incompressible base bytes, so the saving measured is the DELTA's and
        // not the codec's — the whole point of the exercise.
        let mut st = 0x243f_6a88_85a3_08d3u64;
        let gen_n: Vec<u8> = (0..400_000u32)
            .map(|_| {
                st = st.wrapping_add(0x9e37_79b9_7f4a_7c15);
                let mut z = st;
                z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
                (z ^ (z >> 27)) as u8
            })
            .collect();
        // N+1 keeps N's bytes and appends a little, as a later pack does.
        let mut gen_n1 = gen_n.clone();
        gen_n1.extend_from_slice(&gen_n[..20_000]);

        create_archive(
            &archive,
            &[
                ("pack-N.pack".to_string(), gen_n.clone()),
                ("pack-N1.pack".to_string(), gen_n1.clone()),
            ],
            3,
        )
        .unwrap();
        let before = std::fs::metadata(&archive).unwrap().len();

        let out = supersede_as_delta(&archive, "pack-N.pack", "pack-N1.pack", 0, 3).unwrap();
        let (stored, delta_bytes) = match out {
            SupersedeOutcome::Delta { stored_bytes, delta_bytes, chain_depth } => {
                assert_eq!(chain_depth, 1);
                (stored_bytes, delta_bytes)
            }
            other => panic!("expected a delta, got {other:?}"),
        };
        assert_eq!(stored, gen_n.len() as u64);
        assert!(
            delta_bytes * 20 < stored,
            "N is a near-prefix of N+1, so the delta must be a small fraction of it; \
             got {delta_bytes} against {stored}"
        );

        // THE ASSERTION THAT MATTERS: both entries still read back byte-exact,
        // and the delta'd one through the delta path.
        let ar = ZnippyArchive::open(&archive).unwrap();
        assert_eq!(ar.extract_file("pack-N.pack").unwrap(), gen_n, "the superseded generation");
        assert_eq!(ar.extract_file("pack-N1.pack").unwrap(), gen_n1, "the live generation");
        assert_eq!(ar.extract_file_verified("pack-N.pack").unwrap(), gen_n);
        assert_eq!(ar.file_size("pack-N.pack"), Some(gen_n.len() as u64));

        // The live generation is still a WHOLE entry. If it were not, `P-001`'s
        // copy path would be reading through a delta chain, which is the one
        // outcome this direction exists to prevent.
        assert!(
            read_delta_map(&archive)
                .unwrap()
                .iter()
                .all(|(p, _, _)| p == "pack-N.pack"),
            "only the superseded generation may be a delta"
        );

        let _ = before;
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// The two refusals, both stated rather than silent.
    #[test]
    fn the_writer_refuses_a_pointless_delta_and_an_over_long_chain() {
        let dir = unique_dir("supersede_refuse");
        let archive = dir.join("a.znippy");
        let mut st = 0x1234_5678_9abc_def0u64;
        let noise = |n: usize, st: &mut u64| -> Vec<u8> {
            (0..n)
                .map(|_| {
                    *st = st.wrapping_add(0x9e37_79b9_7f4a_7c15);
                    let mut z = *st;
                    z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
                    (z ^ (z >> 27)) as u8
                })
                .collect()
        };
        let a = noise(60_000, &mut st);
        let b = noise(60_000, &mut st); // unrelated: no delta can win
        create_archive(
            &archive,
            &[("a.pack".to_string(), a.clone()), ("b.pack".to_string(), b.clone())],
            3,
        )
        .unwrap();

        match supersede_as_delta(&archive, "a.pack", "b.pack", 0, 3).unwrap() {
            SupersedeOutcome::NotSmaller { .. } => {}
            other => panic!("unrelated bytes must not produce a delta, got {other:?}"),
        }
        // Untouched, and still readable.
        let ar = ZnippyArchive::open(&archive).unwrap();
        assert_eq!(ar.extract_file("a.pack").unwrap(), a);
        assert!(read_delta_map(&archive).unwrap().is_empty());

        // And the chain bound, which is a WRITER policy: it declines rather than
        // building a chain the reader would later refuse.
        assert_eq!(
            supersede_as_delta(&archive, "a.pack", "b.pack", MAX_GENERATION_CHAIN, 3).unwrap(),
            SupersedeOutcome::ChainTooLong
        );
        assert!(supersede_as_delta(&archive, "a.pack", "a.pack", 0, 3).is_err());

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// **A chain of generations, and every one of them still exact.**
    ///
    /// Four generations, each superseded against its successor as it arrives, so
    /// the oldest sits at depth 3 and the newest is whole. This is the shape a
    /// repository `gc`'d repeatedly produces, and the property is that reading
    /// generation 0 walks three links and still returns the bytes that were
    /// stored.
    #[test]
    fn a_generation_chain_reads_every_generation_back_exactly() {
        let dir = unique_dir("genchain");
        let archive = dir.join("g.znippy");
        let mut st = 0x0f0f_0f0f_dead_beefu64;
        let mut gens: Vec<Vec<u8>> = Vec::new();
        let mut cur: Vec<u8> = (0..120_000u32)
            .map(|_| {
                st = st.wrapping_add(0x9e37_79b9_7f4a_7c15);
                let mut z = st;
                z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
                (z ^ (z >> 27)) as u8
            })
            .collect();
        gens.push(cur.clone());
        for g in 1..4 {
            cur.extend_from_slice(format!("generation {g} tail ").repeat(200).as_bytes());
            gens.push(cur.clone());
        }
        let files: Vec<(String, Vec<u8>)> = gens
            .iter()
            .enumerate()
            .map(|(i, b)| (format!("pack-{i}.pack"), b.clone()))
            .collect();
        create_archive(&archive, &files, 3).unwrap();

        // Supersede each older generation against the one after it, oldest last,
        // so each base is whole at the moment it is used.
        for i in (0..3).rev() {
            let out = supersede_as_delta(
                &archive,
                &format!("pack-{i}.pack"),
                &format!("pack-{}.pack", i + 1),
                3 - 1 - i,
                3,
            )
            .unwrap();
            assert!(matches!(out, SupersedeOutcome::Delta { .. }), "gen {i}: {out:?}");
        }

        let ar = ZnippyArchive::open(&archive).unwrap();
        for (i, want) in gens.iter().enumerate() {
            assert_eq!(
                &ar.extract_file(&format!("pack-{i}.pack")).unwrap(),
                want,
                "generation {i} at chain depth {}",
                3 - i
            );
        }
        // The newest is whole: nothing in the map names it.
        assert!(
            read_delta_map(&archive)
                .unwrap()
                .iter()
                .all(|(p, _, _)| p != "pack-3.pack"),
            "the live generation must stay whole"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }


    /// **What the writer costs and what it saves, on a REAL generation chain.**
    ///
    /// `#[ignore]`d: it wants real `gc` generations on disk. `ZNIPPY_CHAIN_DIR`
    /// names a directory of `pack-<i>.pack`, `i` ascending, oldest first.
    #[test]
    #[ignore]
    fn perf_real_generation_chain() {
        let Ok(d) = std::env::var("ZNIPPY_CHAIN_DIR") else { return };
        let dir = unique_dir("realchain");
        let archive = dir.join("g.znippy");
        let mut files: Vec<(String, Vec<u8>)> = Vec::new();
        for i in 0.. {
            let p = std::path::Path::new(&d).join(format!("pack-{i}.pack"));
            if !p.is_file() {
                break;
            }
            files.push((format!("pack-{i}.pack"), std::fs::read(&p).unwrap()));
        }
        let n = files.len();
        let raw: u64 = files.iter().map(|(_, b)| b.len() as u64).sum();

        let t0 = std::time::Instant::now();
        create_archive(&archive, &files, 3).unwrap();
        let seal_ms = t0.elapsed().as_secs_f64() * 1e3;
        let sealed = std::fs::metadata(&archive).unwrap().len();

        // Supersede oldest-last, so each base is whole when it is used.
        println!("gen,stored_b,delta_b,ratio_x,supersede_ms");
        let mut encode_ms_total = 0.0;
        let mut delta_total = 0u64;
        for i in (0..n - 1).rev() {
            let t = std::time::Instant::now();
            let out = supersede_as_delta(
                &archive,
                &format!("pack-{i}.pack"),
                &format!("pack-{}.pack", i + 1),
                n - 2 - i,
                3,
            )
            .unwrap();
            let ms = t.elapsed().as_secs_f64() * 1e3;
            encode_ms_total += ms;
            match out {
                SupersedeOutcome::Delta { stored_bytes, delta_bytes, .. } => {
                    delta_total += delta_bytes;
                    println!(
                        "{i},{stored_bytes},{delta_bytes},{:.2},{ms:.0}",
                        stored_bytes as f64 / delta_bytes as f64
                    );
                }
                other => println!("{i},-,-,-,{ms:.0} ({other:?})"),
            }
        }

        // A rewrite reclaims the superseded blobs. Measured by rebuilding the
        // archive from what it now holds, which is what a compaction would do.
        let ar = ZnippyArchive::open(&archive).unwrap();
        let mut live: Vec<(String, Vec<u8>)> = Vec::new();
        let t = std::time::Instant::now();
        for i in 0..n {
            let name = format!("pack-{i}.pack");
            let got = ar.extract_file(&name).unwrap();
            assert_eq!(got, files[i].1, "generation {i} did not read back exactly");
            live.push((name, got));
        }
        let read_all_ms = t.elapsed().as_secs_f64() * 1e3;

        // Per-generation read cost, by chain depth.
        println!("gen,depth,read_ms");
        for i in 0..n {
            let name = format!("pack-{i}.pack");
            let t = std::time::Instant::now();
            for _ in 0..5 {
                let _ = ar.extract_file(&name).unwrap();
            }
            println!("{i},{},{:.2}", n - 1 - i, t.elapsed().as_secs_f64() * 1e3 / 5.0);
        }

        let compacted = dir.join("c.znippy");
        let mut packed: Vec<(String, Vec<u8>)> = Vec::new();
        for i in 0..n {
            packed.push(live[i].clone());
        }
        // The compacted size is what the format would cost steady-state: the live
        // generation whole plus one delta per superseded one, and nothing else.
        let overhead = sealed - raw;
        let steady = files[n - 1].1.len() as u64 + delta_total + overhead;
        println!(
            "SUMMARY generations={n} raw={raw} sealed={sealed} live_whole={} deltas={delta_total} \
             steady_state={steady} saving_x={:.2} seal_ms={seal_ms:.0} \
             supersede_ms_total={encode_ms_total:.0} read_all_ms={read_all_ms:.0}",
            files[n - 1].1.len(),
            raw as f64 / steady as f64
        );
        let _ = (compacted, packed);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn archive_reader_matches_free_functions() {
        let dir = unique_dir("reader");
        let archive = dir.join("store.znippy");
        let files = synth(600, 13);
        write_fresh(&archive, &files, ArrowIpcSink::new);

        let reader = crate::ArchiveReader::open(&archive).unwrap();
        assert_eq!(reader.row_count(), files.len(), "one chunk per synth file");

        // Present files: cached locate == free-function locate.
        for (p, _) in files.iter().step_by(37) {
            let cached = reader.locate(p);
            let free = crate::locate_file(&archive, p).unwrap();
            assert!(!cached.is_empty(), "cached reader failed to locate {p}");
            assert_eq!(cached, free, "cached vs free locate diverged for {p}");
        }

        // Absent file: both return empty.
        let missing = "repo/does/not/exist.bin";
        assert!(reader.locate(missing).is_empty());
        assert!(crate::locate_file(&archive, missing).unwrap().is_empty());

        // Whole-archive metadata parity.
        assert_eq!(
            reader.files_meta(),
            crate::get_all_files_meta(&archive).unwrap(),
            "cached files_meta diverged from get_all_files_meta"
        );

        // Prefix window parity (synth spreads paths across repo/grpNNN/).
        for prefix in ["repo/grp001/", "repo/grp0", "repo/", ""] {
            assert_eq!(
                reader.files_meta_with_prefix(prefix),
                crate::get_files_meta_with_prefix(&archive, prefix).unwrap(),
                "cached prefix meta diverged for {prefix:?}"
            );
        }

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// **A supersede does not shrink the file, and this is what does.**
    ///
    /// The first assertion is the fact the writer's own numbers do not state:
    /// after `supersede_as_delta` the archive is no smaller — the superseded
    /// blob is still in it, unreferenced. The second is that `compact_archive`
    /// removes it and that every entry, delta chunks included, reads back exactly
    /// the same bytes at the same depth afterwards.
    ///
    /// Seen RED three ways, each restored:
    ///
    /// * `sink.delta_map = Vec::new()` instead of carrying it — the compacted
    ///   archive hands back generation 0's delta instruction stream (11 bytes)
    ///   as its content and the byte comparison fires.
    /// * copying `loc` unchanged instead of moving `blob_offset` — every entry
    ///   reads garbage from the wrong offset.
    /// * skipping the rename — the size assertion fires, because the original
    ///   file is still the one on disk.
    #[test]
    fn compaction_reclaims_the_superseded_blob_and_changes_no_entry() {
        let dir = unique_dir("compact");
        let archive = dir.join("c.znippy");

        let mut st = 0x5151_2323_abcd_ef01u64;
        let base: Vec<u8> = (0..600_000u32)
            .map(|_| {
                st = st.wrapping_add(0x9e37_79b9_7f4a_7c15);
                let mut z = st;
                z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
                (z ^ (z >> 27)) as u8
            })
            .collect();
        let mut gens: Vec<Vec<u8>> = vec![base.clone()];
        for g in 1..4 {
            let mut next = gens[g - 1].clone();
            next.extend_from_slice(format!("generation {g} tail ").repeat(300).as_bytes());
            gens.push(next);
        }
        let files: Vec<(String, Vec<u8>)> = gens
            .iter()
            .enumerate()
            .map(|(i, b)| (format!("pack-{i}.pack"), b.clone()))
            .collect();
        create_archive(&archive, &files, 3).unwrap();
        let raw: u64 = gens.iter().map(|g| g.len() as u64).sum();
        let whole = std::fs::metadata(&archive).unwrap().len();

        for i in (0..3).rev() {
            let out = supersede_as_delta(
                &archive,
                &format!("pack-{i}.pack"),
                &format!("pack-{}.pack", i + 1),
                3 - 1 - i,
                3,
            )
            .unwrap();
            assert!(matches!(out, SupersedeOutcome::Delta { .. }), "gen {i}: {out:?}");
        }

        // The finding, as an assertion. Three of four generations are now
        // deltas of a few hundred bytes each, and the file is not smaller.
        let after_supersede = std::fs::metadata(&archive).unwrap().len();
        assert!(
            after_supersede >= whole,
            "supersede shrank the file ({whole} -> {after_supersede}); if that is now true, this \
             test and everything built on the dead-payload finding wants revisiting"
        );

        let report = compact_archive(&archive).unwrap();
        let compacted = std::fs::metadata(&archive).unwrap().len();
        assert_eq!(report.bytes_after, compacted);
        assert_eq!(report.rows, 4, "a compaction must not change the row count");
        assert_eq!(report.delta_rows, 3, "the delta map must travel across it");
        assert!(
            compacted * 2 < raw,
            "the compacted archive is {compacted} bytes for {raw} bytes of generations — the dead \
             payload was not reclaimed"
        );

        // Every generation, at every depth, byte for byte.
        let ar = ZnippyArchive::open(&archive).unwrap();
        for (i, want) in gens.iter().enumerate() {
            assert_eq!(
                &ar.extract_file(&format!("pack-{i}.pack")).unwrap(),
                want,
                "generation {i} did not survive the compaction"
            );
        }
        // The live generation is still whole, and the map still says so.
        let map = read_delta_map(&archive).unwrap();
        assert!(
            map.iter().all(|(p, _, _)| p != "pack-3.pack"),
            "compaction must not put the live generation behind a link: {map:?}"
        );

        // Idempotent: a second compaction of a compact archive changes nothing
        // it should not, which is what a `gc` on a timer depends on.
        let again = compact_archive(&archive).unwrap();
        assert_eq!(again.rows, 4);
        assert_eq!(again.delta_rows, 3);
        let ar = ZnippyArchive::open(&archive).unwrap();
        for (i, want) in gens.iter().enumerate() {
            assert_eq!(&ar.extract_file(&format!("pack-{i}.pack")).unwrap(), want);
        }

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A compaction must carry the independent reserved logs too. gunnar's
    /// `__gunnar_refs__` is the section this archive format has already lost once
    /// (`25850cd`), and a rewrite is the same opportunity to lose it.
    #[test]
    fn compaction_carries_the_reserved_logs() {
        use crate::index::GUNNAR_REFS_MODULE;
        use crate::meta_sink::{ReservedSection, ReservedSectionBuilder};

        let dir = unique_dir("compact_reserved");
        let archive = dir.join("r.znippy");
        let refs_bytes = b"refs/heads/main 0123456789abcdef".to_vec();

        {
            use std::os::unix::fs::FileExt;
            let f = Arc::new(File::create(&archive).unwrap());
            let payload = b"an entry".to_vec();
            f.write_all_at(&payload, 0).unwrap();
            let blobs = vec![BlobMeta {
                blob_offset: 0,
                blob_size: payload.len() as u64,
                chunk_meta: ChunkMeta {
                    fdata_offset: 0,
                    file_index: 0,
                    chunk_seq: 0,
                    checksum: *blake3::hash(&payload).as_bytes(),
                    compressed: false,
                    uncompressed_size: payload.len() as u64,
                    compressed_size: payload.len() as u64,
                },
            }];
            let batch =
                crate::index::build_metadata_batch(&blobs, |_| "obj/a.bin".to_string(), &[], &[])
                    .unwrap();
            let mut sink = ArrowIpcSink::new(Arc::clone(&f), payload.len() as u64);
            let carried = refs_bytes.clone();
            let builder: ReservedSectionBuilder =
                Box::new(move |_lookup| Ok(vec![ReservedSection::raw(GUNNAR_REFS_MODULE, carried)]));
            sink = sink.with_reserved_builder(builder);
            sink.push_subindex(
                crate::index::lookup_schema().as_ref(),
                &[batch],
                GroupKey {
                    pkg_type: 0,
                    repo: String::new(),
                    module_name: String::new(),
                },
            )
            .unwrap();
            Box::new(sink).finish().unwrap();
        }
        assert_eq!(
            read_reserved_section_bytes(&archive, GUNNAR_REFS_MODULE).unwrap(),
            Some(refs_bytes.clone())
        );

        compact_archive(&archive).unwrap();
        assert_eq!(
            read_reserved_section_bytes(&archive, GUNNAR_REFS_MODULE).unwrap(),
            Some(refs_bytes),
            "the compaction dropped __gunnar_refs__"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }
}

/// Decode an Arrow-IPC sub-index stream of the base schema into parallel
/// `(paths, locs)` vectors (mirrors `index::decode_lookup`, kept here so the
/// original stays untouched).
pub(crate) fn decode_base_rows(bytes: &[u8]) -> Result<(Vec<String>, Vec<ChunkLoc>)> {
    use arrow::ipc::reader::StreamReader;

    let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)
        .map_err(|e| anyhow!("append: lookup ipc reader: {e}"))?;
    let mut paths = Vec::new();
    let mut locs = Vec::new();
    for batch in reader {
        let batch = batch.map_err(|e| anyhow!("append: lookup batch decode: {e}"))?;
        let get = |n: &str| batch.column_by_name(n)
            .ok_or_else(|| anyhow!("append: lookup missing column {n}"));
        let p = get("relative_path")?.as_any().downcast_ref::<StringArray>()
            .ok_or_else(|| anyhow!("relative_path type"))?;
        let seq = get("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()
            .ok_or_else(|| anyhow!("chunk_seq type"))?;
        let fdata = get("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow!("fdata_offset type"))?;
        let comp = get("compressed")?.as_any().downcast_ref::<BooleanArray>()
            .ok_or_else(|| anyhow!("compressed type"))?;
        let usz = get("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow!("uncompressed_size type"))?;
        let boff = get("blob_offset")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow!("blob_offset type"))?;
        let bsz = get("blob_size")?.as_any().downcast_ref::<UInt64Array>()
            .ok_or_else(|| anyhow!("blob_size type"))?;
        let ck = get("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()
            .ok_or_else(|| anyhow!("checksum type"))?;
        for i in 0..batch.num_rows() {
            let mut c = [0u8; 32];
            c.copy_from_slice(ck.value(i));
            paths.push(p.value(i).to_string());
            locs.push(ChunkLoc {
                chunk_seq: seq.value(i),
                fdata_offset: fdata.value(i),
                blob_offset: boff.value(i),
                blob_size: bsz.value(i),
                uncompressed_size: usz.value(i),
                compressed: comp.value(i),
                checksum: c,
            });
        }
    }
    Ok((paths, locs))
}