znippy_common/

hot.rs

//! `HotArchive` — the **dynamic half** of a znippy archive: append blobs and
//! index rows at O(1) per append, and pay the static seal exactly once.
//!
//! # The defect this exists to remove, measured
//!
//! [`append_files`](crate::append_files) re-seals the *whole* metadata tail on
//! every call: [`ArrowIpcSinkAppend::open_existing`](crate::ArrowIpcSinkAppend::open_existing)
//! decodes every pre-existing row out of the lookup section, `finish()` re-emits
//! them as a data sub-index, re-sorts all of them, rebuilds the fst trie over all
//! of them, and rewrites manifest and footer. So one append costs O(rows already
//! in the archive) and N appends cost **O(N²)**.
//!
//! That is not a theory. **MEASURED on oden 2026-08-04**, loadavg 14–17, 32
//! cores / 512 GB, release build, archive on **md1** (`/dev/md1`, not tmpfs — so
//! the two fsyncs per append are real), one archive, two entries appended per
//! call (a `pack-*.pack` and its `.idx`, ~1.5 KiB together),
//! `SkipPolicy::already_compressed()`. Same fixture, same rungs, A/B in one
//! binary:
//!
//! | appends | CPU/append, re-seal | CPU/append, hot | wall total, re-seal | wall total, hot |
//! |---|---|---|---|---|
//! | 800 | 1.88 ms | — | | |
//! | 2 000 | 2.66 ms | 0.075 ms | | |
//! | 4 000 | 5.38 ms | 0.075 ms | | |
//! | 8 000 | **10.18 ms** | **0.060 ms** | **61.16 s** | **2.29 s** |
//! | 32 000 | (not run) | 0.061 ms | | 5.62 s |
//!
//! Re-seal's per-append cost doubles every time N doubles — a straight line
//! through the origin, the signature of a quadratic total. The hot path's is
//! **flat from 2 000 to 32 000 appends**, which is the claim. At 8 000 appends
//! that is **26.7× on wall clock and 170× on CPU per append**, and both ratios
//! grow with N because only one side of them grows.
//!
//! INFERRED, not measured: extrapolating re-seal's line, the 100 000th append
//! costs ~127 ms and getting there costs ~1.8 h of CPU. The hot path's 100 000th
//! append costs what its 2 000th did.
//!
//! The seal is the once-only price: **0.08 s of CPU for 64 000 rows**. At 8 000
//! appends the two paths land on the same archive — 15 929 871 B sealed from hot
//! against 15 932 909 B re-sealed, the 3 038 B being the seed entry
//! `append_files` needs and the hot path does not (it can start from nothing).
//!
//! # The two-phase shape
//!
//! ```text
//!   HOT                                        COLD
//!   foo.znippy        blobs only, pure append   foo.znippy   [blobs][index][lookup][trie][manifest][footer]
//!   foo.znippy.hot.NNNNN   Arrow IPC journal    (journal removed)
//! ```
//!
//! * [`HotArchive::append`] writes the blob bytes past the blob cursor and
//!   *one* Arrow IPC RecordBatch of the new index rows to the journal. It never
//!   reads, sorts or rewrites a row it wrote earlier, so its cost depends only on
//!   what is being appended. **O(1) in the archive's size.**
//! * [`HotArchive::seal`] pays the static cost once: one sorted lookup, one fst
//!   trie, one manifest, one footer.
//!
//! This is znippy's own `znippy-iceberg` architecture — dynamic metadata store,
//! then `seal()` freezing it into the static `.znippy` with the blobs copied
//! verbatim — with the metadata store made local and cheap instead of an Iceberg
//! warehouse. Iceberg's `fast_append` rewrites its manifest list on every commit,
//! which is the same O(N²) in a different format.
//!
//! # Why the sealed output is byte-identical to `create_archive`
//!
//! The blobs are written in the same order at the same offsets, and the seal
//! pushes exactly one data sub-index carrying the rows in insertion order —
//! literally the same call [`create_archive`](crate::create_archive) makes. So
//! *appending N files one at a time and sealing* produces the **same bytes** as
//! *creating the archive from all N at once*. `seal_matches_create_archive_byte_for_byte`
//! asserts that on real archives rather than asserting a row count, because a
//! seal that silently dropped the trie would still be able to report the right
//! number of rows.
//!
//! # Journal segments, and why there is more than one file
//!
//! An Arrow IPC *stream* begins with a schema message, so appending batches to
//! one across process restarts is not possible without re-emitting the schema —
//! 424 B of flatbuffer metadata on a four-row batch, paid per push. Instead each
//! **process** that opens the archive for append starts a new segment
//! `foo.znippy.hot.NNNNN`; the schema is written once per segment, and a push
//! costs one RecordBatch message. Segment count grows with restarts, not with
//! pushes.
//!
//! A torn segment (killed mid-write) is read up to its last complete message and
//! the rest is discarded, which is what a log reader must do; the blob bytes
//! whose rows were lost stay in the file as dead payload the seal drops, exactly
//! like [`AppendReport::rows_replaced`](crate::AppendReport::rows_replaced).
//!
//! # Ordering, so a row never points at a blob that is not there
//!
//! `append` fsyncs the blob bytes **before** the journal batch that references
//! them. A crash between the two loses the append and nothing else; a crash the
//! other way round would leave an index row pointing into a hole.

use std::collections::HashMap;
use std::fs::File;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{Result, anyhow, bail};
use arrow::ipc::reader::StreamReader;
use arrow::ipc::writer::StreamWriter;

use crate::archive::ZnippyReader;
use crate::codec::CompressCtx;
use crate::index::{ChunkLoc, data_subindex_schema};
use crate::meta_sink::{ArchiveMetaSink, ArrowIpcSink, GroupKey, ReservedSectionBuilder};
use crate::meta_sink_append::{base_batch_from_rows, decode_base_rows, write_blobs};

/// Suffix of a journal segment: `<archive>.hot.<NNNNN>`.
const JOURNAL_INFIX: &str = ".hot.";

/// What one [`HotArchive::append`] did. Deliberately the same three counts
/// [`AppendReport`](crate::AppendReport) reports, so a caller can swap paths
/// without re-reading its own accounting.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct HotAppendReport {
    /// Index rows written by this append (one per chunk).
    pub rows_added: u64,
    /// Rows already in the journal that this append superseded, by
    /// `relative_path`. Their blob bytes stay in the file as dead payload.
    pub rows_replaced: u64,
    /// Byte offset the appended blob region started at.
    pub blob_append_offset: u64,
    /// Blob payload bytes written (compressed or stored raw).
    pub blob_bytes_added: u64,
    /// Bytes the journal batch cost — the index-side price of this append.
    pub journal_bytes_added: u64,
}

/// What one [`HotArchive::seal`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HotSealReport {
    pub archive: PathBuf,
    /// Rows carried out of the journal into the static index.
    pub rows_sealed: u64,
    /// Journal segments removed.
    pub segments_removed: usize,
    /// Journal bytes reclaimed.
    pub journal_bytes_reclaimed: u64,
    /// Final size of the sealed archive.
    pub sealed_total_bytes: u64,
}

/// A znippy archive in its **hot** phase: blobs appended in place, index rows in
/// a journal beside it.
///
/// [`open`](Self::open) reads the journal once (O(rows)); every [`append`](Self::append)
/// after that is O(what is appended). Holding the rows in memory is what buys
/// that, and it is the same set of rows the old path decoded *per append*.
pub struct HotArchive {
    archive: PathBuf,
    blobs: Arc<File>,
    blob_end: u64,
    /// Every live row, in insertion order. The seal emits exactly this.
    paths: Vec<String>,
    locs: Vec<ChunkLoc>,
    /// `relative_path` → index into `paths`/`locs`, for replace semantics.
    by_path: HashMap<String, usize>,
    /// The segment this process appends to, schema already written.
    journal: StreamWriter<File>,
    journal_path: PathBuf,
    journal_len: u64,
    /// Long-lived, per Design Law 2 — a codec context built per push would put
    /// the setup cost back into the hot path this module exists to empty. Behind
    /// a `Mutex` only because `ZnippyReader` is `Sync` and the raw OpenZL context
    /// is not; `append` takes `&mut self`, so the lock is never contended.
    ctx: std::sync::Mutex<CompressCtx>,
    policy: crate::SkipPolicy,
}

impl HotArchive {
    /// Open `archive` for hot appends, creating it if absent.
    ///
    /// Three entry states, all handled here so no caller has to know which it is:
    ///
    /// * **already hot** — journal segments exist; their rows are loaded and the
    ///   blob cursor is taken from the last one.
    /// * **sealed** — a static `.znippy`; its rows are recovered into a fresh
    ///   segment and the (rebuildable) metadata tail is dropped. O(rows), once.
    /// * **absent** — a new archive at blob offset 0.
    pub fn open(archive: &Path, level: i32, policy: crate::SkipPolicy) -> Result<Self> {
        let archive = archive.to_path_buf();
        let segments = journal_segments(&archive)?;

        let (mut paths, mut locs) = (Vec::new(), Vec::new());
        let mut reheated = false;

        if !segments.is_empty() {
            for (_, seg) in &segments {
                let (p, l) = read_segment(seg)?;
                paths.extend(p);
                locs.extend(l);
            }
        } else if archive.is_file() && crate::index::read_znippy_full_manifest(&archive).is_ok() {
            // Sealed → hot. Recover the rows out of the static index, then drop
            // the tail. The journal is made durable BEFORE the truncation, so a
            // crash in between leaves the rows recoverable and the stale tail is
            // simply overwritten by the next append (and `set_len` below).
            let (entries, _) = crate::index::read_znippy_full_manifest(&archive)?;
            let (p, l) = crate::meta_sink_append::recover_rows(&archive, &entries)?;
            paths = p;
            locs = l;
            reheated = true;
        }

        // Dedup to live rows: last writer for a path wins, insertion order kept.
        let (paths, locs) = live_rows(paths, locs);

        // The blob cursor is the end of the last live blob — NOT the file length,
        // which on the sealed path still includes the metadata tail and after a
        // torn append may include payload no row references.
        let blob_end = locs
            .iter()
            .map(|l| l.blob_offset + l.blob_size)
            .max()
            .unwrap_or(0);

        let next = segments.last().map(|(n, _)| n + 1).unwrap_or(0);
        let journal_path = segment_path(&archive, next);
        let mut journal = new_segment(&journal_path)?;
        if reheated {
            // Segment 0 of a re-heated archive carries the recovered rows, so the
            // journal alone is a complete statement of the archive's index.
            let batch = base_batch_from_rows(&paths, &locs)?;
            journal
                .write(&batch)
                .map_err(|e| anyhow!("hot: seeding journal from sealed index: {e}"))?;
            journal.flush().map_err(|e| anyhow!("hot: journal flush: {e}"))?;
            journal
                .get_ref()
                .sync_all()
                .map_err(|e| anyhow!("hot: journal fsync: {e}"))?;
            sync_parent_dir(&journal_path);
        }
        let journal_len = std::fs::metadata(&journal_path).map(|m| m.len()).unwrap_or(0);

        let blobs = Arc::new(
            std::fs::OpenOptions::new()
                .read(true)
                .write(true)
                .create(true)
                .truncate(false)
                .open(&archive)
                .map_err(|e| anyhow!("hot: open {} : {e}", archive.display()))?,
        );
        // Drop anything past the live blob region — a stale metadata tail from
        // the sealed path, or a torn append's orphan payload. Idempotent.
        if blobs.metadata()?.len() > blob_end {
            blobs.set_len(blob_end)?;
            blobs.sync_all()?;
        }

        let mut by_path = HashMap::with_capacity(paths.len());
        for (i, p) in paths.iter().enumerate() {
            by_path.insert(p.clone(), i);
        }

        Ok(Self {
            archive,
            blobs,
            blob_end,
            paths,
            locs,
            by_path,
            journal,
            journal_path,
            journal_len,
            ctx: std::sync::Mutex::new(CompressCtx::new(level)?),
            policy,
        })
    }

    /// Whether `archive` currently has a journal — i.e. is in its hot phase.
    pub fn is_hot(archive: &Path) -> bool {
        journal_segments(archive).map(|s| !s.is_empty()).unwrap_or(false)
    }

    /// The archive path.
    pub fn archive(&self) -> &Path {
        &self.archive
    }

    /// Live index rows.
    pub fn rows(&self) -> u64 {
        self.paths.len() as u64
    }

    /// Byte offset one past the last blob.
    pub fn blob_end(&self) -> u64 {
        self.blob_end
    }

    /// Append `files`, paying only for `files`.
    ///
    /// Replace semantics match [`append_files`](crate::append_files): a
    /// `relative_path` already present is superseded and its old blob bytes
    /// become dead payload.
    pub fn append(&mut self, files: &[(String, Vec<u8>)]) -> Result<HotAppendReport> {
        if files.is_empty() {
            return Ok(HotAppendReport {
                blob_append_offset: self.blob_end,
                ..Default::default()
            });
        }
        let blob_append_offset = self.blob_end;

        // Blobs first, and durable before any row points at them.
        let (new_paths, new_locs, cursor) =
            write_blobs(&self.blobs, self.blob_end, files, &mut self.ctx.lock().unwrap(), self.policy)?;
        let blob_bytes_added = cursor - blob_append_offset;
        self.blobs.sync_all()?;
        self.blob_end = cursor;

        // One RecordBatch, one message on the journal. No re-sort, no trie, no
        // manifest — that is the entire difference from `append_files`.
        let batch = base_batch_from_rows(&new_paths, &new_locs)?;
        self.journal
            .write(&batch)
            .map_err(|e| anyhow!("hot: journal write: {e}"))?;
        self.journal.flush().map_err(|e| anyhow!("hot: journal flush: {e}"))?;
        self.journal
            .get_ref()
            .sync_all()
            .map_err(|e| anyhow!("hot: journal fsync: {e}"))?;
        let journal_len = std::fs::metadata(&self.journal_path)?.len();
        let journal_bytes_added = journal_len - self.journal_len;
        self.journal_len = journal_len;

        let mut rows_replaced = 0u64;
        for (p, l) in new_paths.into_iter().zip(new_locs) {
            match self.by_path.get(&p).copied() {
                Some(i) => {
                    rows_replaced += 1;
                    self.locs[i] = l;
                }
                None => {
                    self.by_path.insert(p.clone(), self.paths.len());
                    self.paths.push(p);
                    self.locs.push(l);
                }
            }
        }

        Ok(HotAppendReport {
            rows_added: batch.num_rows() as u64,
            rows_replaced,
            blob_append_offset,
            blob_bytes_added,
            journal_bytes_added,
        })
    }

    /// Freeze the hot archive into a static `.znippy` and drop the journal.
    ///
    /// The blobs are already in place and are not touched: the seal writes the
    /// metadata tail only. `reserved` is the same hook
    /// [`ArrowIpcSink::with_reserved_builder`](crate::ArrowIpcSink::with_reserved_builder)
    /// offers, so a caller with its own sections (gunnar's `__gunnar_refs__`)
    /// seals them here rather than needing a second seal.
    pub fn seal(self, reserved: Option<ReservedSectionBuilder>) -> Result<HotSealReport> {
        let Self {
            archive,
            blobs,
            blob_end,
            paths,
            locs,
            journal,
            journal_path,
            ..
        } = self;
        drop(journal);

        // The ORIGINAL sink, not the append clone: the seal has every row in
        // hand, so it needs none of the resume machinery — and this is the sink
        // that already carries the reserved-section hook gunnar's
        // `__gunnar_refs__` needs. The two are proven byte-identical on the fresh
        // path by `clone_fresh_path_is_byte_identical_to_original`.
        let rows_sealed = paths.len() as u64;
        let mut sink = ArrowIpcSink::new(blobs, blob_end);
        if let Some(b) = reserved {
            sink = sink.with_reserved_builder(b);
        }
        if !paths.is_empty() {
            let batch = base_batch_from_rows(&paths, &locs)?;
            let schema = data_subindex_schema();
            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()?;

        // The static archive is durable; only now is the journal redundant.
        let segments = journal_segments(&archive)?;
        let mut journal_bytes_reclaimed = 0u64;
        for (_, seg) in &segments {
            journal_bytes_reclaimed += std::fs::metadata(seg).map(|m| m.len()).unwrap_or(0);
            std::fs::remove_file(seg)
                .map_err(|e| anyhow!("hot: removing journal {}: {e}", seg.display()))?;
        }
        sync_parent_dir(&journal_path);

        Ok(HotSealReport {
            archive,
            rows_sealed,
            segments_removed: segments.len(),
            journal_bytes_reclaimed,
            sealed_total_bytes,
        })
    }
}

/// The hot archive reads without being sealed — that is the other half of "worth
/// having". Same trait the static [`ZnippyArchive`](crate::ZnippyArchive)
/// implements, so a caller does not branch on phase.
impl ZnippyReader for HotArchive {
    fn list_files(&self) -> Result<Vec<String>> {
        Ok(self.paths.clone())
    }

    fn contains(&self, relative_path: &str) -> bool {
        self.by_path.contains_key(relative_path)
    }

    fn file_size(&self, relative_path: &str) -> Option<u64> {
        self.by_path
            .get(relative_path)
            .map(|&i| self.locs[i].uncompressed_size)
    }

    fn extract_file(&self, relative_path: &str) -> Result<Vec<u8>> {
        let &i = self
            .by_path
            .get(relative_path)
            .ok_or_else(|| anyhow!("file not found in archive: {relative_path}"))?;
        let loc = &self.locs[i];
        let mut blob = vec![0u8; loc.blob_size as usize];
        self.blobs.read_exact_at(&mut blob, loc.blob_offset)?;
        if loc.compressed {
            let mut out = Vec::new();
            crate::codec::decompress_into(&blob, &mut out)?;
            Ok(out)
        } else {
            Ok(blob)
        }
    }
}

/// Keep the last row written for each `relative_path`, in first-insertion order —
/// the same last-writer-wins the re-sealing path gets from `drop_carried_paths`.
fn live_rows(paths: Vec<String>, locs: Vec<ChunkLoc>) -> (Vec<String>, Vec<ChunkLoc>) {
    let mut at: HashMap<&str, usize> = HashMap::with_capacity(paths.len());
    let mut order: Vec<usize> = Vec::with_capacity(paths.len());
    for (i, p) in paths.iter().enumerate() {
        match at.get(p.as_str()).copied() {
            Some(slot) => order[slot] = i,
            None => {
                at.insert(p.as_str(), order.len());
                order.push(i);
            }
        }
    }
    let order: Vec<usize> = order;
    let mut out_p = Vec::with_capacity(order.len());
    let mut out_l = Vec::with_capacity(order.len());
    for &i in &order {
        out_p.push(paths[i].clone());
        out_l.push(locs[i].clone());
    }
    (out_p, out_l)
}

fn segment_path(archive: &Path, n: u32) -> PathBuf {
    let mut s = archive.as_os_str().to_os_string();
    s.push(format!("{JOURNAL_INFIX}{n:05}"));
    PathBuf::from(s)
}

/// Every journal segment for `archive`, sorted by sequence number.
fn journal_segments(archive: &Path) -> Result<Vec<(u32, PathBuf)>> {
    let dir = archive.parent().filter(|p| !p.as_os_str().is_empty());
    let dir = dir.map(|d| d.to_path_buf()).unwrap_or_else(|| PathBuf::from("."));
    let stem = archive
        .file_name()
        .and_then(|n| n.to_str())
        .ok_or_else(|| anyhow!("hot: {} has no usable file name", archive.display()))?;
    let prefix = format!("{stem}{JOURNAL_INFIX}");
    let mut out = Vec::new();
    let rd = match std::fs::read_dir(&dir) {
        Ok(rd) => rd,
        Err(_) => return Ok(out),
    };
    for e in rd.flatten() {
        let name = e.file_name();
        let Some(name) = name.to_str() else { continue };
        let Some(tail) = name.strip_prefix(&prefix) else { continue };
        let Ok(n) = tail.parse::<u32>() else { continue };
        out.push((n, e.path()));
    }
    out.sort_by_key(|(n, _)| *n);
    Ok(out)
}

/// Buffer alignment for the journal, in bytes.
///
/// **8, not arrow-rs's default 64.** The 64 is arrow-rs's choice, not Arrow's —
/// arrow-cpp and pyarrow have always written 8 — so this is what the C++
/// reference implementation does and not a bent format. It matters here and
/// almost nowhere else, because a journal batch is *small*: the base schema has
/// eight columns and a two-row batch therefore carries a dozen-plus buffers that
/// are each rounded up. MEASURED on oden, 2026-08-04, two rows per append
/// (a `pack-*.pack` and its `.idx`, 40-hex names): **1 664 B/append at 64-byte
/// alignment, 824 B at 8** — 2.02×, for byte-identical rows. Flat at 824 B all
/// the way to 32 000 appends, so it is a per-append constant and not a curve.
///
/// The **sealed** archive is untouched by this: it is written by `ArrowIpcSink`
/// with arrow-rs's default, so every `.znippy` on disk keeps the bytes its
/// readers already expect. This is the transient half only.
const JOURNAL_ALIGNMENT: usize = 8;

fn new_segment(path: &Path) -> Result<StreamWriter<File>> {
    let f = File::create(path)
        .map_err(|e| anyhow!("hot: create journal segment {}: {e}", path.display()))?;
    let schema = data_subindex_schema();
    let opts = arrow::ipc::writer::IpcWriteOptions::try_new(
        JOURNAL_ALIGNMENT,
        false,
        arrow::ipc::MetadataVersion::V5,
    )
    .map_err(|e| anyhow!("hot: journal write options: {e}"))?;
    let w = StreamWriter::try_new_with_options(f, schema.as_ref(), opts)
        .map_err(|e| anyhow!("hot: journal schema: {e}"))?;
    sync_parent_dir(path);
    Ok(w)
}

/// Read one segment, tolerating a torn tail.
///
/// A killed process leaves a partial final message. Every complete message
/// before it is real and is kept; the partial one is dropped, which is what makes
/// the journal a log rather than a document. `decode_base_rows` cannot be reused
/// here for exactly that reason — it fails the whole stream on the first bad
/// batch, which is right for a sealed section and wrong for a log.
fn read_segment(path: &Path) -> Result<(Vec<String>, Vec<ChunkLoc>)> {
    let bytes = std::fs::read(path)
        .map_err(|e| anyhow!("hot: reading journal {}: {e}", path.display()))?;
    if bytes.is_empty() {
        return Ok((Vec::new(), Vec::new()));
    }
    let reader = match StreamReader::try_new(std::io::Cursor::new(&bytes), None) {
        Ok(r) => r,
        // A segment whose schema message never landed carries nothing.
        Err(_) => return Ok((Vec::new(), Vec::new())),
    };
    let mut paths = Vec::new();
    let mut locs = Vec::new();
    let mut complete: Vec<arrow::record_batch::RecordBatch> = Vec::new();
    for batch in reader {
        match batch {
            Ok(b) => complete.push(b),
            Err(_) => break, // torn tail
        }
    }
    for b in &complete {
        let mut ipc: Vec<u8> = Vec::new();
        {
            let mut w = StreamWriter::try_new(&mut ipc, b.schema().as_ref())
                .map_err(|e| anyhow!("hot: re-encode: {e}"))?;
            w.write(b).map_err(|e| anyhow!("hot: re-encode write: {e}"))?;
            w.finish().map_err(|e| anyhow!("hot: re-encode finish: {e}"))?;
        }
        let (p, l) = decode_base_rows(&ipc)?;
        paths.extend(p);
        locs.extend(l);
    }
    Ok((paths, locs))
}

/// A newly-created or removed file is only durable once its directory entry is.
fn sync_parent_dir(path: &Path) {
    let parent = match path.parent() {
        Some(p) if !p.as_os_str().is_empty() => p.to_path_buf(),
        _ => PathBuf::from("."),
    };
    if let Ok(f) = File::open(parent) {
        let _ = f.sync_all();
    }
}

/// `HotArchive::seal` as a free function, for a caller that has a path and not a
/// handle. Refuses an archive with no journal rather than silently producing an
/// empty one.
pub fn seal_hot(archive: &Path, level: i32, policy: crate::SkipPolicy) -> Result<HotSealReport> {
    if !HotArchive::is_hot(archive) {
        bail!("{} has no journal; it is not hot", archive.display());
    }
    HotArchive::open(archive, level, policy)?.seal(None)
}

// ── tests ────────────────────────────────────────────────────────────────────
// Every test here seals a real archive through a `CompressCtx`, so all of them
// need the codec — the same gate `meta_sink_append::tests` carries.
#[cfg(all(test, feature = "openzl"))]
mod tests {
    use super::*;
    use crate::{ZnippyArchive, create_archive};
    use std::time::{SystemTime, UNIX_EPOCH};

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

    /// Deterministic, lexicographically spread, and *compressible* — so the codec
    /// really runs and a byte-identity claim is about compressed output, not about
    /// two stored-raw copies of the same input.
    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) % 997;
                let p = format!("repo/grp{g:03}/file{i:08}_{salt}.bin");
                let body = format!("payload {i} salt {salt} {}\n", "z".repeat(8 + (i % 40)));
                (p, body.into_bytes())
            })
            .collect()
    }

    /// **The correctness claim the whole module rests on.** Appending N files one
    /// at a time through the hot journal and then sealing must produce the SAME
    /// BYTES as creating the archive from all N at once — same blob offsets, same
    /// data sub-index, same sorted lookup, same fst trie, same manifest, same
    /// footer.
    ///
    /// This is an applied-output assertion, not a round trip: a seal that dropped
    /// the trie, or emitted rows in journal-segment order instead of insertion
    /// order, would still list N files and still read every one of them back
    /// correctly. Only the bytes catch it.
    ///
    /// Seen RED by emitting the rows sorted instead of in insertion order in
    /// `HotArchive::seal` — the data sub-index differs and the assertion fires
    /// with a first-difference offset inside it.
    #[test]
    fn seal_matches_create_archive_byte_for_byte() {
        let dir = unique_dir("identity");
        let files = synth(400, 11);

        let reference = dir.join("ref.znippy");
        create_archive(&reference, &files, 3).unwrap();

        let hot = dir.join("hot.znippy");
        {
            let mut h = HotArchive::open(&hot, 3, crate::SkipPolicy::resolve()).unwrap();
            for f in &files {
                h.append(std::slice::from_ref(f)).unwrap();
            }
            let report = h.seal(None).unwrap();
            assert_eq!(report.rows_sealed, files.len() as u64);
            assert_eq!(report.segments_removed, 1, "one process, one segment");
        }

        let a = std::fs::read(&reference).unwrap();
        let b = std::fs::read(&hot).unwrap();
        assert_eq!(
            a.len(),
            b.len(),
            "sealed length differs: create_archive {} vs hot-then-seal {}",
            a.len(),
            b.len()
        );
        let first_diff = a.iter().zip(&b).position(|(x, y)| x != y);
        assert!(
            first_diff.is_none(),
            "hot-then-seal is NOT byte-identical to create_archive; first difference at byte {:?}",
            first_diff
        );

        // …and the journal is gone, so the archive is genuinely cold.
        assert!(!HotArchive::is_hot(&hot), "seal must remove the journal");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// The hot archive is READABLE before it is sealed — otherwise a two-phase
    /// tier would be write-only until compaction, which is not "worth having".
    #[test]
    fn a_hot_archive_serves_reads_before_the_seal() {
        let dir = unique_dir("read");
        let path = dir.join("a.znippy");
        let files = synth(120, 3);
        let mut h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
        for f in &files {
            h.append(std::slice::from_ref(f)).unwrap();
        }
        for (p, bytes) in &files {
            assert!(h.contains(p), "hot archive must contain {p}");
            assert_eq!(h.file_size(p), Some(bytes.len() as u64));
            assert_eq!(&h.extract_file(p).unwrap(), bytes, "hot read mismatch for {p}");
        }
        assert_eq!(h.list_files().unwrap().len(), files.len());

        // And after the seal the ORDINARY static reader answers the same bytes.
        h.seal(None).unwrap();
        let ar = ZnippyArchive::open(&path).unwrap();
        for (p, bytes) in &files {
            assert_eq!(&ar.extract_file(p).unwrap(), bytes, "sealed read mismatch for {p}");
        }
        // Random access through the rebuilt trie+lookup.
        assert!(!crate::locate_file(&path, &files[77].0).unwrap().is_empty());
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A restart must not lose the journal, and must not corrupt the blob cursor.
    /// Three processes' worth of appends (three segments), then one seal.
    #[test]
    fn reopening_recovers_every_segment() {
        let dir = unique_dir("reopen");
        let path = dir.join("a.znippy");
        let mut all: Vec<(String, Vec<u8>)> = Vec::new();
        for round in 0..3u64 {
            let files = synth(40, round + 1);
            let mut h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
            assert_eq!(h.rows(), all.len() as u64, "reopen must recover prior rows");
            for f in &files {
                h.append(std::slice::from_ref(f)).unwrap();
            }
            all.extend(files);
        }
        let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
        assert_eq!(h.rows(), all.len() as u64);
        let report = h.seal(None).unwrap();
        assert_eq!(report.segments_removed, 4, "three appending opens + the final one");

        let ar = ZnippyArchive::open(&path).unwrap();
        for (p, bytes) in &all {
            assert_eq!(&ar.extract_file(p).unwrap(), bytes, "byte mismatch after restarts for {p}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A SEALED archive goes hot again without losing anything: the static rows
    /// are recovered into segment 0 and the rebuildable tail is dropped. This is
    /// the path gunnar's cold tier takes after a `repack()` fold.
    #[test]
    fn a_sealed_archive_reheats_and_keeps_its_rows() {
        let dir = unique_dir("reheat");
        let path = dir.join("a.znippy");
        let sealed = synth(60, 5);
        create_archive(&path, &sealed, 3).unwrap();
        let cold_len = std::fs::metadata(&path).unwrap().len();

        let mut h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
        assert_eq!(h.rows(), sealed.len() as u64, "re-heat must recover the sealed rows");
        assert!(
            h.blob_end() < cold_len,
            "the metadata tail must be dropped: blob_end {} vs sealed length {cold_len}",
            h.blob_end()
        );
        let extra = synth(20, 6);
        for f in &extra {
            h.append(std::slice::from_ref(f)).unwrap();
        }
        h.seal(None).unwrap();

        let ar = ZnippyArchive::open(&path).unwrap();
        for (p, bytes) in sealed.iter().chain(extra.iter()) {
            assert_eq!(&ar.extract_file(p).unwrap(), bytes, "byte mismatch after re-heat for {p}");
        }
        let mut listed = ar.list_files().unwrap();
        listed.sort();
        let mut want: Vec<String> =
            sealed.iter().chain(extra.iter()).map(|(p, _)| p.clone()).collect();
        want.sort();
        assert_eq!(listed, want);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Last writer wins, exactly as `append_files` does — one live row per path,
    /// and the bytes read back are the NEW ones. The old blob stays in the file as
    /// dead payload, which is the documented cost, so the assertion is on the row
    /// count and the bytes, not on the file size.
    #[test]
    fn re_appending_a_path_replaces_it() {
        let dir = unique_dir("replace");
        let path = dir.join("a.znippy");
        let mut h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
        h.append(&[("x.bin".into(), b"first version".to_vec())]).unwrap();
        h.append(&[("y.bin".into(), b"other".to_vec())]).unwrap();
        let r = h.append(&[("x.bin".into(), b"SECOND version, longer".to_vec())]).unwrap();
        assert_eq!(r.rows_replaced, 1, "re-appending x.bin must replace, not duplicate");
        assert_eq!(h.rows(), 2);
        assert_eq!(h.extract_file("x.bin").unwrap(), b"SECOND version, longer");
        h.seal(None).unwrap();

        let ar = ZnippyArchive::open(&path).unwrap();
        assert_eq!(ar.list_files().unwrap().len(), 2, "the seal must carry ONE row per path");
        assert_eq!(ar.extract_file("x.bin").unwrap(), b"SECOND version, longer");
        assert_eq!(ar.extract_file("y.bin").unwrap(), b"other");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A killed process leaves a partial final message. Everything before it is
    /// real and must survive; the partial one must be dropped rather than failing
    /// the open. Simulated by truncating the segment inside its last message.
    #[test]
    fn a_torn_journal_tail_is_dropped_not_fatal() {
        let dir = unique_dir("torn");
        let path = dir.join("a.znippy");
        let files = synth(30, 2);
        {
            let mut h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
            for f in &files {
                h.append(std::slice::from_ref(f)).unwrap();
            }
        }
        let seg = segment_path(&path, 0);
        let len = std::fs::metadata(&seg).unwrap().len();
        // Cut 40 bytes: inside the last batch message, past many complete ones.
        std::fs::OpenOptions::new().write(true).open(&seg).unwrap().set_len(len - 40).unwrap();

        let h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
        let rows = h.rows();
        assert!(
            rows > 0 && rows < files.len() as u64,
            "a torn tail must cost SOME rows and not all of them; got {rows} of {}",
            files.len()
        );
        // Every surviving row still reads back byte-exact.
        for p in h.list_files().unwrap() {
            let want = &files.iter().find(|(q, _)| *q == p).unwrap().1;
            assert_eq!(&h.extract_file(&p).unwrap(), want, "torn-journal read mismatch for {p}");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// **The O(1) claim, as an applied output.** The index-side cost of an append
    /// is the journal bytes it writes. If that is flat across a 100× growth in
    /// archive size, the append is not paying for the archive; if the old
    /// re-sealing path were behind this API the number would grow with N.
    ///
    /// Bytes, not CPU: instructions and wall clock are unusable as absolutes on a
    /// contended box, and a byte count is exact.
    #[test]
    fn the_index_cost_of_an_append_does_not_grow_with_the_archive() {
        let dir = unique_dir("flat");
        let path = dir.join("a.znippy");
        let mut h = HotArchive::open(&path, 3, crate::SkipPolicy::resolve()).unwrap();
        let mut early = 0u64;
        let mut late = 0u64;
        for i in 0..2_000usize {
            // Constant-shape rows so the comparison is about N, not about payload.
            let f = vec![(format!("pack-{i:040x}.pack"), vec![b'q'; 64])];
            let r = h.append(&f).unwrap();
            if i == 10 {
                early = r.journal_bytes_added;
            }
            if i == 1_999 {
                late = r.journal_bytes_added;
            }
        }
        assert!(early > 0, "an append must cost SOME journal bytes; the probe read 0");
        assert_eq!(
            early, late,
            "the index cost of an append grew from {early} B at row 10 to {late} B at row 2000 — \
             the append is paying for the archive again"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }
}