znippy-compress 0.9.12

//! Two-pass directory compression.
//!
//! Pass 1 — BIG files (> slice_size): sequential chunked reads, metadata by re-reading file.
//! Pass 2 — SMALL files (≤ slice_size): read into slot, metadata from in-memory data.
//!
//! Both passes ride the shared no-barrier **gatling** ordered-streaming-sink
//! engine ([`znippy_zoomies::gatling::ordered::run_ordered_sink`]), exactly like
//! the streaming compressor (`stream_packer.rs`). The reader thread stays the I/O
//! producer — it fills `Magazine` slots (io_uring batched for small files) and
//! yields `Round`s — while the gatling engine replaces the old hand-rolled
//! `thread::spawn` compress-worker pool + writer thread with N self-dispatching
//! map workers and an ordered sink. rayon is forbidden in the constellation; this
//! is the pool. Zero-copy is preserved: the skip / incompressible path hands the
//! slot `Round` straight to the sink, which pwrites from slot memory and releases
//! it — no memcpy out of the slot.
//!
//! Arrow IPC index is written incrementally — each pass writes its batch as soon as it finishes.
//! No accumulation, no merge.

use anyhow::{Result, anyhow, ensure};
use crossbeam_channel::{Receiver, bounded, unbounded};
use std::cell::RefCell;
use std::fs::File;
use std::io::{self, BufReader, Read};
use std::os::unix::fs::FileExt;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::thread;
use walkdir::WalkDir;

use znippy_zoomies::gatling::ordered::{OrderedSink, run_ordered_sink};

use znippy_common::codec::CompressCtx;
use znippy_common::common_config::CONFIG;
use znippy_common::index::{
    FileExtMeta, build_arrow_metadata_for_config,
    build_metadata_batch, compose_index_schema,
};
use znippy_common::meta::{BlobMeta, ChunkMeta};
use znippy_common::precompressed::{SNIFF_PREFIX_LEN, SkipPolicy, is_zlib_stream, looks_compressed};
use znippy_common::slotpool::{Magazine, PoolPlan};
use znippy_common::CompressionReport;
use znippy_common::{ArchiveMetaSink, ArrowIpcSink, GroupKey};

// The pool geometry is no longer a pair of hardcoded consts here: a
// `Magazine::new(8, 200 MiB, _)` reserved 1.6 GiB before a single file was
// opened, whatever the input and whatever the memory limit — which OOM-kills a
// constrained pod and wastes 1.6 GiB on a 5 MiB staging tree. It is now planned
// per pass by `PoolPlan::plan(input_bytes, num_workers, budget)`.
//
// The plan keeps `slice_size` pinned to the value the old constants implied
// (`200 MiB / num_workers`), because slice_size is the ONLY output-affecting
// part: it is the big/small partition threshold and the big-file cut length, so
// it lands in every chunk boundary and checksum. Only the reservation moves, so
// a memory-shrunk run writes a byte-identical archive — and when neither the
// input nor the budget binds, `plan` returns the historical geometry exactly.

/// TEST SEAM: when true, `run_small_pass` skips io_uring init and takes the
/// graceful fallback path even on a kernel where io_uring is available. This
/// lets the inject-assert test force the fallback without making the kernel
/// fail. Production code never sets this; it defaults to `false`.
#[cfg(test)]
static FORCE_SMALL_FALLBACK: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

/// TEST SEAM: when non-zero, overrides the slot-pool memory budget for the next
/// `compress_dir` call, so a test can drive the constrained-pod plan without
/// mutating process-global environment or needing a cgroup. Production code
/// never sets this; `pool_budget()` then reads the real budget.
#[cfg(test)]
static POOL_BUDGET_OVERRIDE: AtomicU64 = AtomicU64::new(0);

/// The slot-pool memory budget for this run — the real detected one, unless a
/// test has injected an override. In non-test builds this is exactly
/// `slot_pool_budget_bytes()`.
#[inline(always)]
fn pool_budget() -> u64 {
    #[cfg(test)]
    {
        let forced = POOL_BUDGET_OVERRIDE.load(Ordering::Relaxed);
        if forced != 0 {
            return forced;
        }
    }
    znippy_common::common_config::slot_pool_budget_bytes()
}

/// Returns whether the small-file pass should force the io_uring fallback.
/// In non-test builds this is a const `false` and optimises away entirely, so
/// there is no production-path overhead or behaviour change.
#[inline(always)]
fn force_small_fallback() -> bool {
    #[cfg(test)]
    {
        FORCE_SMALL_FALLBACK.load(Ordering::Relaxed)
    }
    #[cfg(not(test))]
    {
        false
    }
}

/// What the writer pwrites for one chunk.
enum Payload {
    /// Compressed output owned by the worker; recycled (dropped) after pwrite.
    Buf(Vec<u8>),
    /// PERF (Law 1): zero-copy skip/incompressible path. The bytes live in the
    /// slot; the writer pwrites straight from the slot via `Round::as_slice()`
    /// and releases the slot AFTER the pwrite. No memcpy out of the slot.
    Slot(znippy_common::slotpool::Round),
}

struct WriteJob {
    payload: Payload,
    on_disk_len: usize,
    file_index: u64,
    fdata_offset: u64,
    chunk_seq: u32,
    checksum: [u8; 32],
    compressed: bool,
    uncompressed_size: u64,
}

fn read_fully<R: Read>(r: &mut R, buf: &mut [u8]) -> io::Result<usize> {
    let mut n = 0;
    while n < buf.len() {
        match r.read(&mut buf[n..])? {
            0 => break,
            k => n += k,
        }
    }
    Ok(n)
}

thread_local! {
    /// One OpenZL context + one reusable compress-scratch buffer **per gatling map
    /// worker thread** (the workers are persistent, so this is created once per
    /// worker and reused across every `Round` it compresses). Keeps the hot path
    /// allocation-free — exactly what the old per-worker `cctx` / `reuse_buf` did.
    static COMPRESS_TLS: RefCell<Option<(CompressCtx, Vec<u8>)>> =
        const { RefCell::new(None) };
}

/// The gatling ordered sink: pwrites each in-order chunk at the next archive
/// offset and records its `BlobMeta`. Runs on the pass's calling thread, so it
/// owns the archive cursor directly — no atomic, no writer thread. Mirrors
/// `stream_packer::ArchiveSink`.
struct ArchiveSink {
    file: Arc<File>,
    cursor: u64,
    blobs: Vec<BlobMeta>,
    returner: znippy_common::slotpool::Ejector,
}

impl OrderedSink<Result<WriteJob>> for ArchiveSink {
    fn emit(&mut self, _seq: u64, output: Result<WriteJob>) -> Result<()> {
        let job = output?;
        let off = self.cursor;
        self.cursor += job.on_disk_len as u64;
        match &job.payload {
            Payload::Buf(buf) => {
                self.file.write_all_at(&buf[..job.on_disk_len], off)?;
            }
            Payload::Slot(round) => {
                // PERF (Law 1): pwrite straight from the slot, then release it so
                // the reader can reuse it. The slot stayed alive through the map
                // worker's hand-off precisely so this copy never happened.
                let src = unsafe { round.as_slice() };
                self.file.write_all_at(&src[..job.on_disk_len], off)?;
                self.returner.release_one(round.slot_id);
            }
        }
        self.blobs.push(BlobMeta {
            chunk_meta: ChunkMeta {
                fdata_offset: job.fdata_offset,
                file_index: job.file_index,
                chunk_seq: job.chunk_seq,
                checksum: job.checksum,
                compressed: job.compressed,
                uncompressed_size: job.uncompressed_size,
                compressed_size: job.on_disk_len as u64,
            },
            blob_offset: off,
            blob_size: job.on_disk_len as u64,
        });
        Ok(())
    }
}

/// Drive one pass on the shared gatling ordered engine: the `reader` thread has
/// already begun filling slots and publishing `Round`s into `rx_slice`; here we
/// pull them lazily (producer), compress across N no-barrier map workers, and
/// pwrite the outputs in strict producer order via [`ArchiveSink`]. Returns the
/// pass's `BlobMeta`s and advances `out_cursor` to the pass's end offset.
fn drive_pass(
    rx_slice: Receiver<znippy_common::slotpool::Round>,
    file: &Arc<File>,
    out_cursor: &Arc<AtomicU64>,
    returner: znippy_common::slotpool::Ejector,
    num_workers: usize,
    policy: SkipPolicy,
) -> Result<Vec<BlobMeta>> {
    let level = CONFIG.compression_level;
    let start = out_cursor.load(Ordering::Relaxed);

    // Producer: drain the reader's `Round` channel one item at a time, lazily
    // (pulled only when a worker slot is free). The reader thread remains the I/O
    // producer; `None` (channel closed) ends the stream. The label is unused — a
    // `Round` already carries file_index / fdata_offset / chunk_seq / skip.
    let producer = move || rx_slice.recv().ok().map(|r| ((), r));

    // Map: BLAKE3 over the original bytes, then compress (or store raw). Mirrors
    // the old `spawn_workers` loop, on the shared engine.
    let ret_map = returner.clone();
    let map = move |_label: (), round: znippy_common::slotpool::Round| -> Result<WriteJob> {
        let src = unsafe { round.as_slice() };
        let checksum = *blake3::hash(src).as_bytes();
        let len = src.len();
        let usz = len as u64;
        let (file_index, fdata_offset, chunk_seq) =
            (round.file_index, round.fdata_offset, round.chunk_seq);

        // The `skip` on the `Round` was decided at enumeration time, from the
        // path alone — it cannot see an entry that carries no extension (a git
        // loose object is named for its oid) nor one whose name lies. Now that the
        // real bytes are in hand, refine it with a magic-byte probe over the head
        // of the chunk. Bounded at `SNIFF_PREFIX_LEN` compares, and a no-op when
        // the path already said skip or the batch carries an explicit hint.
        let skip = round.skip
            || policy.skip_by_bytes(&src[..len.min(SNIFF_PREFIX_LEN)]);

        if skip {
            // PERF (Law 1): zero-copy skip path. Hand the slot `Round` straight to
            // the sink — it pwrites from slot bytes and releases the slot afterward.
            return Ok(WriteJob {
                payload: Payload::Slot(round),
                on_disk_len: len,
                file_index,
                fdata_offset,
                chunk_seq,
                checksum,
                compressed: false,
                uncompressed_size: usz,
            });
        }

        COMPRESS_TLS.with(|cell| -> Result<WriteJob> {
            let mut guard = cell.borrow_mut();
            if guard.is_none() {
                *guard = Some((CompressCtx::new(level)?, Vec::new()));
            }
            let (cctx, scratch) = guard.as_mut().unwrap();
            let n = cctx.compress_into(src, scratch)?;
            if n >= len {
                // Incompressible: storing raw is no bigger and skips the decode
                // cost. Zero-copy straight from the slot; `scratch` keeps its
                // capacity for the next chunk (no realloc).
                Ok(WriteJob {
                    payload: Payload::Slot(round),
                    on_disk_len: len,
                    file_index,
                    fdata_offset,
                    chunk_seq,
                    checksum,
                    compressed: false,
                    uncompressed_size: usz,
                })
            } else {
                // Compressed: the source bytes are no longer needed, so release the
                // slot now (this `Round` is done with slot memory). ZERO-ALLOC: hand
                // the already-filled `scratch` buffer to the sink via `take` — no
                // alloc-a-new-buffer-and-memcpy; `compress_into` truncated it to `n`.
                ret_map.release_one(round.slot_id);
                Ok(WriteJob {
                    payload: Payload::Buf(std::mem::take(scratch)),
                    on_disk_len: n,
                    file_index,
                    fdata_offset,
                    chunk_seq,
                    checksum,
                    compressed: true,
                    uncompressed_size: usz,
                })
            }
        })
    };

    // Sink: ordered, streaming pwrite + incremental BlobMeta, on this thread.
    let mut sink = ArchiveSink {
        file: Arc::clone(file),
        cursor: start,
        blobs: Vec::new(),
        returner,
    };
    // In-flight / reorder-buffer bound — matches the old bounded slice-channel depth.
    let cap = num_workers * 4;
    let sink_result = run_ordered_sink(producer, num_workers, cap, map, &mut sink);

    // test-matrix emit: HONEST verdict of the ordered-sink compress run (the
    // slot-packed small-file path draining onto the shared gatling engine).
    #[cfg(feature = "testmatrix")]
    crate::functional_status(
        "znippy-compress/slot_packer",
        "run_ordered_sink",
        sink_result.is_ok(),
        &format!(
            "workers={num_workers} cap={cap} blobs={} ok={}",
            sink.blobs.len(),
            sink_result.is_ok()
        ),
    );
    sink_result?;

    out_cursor.store(sink.cursor, Ordering::Relaxed);
    Ok(sink.blobs)
}

/// Compress a directory tree into a `.znippy` archive.
///
/// `no_skip = true` forces the codec over everything. For anything richer — a
/// caller that already knows its input is compressed — use
/// [`compress_dir_with_policy`].
pub fn compress_dir(
    input_dir: &PathBuf,
    output: &PathBuf,
    no_skip: bool,
    plugin: Option<&znippy_common::plugin::PluginRegistry>,
    repo: Option<&str>,
    sink_factory: Option<znippy_common::MetaSinkFactory>,
) -> Result<CompressionReport> {
    compress_dir_with_policy(
        input_dir, output, SkipPolicy::from_no_skip(no_skip), plugin, repo, sink_factory,
    )
}

/// [`compress_dir`] with an explicit [`SkipPolicy`].
///
/// The policy is batch-level, because that is how the knowledge arrives: a caller
/// sealing a directory of packfiles knows it for the whole directory.
/// [`SkipPolicy::already_compressed`] stores every entry raw without inspecting a
/// byte — deterministic, free, and better informed than any probe.
pub fn compress_dir_with_policy(
    input_dir: &PathBuf,
    output: &PathBuf,
    policy: SkipPolicy,
    plugin: Option<&znippy_common::plugin::PluginRegistry>,
    repo: Option<&str>,
    sink_factory: Option<znippy_common::MetaSinkFactory>,
) -> Result<CompressionReport> {
    let mut total_dirs = 0u64;
    let all_files: Arc<Vec<PathBuf>> = Arc::new(
        WalkDir::new(input_dir)
            .into_iter()
            .filter_map(|e| e.ok())
            .filter_map(|e| {
                if e.file_type().is_dir() {
                    total_dirs += 1;
                    None
                } else if e.file_type().is_file() {
                    Some(e.into_path())
                } else {
                    None
                }
            })
            .collect(),
    );
    let total_files = all_files.len() as u64;

    // FALSE-GREEN GUARD (mirrors the append path's `ensure!(file_count > 0, …)`):
    // an empty input dir — or a tree of only sub-directories / non-regular
    // entries — yields zero files, in which case neither pass runs and we would
    // otherwise return `Ok` over "0 files, 0 chunks" and light the compress
    // surface GREEN. Refuse instead, with a distinct "nothing to pack" error.
    ensure!(
        total_files > 0,
        "inga filer att komprimera hittades under {}",
        input_dir.display()
    );

    let ext_fields: Vec<znippy_common::arrow::datatypes::Field> =
        plugin.map(|r| r.schema_fields()).unwrap_or_default();

    let output_path = output.with_extension("znippy");
    let file = Arc::new(File::create(&output_path)?);
    let out_cursor = Arc::new(AtomicU64::new(0));

    let num_workers = CONFIG.max_core_in_flight.max(1);
    // Unchanged from the old `SLOT_SIZE / num_workers` — the partition threshold
    // and big-file cut length must not move (see the note on the removed consts).
    let slice_size = PoolPlan::slice_size_for(num_workers);

    // ── PARTITION ────────────────────────────────────────────────────────────
    // The walk already stats every file, so summing each pass's bytes here is
    // free — and it is what lets the slot pool be sized for the actual input
    // instead of an unconditional 1.6 GiB.
    let mut big_indices: Vec<usize> = Vec::new();
    let mut small_indices: Vec<usize> = Vec::new();
    let mut big_bytes = 0u64;
    let mut small_bytes = 0u64;
    for (i, path) in all_files.iter().enumerate() {
        let size = path.metadata().map(|m| m.len()).unwrap_or(0);
        if size > slice_size as u64 || size == 0 {
            big_indices.push(i);
            big_bytes += size;
        } else {
            small_indices.push(i);
            small_bytes += size;
        }
    }

    // One budget reading for the whole run (it probes /sys and /proc), shared by
    // both passes — which run strictly one after the other, so each may plan up
    // to the full budget.
    let pool_budget = pool_budget();

    let mut ext_meta: Vec<FileExtMeta> = vec![None; all_files.len()];
    let mut uncompressed_files = 0u64;
    let mut uncompressed_bytes = 0u64;
    let mut compressed_files = 0u64;
    let mut compressed_bytes = 0u64;
    let mut total_chunks = 0u64;

    // Metadata index schema (shared across both passes). The batches are
    // collected and handed to the metadata sink as one sub-index below.
    let meta_map = build_arrow_metadata_for_config(&CONFIG);
    let composed = compose_index_schema(&ext_fields);
    let schema_with_meta =
        arrow::datatypes::Schema::new_with_metadata(composed.fields().to_vec(), meta_map);
    let mut index_batches: Vec<arrow::record_batch::RecordBatch> = Vec::new();

    let input_dir_for_paths = input_dir.clone();
    let all_files_for_paths = Arc::clone(&all_files);

    // ══════════════════════════════════════════════════════════════════════════
    // PASS 1: BIG FILES
    // ══════════════════════════════════════════════════════════════════════════
    if !big_indices.is_empty() {
        let plan = PoolPlan::plan(big_bytes, num_workers, pool_budget);
        log::info!(
            "[slot_pool] big pass: {} slots × {} B = {} B reserved for {} B of input",
            plan.num_slots, plan.slot_size, plan.bytes(), big_bytes
        );
        let (uf, ub, cf, cb, blobs, meta) = run_big_pass(
            &all_files, input_dir, &big_indices, policy, plugin,
            &file, &out_cursor, num_workers, plan,
        )?;
        uncompressed_files += uf; uncompressed_bytes += ub;
        compressed_files += cf; compressed_bytes += cb;
        for (idx, m) in meta { if idx < ext_meta.len() { ext_meta[idx] = m; } }

        total_chunks += blobs.len() as u64;
        let all_f = Arc::clone(&all_files_for_paths);
        let inp = input_dir_for_paths.clone();
        let resolver = |file_index: u64| {
            let idx = file_index as usize;
            all_f[idx].strip_prefix(&inp).unwrap_or(&all_f[idx])
                .to_string_lossy().to_string()
        };
        let batch = build_metadata_batch(&blobs, resolver, &ext_meta, &ext_fields)
            .map_err(|e| anyhow!("big index batch: {e}"))?;
        index_batches.push(batch);
    }

    // ══════════════════════════════════════════════════════════════════════════
    // PASS 2: SMALL FILES
    // ══════════════════════════════════════════════════════════════════════════
    if !small_indices.is_empty() {
        let plan = PoolPlan::plan(small_bytes, num_workers, pool_budget);
        log::info!(
            "[slot_pool] small pass: {} slots × {} B = {} B reserved for {} B of input",
            plan.num_slots, plan.slot_size, plan.bytes(), small_bytes
        );
        let (uf, ub, cf, cb, blobs, meta) = run_small_pass(
            &all_files, input_dir, &small_indices, policy, plugin,
            &file, &out_cursor, num_workers, plan,
        )?;
        uncompressed_files += uf; uncompressed_bytes += ub;
        compressed_files += cf; compressed_bytes += cb;
        for (idx, m) in meta { if idx < ext_meta.len() { ext_meta[idx] = m; } }

        total_chunks += blobs.len() as u64;
        let all_f = Arc::clone(&all_files_for_paths);
        let inp = input_dir_for_paths.clone();
        let resolver = |file_index: u64| {
            let idx = file_index as usize;
            all_f[idx].strip_prefix(&inp).unwrap_or(&all_f[idx])
                .to_string_lossy().to_string()
        };
        let batch = build_metadata_batch(&blobs, resolver, &ext_meta, &ext_fields)
            .map_err(|e| anyhow!("small index batch: {e}"))?;
        index_batches.push(batch);
    }

    // ══════════════════════════════════════════════════════════════════════════
    // FINALIZE: write the metadata layer (one sub-index of all batches) via the sink
    // ══════════════════════════════════════════════════════════════════════════
    let index_offset = out_cursor.load(Ordering::Relaxed);
    let blob_bytes = index_offset;
    let pkg_type_val: i8 = plugin.and_then(|r| r.type_id()).unwrap_or(0);

    let mut sink: Box<dyn ArchiveMetaSink> = match sink_factory {
        Some(make) => make(Arc::clone(&file), blob_bytes),
        None => Box::new(ArrowIpcSink::new(Arc::clone(&file), blob_bytes)),
    };
    sink.push_subindex(
        &schema_with_meta,
        &index_batches,
        GroupKey {
            pkg_type: pkg_type_val,
            repo: repo.unwrap_or("").to_string(),
            module_name: String::new(),
        },
    )?;
    let total_bytes_out = sink.finish()?;

    // REAL committed result: `compressed_files + uncompressed_files` are now only
    // incremented once a file's bytes reached a committed slice (see the passes).
    // Anything the walk enumerated but that failed to open/read is the shortfall.
    let packed = compressed_files + uncompressed_files;
    let files_failed = total_files.saturating_sub(packed);

    Ok(CompressionReport {
        total_files,
        compressed_files,
        uncompressed_files,
        files_failed,
        chunks: total_chunks,
        total_dirs,
        total_bytes_in: compressed_bytes + uncompressed_bytes,
        total_bytes_out,
        compressed_bytes,
        uncompressed_bytes,
        compression_ratio: if uncompressed_bytes > 0 {
            (compressed_bytes as f32 / blob_bytes.max(1) as f32) * 100.0
        } else {
            0.0
        },
    })
}

/// Does this file head identify a container whose **whole** body is already
/// compressed, so the verdict may be carried to every slice of the file?
///
/// ## Why the big pass needs this at all
///
/// `drive_pass` probes offset 0 of **every chunk**. A container's magic is at
/// offset 0 of **chunk 0 only**, so the probe protects `1/chunks` of a big
/// file's bytes and the rest is handed to level-19 zstd, which returns a result
/// no smaller and is discarded by the `n >= len` branch. The waste never reaches
/// the archive — it is pure CPU, and it is invisible in the output.
///
/// MEASURED on oden 2026-08-03, extensionless zstd blobs, `slice_size` 6.90 MiB:
///
/// | object | chunks | user CPU | per MB |
/// |---|---|---|---|
/// | 6.0 MB (one chunk) | 1 | 0.02 s | 0.003 s/MB |
/// | 60.0 MB | 9 | 9.54 s | 0.159 s/MB |
///
/// Both archives are byte-identical (ratio 1.0000). 48x the CPU per byte for
/// nothing, from crossing the big/small threshold alone. At 200 MB it was 35.6 s
/// against 0.07 s for the same bytes named `.zst`, where the extension table
/// already makes exactly this whole-file claim.
///
/// ## Why the zlib probe is deliberately NOT carried
///
/// [`is_zlib_stream`] is a **two-byte** test — a method nibble, a window bound,
/// a flag bit and a mod-31 checksum. Its predicate was evaluated over **all
/// 65 536** two-byte heads, exhaustively rather than sampled: it accepts **32 of
/// them, exactly 1 in 2 048**. That is fine for its actual job, which is
/// one whole small file (a git loose object is named for its oid, has no
/// extension, and never reaches this pass — it is far under `slice_size`).
/// It is *not* fine as a claim about 8 GiB: one big compressible file in ~2 000
/// would be stored raw, and a silently larger archive is a worse failure than a
/// missed optimisation, because nothing reports it.
///
/// The magics this does carry are 3-8 bytes anchored at offset 0 and each names
/// a container format that is compressed end to end — the same claim
/// `is_probably_compressed` already makes from a `.gz` or `.zst` name, derived
/// from the bytes instead of from the name.
fn carries_file_wide(head: &[u8]) -> bool {
    looks_compressed(head) && !is_zlib_stream(head)
}

// ─────────────────────────────────────────────────────────────────────────────
// PASS 1: big files — sequential chunked reads, re-read for metadata
// ─────────────────────────────────────────────────────────────────────────────
fn run_big_pass(
    all_files: &Arc<Vec<PathBuf>>,
    input_dir: &PathBuf,
    big_indices: &[usize],
    policy: SkipPolicy,
    plugin: Option<&znippy_common::plugin::PluginRegistry>,
    file: &Arc<File>,
    out_cursor: &Arc<AtomicU64>,
    num_workers: usize,
    plan: PoolPlan,
) -> Result<(u64, u64, u64, u64, Vec<BlobMeta>, Vec<(usize, FileExtMeta)>)> {
    let num_slots = plan.num_slots;
    let pool = Magazine::from_plan(plan);
    let returner = pool.returner();
    let (tx_slice, rx_slice) = bounded(num_slots * 4);
    let (tx_meta, rx_meta) = unbounded::<(usize, FileExtMeta)>();

    let plugin_addr: usize = plugin.map(|p| p as *const _ as usize).unwrap_or(0);

    let reader = {
        let all_files = Arc::clone(all_files);
        let input_dir = input_dir.clone();
        let big_indices = big_indices.to_vec();
        let tx_meta = tx_meta.clone();
        thread::spawn(move || -> (u64, u64, u64, u64) {
            let plugin_ref: Option<&znippy_common::plugin::PluginRegistry> =
                if plugin_addr != 0 { Some(unsafe { &*(plugin_addr as *const _) }) } else { None };

            let mut uf = 0u64; let mut ub = 0u64;
            let mut cf = 0u64; let mut cb = 0u64;
            let mut cur = None;
            let ss = pool.slice_size();
            // PERF (Law 2): one buffer reused for the big-file metadata re-read,
            // instead of a fresh `std::fs::read` allocation per matching file.
            let mut meta_buf: Vec<u8> = Vec::new();

            for &file_index in &big_indices {
                let path = &all_files[file_index];
                let file_size = path.metadata().map(|m| m.len()).unwrap_or(0);
                let skip = policy.skip_by_path(path);
                // NB: the per-file uf/cf/ub/cb counters are NOT bumped here up
                // front. They are the archive's honest "files packed" count, so a
                // file is counted ONLY once its bytes reach a committed slice
                // below — an open/read failure must not inflate the green count.

                if file_size == 0 {
                    ensure_room(&pool, &tx_slice, &mut cur, 0);
                    cur.as_mut().unwrap().commit_slice(0, skip, file_index as u64, 0, 0);
                    // A zero-byte file legitimately produces a committed (empty)
                    // index entry, so it counts as packed.
                    if skip { uf += 1; } else { cf += 1; }
                    continue;
                }

                let f = match File::open(path) {
                    // open failed → the file is dropped; do NOT count it.
                    Ok(f) => f,
                    Err(e) => { log::warn!("[big] open {}: {}", path.display(), e); continue; }
                };
                let mut rdr = BufReader::new(f);
                let mut fdata_offset = 0u64;
                let mut chunk_seq = 0u32;
                let mut remaining = file_size;
                // The container verdict, decided once from the head of the file
                // and then true of every slice of it. See `carries_file_wide`.
                let mut skip = skip;

                while remaining > 0 {
                    let want = ss.min(remaining as usize);
                    ensure_room(&pool, &tx_slice, &mut cur, want);
                    let fill = cur.as_mut().unwrap();
                    let buf = fill.writable(want);
                    let got = match read_fully(&mut rdr, buf) {
                        Ok(g) => g,
                        Err(e) => { log::warn!("[big] read {}: {}", path.display(), e); break; }
                    };
                    if got == 0 { break; }
                    if chunk_seq == 0 && !skip {
                        let head = &buf[..got.min(SNIFF_PREFIX_LEN)];
                        // `skip_by_bytes` FIRST, so the batch hint keeps its veto
                        // in both directions — `--no-skip` must not be overruled
                        // by a magic. `carries_file_wide` then narrows it to the
                        // magics that describe a whole file.
                        if policy.skip_by_bytes(head) && carries_file_wide(head) {
                            skip = true;
                        }
                    }
                    fill.commit_slice(got, skip, file_index as u64, fdata_offset, chunk_seq);
                    fdata_offset += got as u64;
                    chunk_seq += 1;
                    remaining = remaining.saturating_sub(got as u64);
                    if got < want { break; }
                }

                // Count the file as packed only if it produced at least one
                // committed chunk. If the very first read failed (chunk_seq == 0)
                // the file opened but no bytes reached a blob → it is a silent
                // failure, left uncounted so `files_failed` reflects it.
                if chunk_seq > 0 {
                    if skip { uf += 1; ub += file_size; } else { cf += 1; cb += file_size; }
                } else {
                    log::warn!("[big] {}: no bytes committed, dropping from archive", path.display());
                }

                // Big file metadata: re-read the file (acceptable for large files).
                if let Some(reg) = plugin_ref {
                    let rel = path.strip_prefix(&input_dir).unwrap_or(path).to_string_lossy();
                    if reg.matches(&rel) {
                        // Reuse meta_buf: truncate then read the whole file into it.
                        meta_buf.clear();
                        match File::open(path)
                            .and_then(|mut f| f.read_to_end(&mut meta_buf))
                        {
                            Ok(_) => {
                                if let Some((tid, row)) = reg.extract_typed(&rel, &meta_buf) {
                                    tx_meta.send((file_index, Some((tid, row)))).ok();
                                }
                            }
                            Err(e) => log::warn!("[big] meta re-read {}: {}", path.display(), e),
                        }
                    }
                }
            }

            if let Some(fill) = cur.take() {
                for s in fill.publish() { tx_slice.send(s).ok(); }
            }

            // Drain: reclaim all slots to prove workers/writer are done with slot memory.
            for _ in 0..pool.num_slots() {
                if pool.claim().is_none() { break; }
            }

            drop(tx_slice);
            drop(tx_meta);
            drop(pool);
            (uf, ub, cf, cb)
        })
    };

    // Main-thread copy of tx_meta dropped so rx_meta closes once the reader's
    // clone is gone. The reader owns tx_slice; when it finishes and drops it, the
    // gatling producer sees the channel close and ends the stream.
    drop(tx_meta);

    // Gatling ordered engine (replaces the raw compress-worker pool + writer
    // thread): drains the reader's Rounds, compresses across N no-barrier workers,
    // and pwrites in producer order. Runs on this thread, concurrent with the reader.
    let blobs = drive_pass(rx_slice, file, out_cursor, returner, num_workers, policy)?;

    let (uf, ub, cf, cb) = reader.join().map_err(|_| anyhow!("big reader panicked"))?;

    let mut meta = Vec::new();
    while let Ok(m) = rx_meta.try_recv() { meta.push(m); }

    Ok((uf, ub, cf, cb, blobs, meta))
}

// ─────────────────────────────────────────────────────────────────────────────
// PASS 2: small files — read into slot, metadata from in-memory data
// ─────────────────────────────────────────────────────────────────────────────
fn run_small_pass(
    all_files: &Arc<Vec<PathBuf>>,
    input_dir: &PathBuf,
    small_indices: &[usize],
    policy: SkipPolicy,
    plugin: Option<&znippy_common::plugin::PluginRegistry>,
    file: &Arc<File>,
    out_cursor: &Arc<AtomicU64>,
    num_workers: usize,
    plan: PoolPlan,
) -> Result<(u64, u64, u64, u64, Vec<BlobMeta>, Vec<(usize, FileExtMeta)>)> {
    let num_slots = plan.num_slots;
    let pool = Magazine::from_plan(plan);
    let returner = pool.returner();
    let (tx_slice, rx_slice) = bounded(num_slots * 4);
    let (tx_meta, rx_meta) = unbounded::<(usize, FileExtMeta)>();

    let plugin_addr: usize = plugin.map(|p| p as *const _ as usize).unwrap_or(0);

    let reader = {
        let all_files = Arc::clone(all_files);
        let input_dir = input_dir.clone();
        let small_indices = small_indices.to_vec();
        let tx_meta = tx_meta.clone();
        thread::spawn(move || -> (u64, u64, u64, u64) {
            let plugin_ref: Option<&znippy_common::plugin::PluginRegistry> =
                if plugin_addr != 0 { Some(unsafe { &*(plugin_addr as *const _) }) } else { None };

            let mut uf = 0u64; let mut ub = 0u64;
            let mut cf = 0u64; let mut cb = 0u64;
            let mut cur = None;

            // io_uring may be unavailable (old kernel < 5.1, seccomp SCMP_ACT_KILL,
            // Docker with default seccomp). Try to initialise and fall back to
            // std::fs::read if creation fails.
            // io_uring may be unavailable. Also, the test seam can force the
            // fallback by leaving `ring` as None even when init would succeed.
            let mut ring: Option<io_uring::IoUring> = if force_small_fallback() {
                None
            } else {
                io_uring::IoUring::new(256).ok()
            };
            let mut idx = 0usize;
            let n = small_indices.len();

            // PERF (Law 2): per-batch scratch buffers hoisted out of the loop and
            // cleared each batch — no per-128-file reallocation of these four Vecs.
            let mut batch: Vec<(usize, usize, bool)> = Vec::with_capacity(128); // (file_index, size, skip)
            let mut cstrings: Vec<std::ffi::CString> = Vec::with_capacity(128);
            let mut fds: Vec<i32> = Vec::with_capacity(128);
            let mut offsets: Vec<usize> = Vec::with_capacity(128);
            let mut read_results: Vec<usize> = Vec::with_capacity(128);

            while idx < n {
                // Collect a batch that fits in current slot
                if cur.is_none() { cur = pool.claim(); }
                batch.clear();
                let mut batch_total = 0usize;

                while idx < n && batch.len() < 128 {
                    let file_index = small_indices[idx];
                    let path = &all_files[file_index];
                    let file_size = path.metadata().map(|m| m.len()).unwrap_or(0) as usize;
                    let skip = policy.skip_by_path(path);

                    let remaining = cur.as_ref().unwrap().remaining();
                    if batch_total + file_size > remaining {
                        if batch_total == 0 {
                            // Slot is too full for even one file — publish and get new slot
                            if let Some(fill) = cur.take() {
                                for s in fill.publish() { tx_slice.send(s).ok(); }
                            }
                            cur = pool.claim();
                            continue; // retry with new slot
                        }
                        break; // process what we have
                    }

                    // NB: uf/cf/ub/cb are NOT bumped here. A file is counted as
                    // packed only in Phase 4 below, once its read succeeded (or it
                    // is a legitimately empty file) — an open/read failure must
                    // not inflate the archive's honest "files packed" count.
                    batch.push((file_index, file_size, skip));
                    batch_total += file_size;
                    idx += 1;
                }

                if batch.is_empty() { continue; }

                let blen = batch.len();

                // Phase 2: batch read directly into slot via writable()
                // writable() gives a slice at cursor without advancing — use it for
                // the entire batch, then commit each file.
                let fill = cur.as_mut().unwrap();
                let slot_buf = fill.writable(batch_total);
                let slot_ptr = slot_buf.as_mut_ptr();

                offsets.clear();
                let mut off = 0usize;
                for &(_, size, _) in &batch {
                    offsets.push(off);
                    off += size;
                }

                read_results.clear();
                read_results.resize(blen, 0);

                if let Some(ref mut ring) = ring {
                    // Phase 1 (io_uring path): batch open
                    cstrings.clear();
                    for &(fi, _, _) in &batch {
                        let p = all_files[fi].as_os_str().as_encoded_bytes();
                        cstrings.push(unsafe { std::ffi::CString::from_vec_unchecked(p.to_vec()) });
                    }

                    fds.clear();
                    fds.resize(blen, -1);
                    // PERF: step over the range in 256-wide windows instead of
                    // materializing a throwaway index Vec just to `.chunks(256)`.
                    let mut start = 0usize;
                    while start < blen {
                        let end = (start + 256).min(blen);
                        for i in start..end {
                            let open_e = io_uring::opcode::OpenAt::new(
                                io_uring::types::Fd(libc::AT_FDCWD),
                                cstrings[i].as_ptr(),
                            )
                            .flags(libc::O_RDONLY | libc::O_CLOEXEC)
                            .build()
                            .user_data(i as u64);
                            unsafe { ring.submission().push(&open_e).ok(); }
                        }
                        let want = end - start;
                        ring.submit_and_wait(want).ok();
                        let mut got = 0;
                        while got < want {
                            if let Some(cqe) = ring.completion().next() {
                                fds[cqe.user_data() as usize] = cqe.result();
                                got += 1;
                            }
                        }
                        start = end;
                    }

                    // Phase 2 (io_uring path): batch read
                    // PERF: 256-wide windows via step, no throwaway index Vec.
                    let mut start = 0usize;
                    while start < blen {
                        let end = (start + 256).min(blen);
                        let mut to_submit = 0;
                        for i in start..end {
                            if fds[i] < 0 { continue; }
                            let (_, size, _) = batch[i];
                            let dst = unsafe { slot_ptr.add(offsets[i]) };
                            let read_e = io_uring::opcode::Read::new(
                                io_uring::types::Fd(fds[i]),
                                dst,
                                size as u32,
                            )
                            .build()
                            .user_data(i as u64);
                            unsafe { ring.submission().push(&read_e).ok(); }
                            to_submit += 1;
                        }
                        if to_submit > 0 {
                            ring.submit_and_wait(to_submit).ok();
                            let mut got = 0;
                            while got < to_submit {
                                if let Some(cqe) = ring.completion().next() {
                                    let i = cqe.user_data() as usize;
                                    read_results[i] = if cqe.result() > 0 { cqe.result() as usize } else { 0 };
                                    got += 1;
                                }
                            }
                        }
                        start = end;
                    }

                    // Phase 3 (io_uring path): close fds
                    for &fd in &fds {
                        if fd >= 0 { unsafe { libc::close(fd); } }
                    }
                } else {
                    // Fallback path: io_uring unavailable — use std::fs::read per file.
                    for i in 0..blen {
                        let (fi, size, _) = batch[i];
                        let path = &all_files[fi];
                        let dst = unsafe { std::slice::from_raw_parts_mut(slot_ptr.add(offsets[i]), size) };
                        if let Ok(mut f) = std::fs::File::open(path) {
                            // `size` comes from metadata().len(); read the whole file
                            // (looping past short reads) and record the true byte
                            // count, so the slot holds exactly `got` valid bytes —
                            // symmetric with the io_uring Read path.
                            read_results[i] = match read_fully(&mut f, dst) {
                                Ok(g) => g,
                                Err(e) => {
                                    log::warn!("[small] read {}: {}", path.display(), e);
                                    0
                                }
                            };
                        }
                    }
                }

                // Phase 4: plugin extraction + commit each file
                let fill = cur.as_mut().unwrap();
                for i in 0..batch.len() {
                    let (file_index, size, skip) = batch[i];
                    let got = read_results[i];

                    // Count as packed iff the file's bytes actually landed in the
                    // slot: `got > 0`, or the file is legitimately empty (size 0,
                    // which reads 0 bytes and commits an empty entry). A file with
                    // size > 0 but got == 0 failed to open/read (fd < 0 in the
                    // io_uring path, or a fallback read error) and is left
                    // uncounted so `files_failed` surfaces it.
                    if got > 0 || size == 0 {
                        if skip { uf += 1; ub += size as u64; }
                        else { cf += 1; cb += size as u64; }
                    } else {
                        log::warn!(
                            "[small] {}: read 0 of {} bytes, dropping from archive",
                            all_files[file_index].display(), size
                        );
                    }

                    if let Some(reg) = plugin_ref {
                        if got > 0 {
                            let path = &all_files[file_index];
                            let rel = path.strip_prefix(&input_dir).unwrap_or(path).to_string_lossy();
                            if reg.matches(&rel) {
                                let data = unsafe {
                                    std::slice::from_raw_parts(slot_ptr.add(offsets[i]), got)
                                };
                                if let Some((tid, row)) = reg.extract_typed(&rel, data) {
                                    tx_meta.send((file_index, Some((tid, row)))).ok();
                                }
                            }
                        }
                    }

                    fill.commit_slice(got, skip, file_index as u64, 0, 0);
                }
            }

            if let Some(fill) = cur.take() {
                for s in fill.publish() { tx_slice.send(s).ok(); }
            }

            for _ in 0..pool.num_slots() {
                if pool.claim().is_none() { break; }
            }

            drop(tx_slice);
            drop(tx_meta);
            drop(pool);
            (uf, ub, cf, cb)
        })
    };

    // Main-thread copy of tx_meta dropped so rx_meta closes once the reader's
    // clone is gone. The reader owns tx_slice; when it finishes and drops it, the
    // gatling producer sees the channel close and ends the stream.
    drop(tx_meta);

    // Gatling ordered engine (replaces the raw compress-worker pool + writer
    // thread): drains the reader's Rounds, compresses across N no-barrier workers,
    // and pwrites in producer order. Runs on this thread, concurrent with the reader.
    let blobs = drive_pass(rx_slice, file, out_cursor, returner, num_workers, policy)?;

    let (uf, ub, cf, cb) = reader.join().map_err(|_| anyhow!("small reader panicked"))?;

    let mut meta = Vec::new();
    while let Ok(m) = rx_meta.try_recv() { meta.push(m); }

    Ok((uf, ub, cf, cb, blobs, meta))
}

// ─────────────────────────────────────────────────────────────────────────────
// Shared helpers
// ─────────────────────────────────────────────────────────────────────────────

fn ensure_room<'p>(
    pool: &'p Magazine,
    tx_slice: &crossbeam_channel::Sender<znippy_common::slotpool::Round>,
    cur: &mut Option<znippy_common::slotpool::Clip<'p>>,
    need: usize,
) {
    loop {
        if cur.is_none() {
            *cur = pool.claim();
            if cur.is_none() { return; }
        }
        if cur.as_ref().unwrap().remaining() >= need { return; }
        let slices = cur.take().unwrap().publish();
        for s in slices { tx_slice.send(s).ok(); }
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Tests
// ─────────────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;
    use std::io::Write;

    /// Unique scratch dir under the system temp dir (no tempfile dev-dep here).
    /// Cleaned up by the caller on success; left in place on panic for triage.
    fn scratch(tag: &str) -> PathBuf {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let dir = std::env::temp_dir().join(format!(
            "znippy_fallback_test_{tag}_{}_{nanos}",
            std::process::id()
        ));
        fs::create_dir_all(&dir).unwrap();
        dir
    }

    /// Walk `root` and collect (relative-path, bytes) for every file, so two
    /// extracted trees can be compared byte-for-byte regardless of walk order.
    fn read_tree(root: &PathBuf) -> std::collections::BTreeMap<String, Vec<u8>> {
        let mut out = std::collections::BTreeMap::new();
        for e in WalkDir::new(root).into_iter().filter_map(|e| e.ok()) {
            if e.file_type().is_file() {
                let rel = e
                    .path()
                    .strip_prefix(root)
                    .unwrap()
                    .to_string_lossy()
                    .to_string();
                out.insert(rel, fs::read(e.path()).unwrap());
            }
        }
        out
    }

    /// INJECT-ASSERT (audit HIGH): the io_uring small-file path and the graceful
    /// fallback path MUST produce byte-identical archive output. We compress the
    /// SAME directory of several small files twice — once with io_uring, once
    /// with the fallback forced via the `FORCE_SMALL_FALLBACK` test seam — and
    /// assert that:
    ///   1. the per-chunk index checksums match exactly (the bytes the codec saw
    ///      were identical, i.e. the fallback read the same bytes into the slot),
    ///   2. the decompressed file trees are byte-for-byte identical,
    ///   3. both equal the original injected input bytes.
    ///
    /// This is a real input → real output assertion, not a "didn't panic" smoke
    /// test: every file carries distinct, deliberately-chosen bytes (including a
    /// multi-batch fan-out, an empty file, an already-compressed `.gz` that is
    /// stored raw, and a highly-compressible file) so a wrong byte, a short read,
    /// or a wrong length would change a checksum and fail the assertion.
    #[test]
    fn fallback_byte_identical_to_io_uring() {
        let input = scratch("input");

        // Inject a directory of several small files with distinct real bytes.
        // Many files so the small-pass batches them and the read loop runs hot.
        fs::write(input.join("alpha.txt"), b"the quick brown fox").unwrap();
        fs::write(input.join("beta.bin"), (0u8..=255).collect::<Vec<u8>>()).unwrap();
        fs::create_dir_all(input.join("nested/deep")).unwrap();
        fs::write(input.join("nested/gamma.txt"), b"nested content here").unwrap();
        fs::write(input.join("nested/deep/delta.dat"), vec![0xABu8; 4096]).unwrap();
        // Empty file: read must record 0 bytes, not short-read into garbage.
        fs::write(input.join("empty.txt"), b"").unwrap();
        // Already-compressed extension: stored raw at full I/O speed.
        fs::write(input.join("packed.gz"), vec![0x1F, 0x8B, 0x08, 0x00, 0x99, 0x42]).unwrap();
        // Highly compressible file: exercises the real codec path.
        fs::write(input.join("zeros.log"), vec![0u8; 9000]).unwrap();
        // A spray of small files to force multiple read iterations in one batch.
        for i in 0..40 {
            let mut f = fs::File::create(input.join(format!("frag_{i:03}.txt"))).unwrap();
            // distinct content per file so any swap/short-read flips a checksum
            writeln!(f, "fragment number {i} :: payload {}", "x".repeat(i)).unwrap();
        }

        let input = input; // PathBuf

        // ── Run A: io_uring path (seam OFF) ─────────────────────────────────
        FORCE_SMALL_FALLBACK.store(false, Ordering::SeqCst);
        let out_a_dir = scratch("out_a");
        let arc_a = out_a_dir.join("a.znippy");
        let report_a = compress_dir(&input, &arc_a, false, None, None, None)
            .expect("io_uring compress");
        let arc_a = arc_a.with_extension("znippy");

        // ── Run B: forced fallback path (seam ON) ───────────────────────────
        FORCE_SMALL_FALLBACK.store(true, Ordering::SeqCst);
        let out_b_dir = scratch("out_b");
        let arc_b = out_b_dir.join("b.znippy");
        let report_b = compress_dir(&input, &arc_b, false, None, None, None)
            .expect("fallback compress");
        let arc_b = arc_b.with_extension("znippy");
        FORCE_SMALL_FALLBACK.store(false, Ordering::SeqCst);

        // ── Assert 1: same file/chunk accounting ────────────────────────────
        assert_eq!(report_a.total_files, report_b.total_files, "file count differs");
        assert_eq!(report_a.chunks, report_b.chunks, "chunk count differs");
        assert_eq!(
            report_a.total_bytes_in, report_b.total_bytes_in,
            "input byte total differs"
        );

        // ── Assert 2: per-chunk index checksums are identical ───────────────
        // Read each archive's index and collect the set of (relative_path,
        // chunk_seq, checksum, uncompressed_size, compressed_size). If the
        // fallback read a single wrong/short byte, a blake3 checksum here would
        // differ and this set comparison would fail.
        let sig_a = chunk_signatures(&arc_a);
        let sig_b = chunk_signatures(&arc_b);
        assert_eq!(
            sig_a, sig_b,
            "io_uring vs fallback chunk checksums differ — NOT byte-identical"
        );

        // ── Assert 3: decompressed trees byte-identical, and == original ────
        let dec_a = scratch("dec_a");
        let dec_b = scratch("dec_b");
        let va = znippy_common::decompress_archive(&arc_a, true, &dec_a)
            .expect("decompress A");
        let vb = znippy_common::decompress_archive(&arc_b, true, &dec_b)
            .expect("decompress B");
        assert_eq!(va.corrupt_files, 0, "io_uring archive had corrupt files");
        assert_eq!(vb.corrupt_files, 0, "fallback archive had corrupt files");

        let tree_orig = read_tree(&input);
        let tree_a = read_tree(&dec_a);
        let tree_b = read_tree(&dec_b);
        assert_eq!(tree_a, tree_b, "decompressed trees differ between paths");
        assert_eq!(tree_a, tree_orig, "io_uring round-trip != original bytes");
        assert_eq!(tree_b, tree_orig, "fallback round-trip != original bytes");

        // Cleanup on success (left behind on panic for triage).
        for d in [&input, &out_a_dir, &out_b_dir, &dec_a, &dec_b] {
            let _ = fs::remove_dir_all(d);
        }
    }

    /// MEMORY BOUND, end to end: the same directory compressed under a starved
    /// 8 MiB slot-pool budget and under an unconstrained one must produce the
    /// SAME archive — same chunk boundaries, same checksums, same sizes — and
    /// round-trip to the same bytes.
    ///
    /// This is the assertion the fix hangs on. `compress_dir` used to reserve a
    /// hardcoded `8 × 200 MiB = 1.6 GiB` before opening a file, which OOM-kills a
    /// constrained pod; the reservation now follows the input and the budget. That
    /// is only safe because `slice_size` — the big/small partition threshold and
    /// the big-file cut length, hence every chunk boundary — is planned
    /// independently of the reservation. If a future change lets the memory plan
    /// move `slice_size`, the chunk signatures below diverge and this fails.
    #[test]
    fn starved_pool_budget_writes_a_byte_identical_archive() {
        let input = scratch("mem_input");

        // Small-pass content, plus an empty file (which the partition sends to
        // the BIG pass), so both pools are exercised.
        fs::write(input.join("empty.txt"), b"").unwrap();
        fs::write(input.join("alpha.txt"), b"the quick brown fox").unwrap();
        fs::write(input.join("beta.bin"), (0u8..=255).collect::<Vec<u8>>()).unwrap();
        fs::write(input.join("zeros.log"), vec![0u8; 9000]).unwrap();
        for i in 0..64 {
            let mut f = fs::File::create(input.join(format!("frag_{i:03}.txt"))).unwrap();
            writeln!(f, "fragment number {i} :: payload {}", "y".repeat(i * 3)).unwrap();
        }

        // A file that is genuinely CUT by the big pass — the only place slot
        // geometry could disturb chunk boundaries. slice_size scales with the
        // host's worker count, so only build it when that stays cheap (it is
        // ~7 MiB on a 32-core box; a 2-core box would want 100 MiB, which is not
        // worth a unit test — the `PoolPlan` unit test covers the invariant there).
        let slice = PoolPlan::slice_size_for(CONFIG.max_core_in_flight.max(1));
        let cut_file = slice <= 16 * 1024 * 1024;
        if cut_file {
            // 2.5 slices' worth, so it is cut into three chunks with a short tail.
            let n = slice * 5 / 2;
            let body: Vec<u8> = (0..n).map(|i| (i.wrapping_mul(31) % 251) as u8).collect();
            fs::write(input.join("big.bin"), &body).unwrap();
        }

        // ── Run A: unconstrained (the historical 1.6 GiB plan) ──────────────
        POOL_BUDGET_OVERRIDE.store(u64::MAX / 4, Ordering::SeqCst);
        let out_a = scratch("mem_out_a");
        let arc_a = out_a.join("a.znippy");
        let report_a = compress_dir(&input, &arc_a, false, None, None, None)
            .expect("unconstrained compress");
        let arc_a = arc_a.with_extension("znippy");

        // ── Run B: starved — 8 MiB, an order of magnitude under the old floor ─
        const STARVED: u64 = 8 * 1024 * 1024;
        POOL_BUDGET_OVERRIDE.store(STARVED, Ordering::SeqCst);
        let out_b = scratch("mem_out_b");
        let arc_b = out_b.join("b.znippy");
        let report_b = compress_dir(&input, &arc_b, false, None, None, None)
            .expect("starved compress");
        let arc_b = arc_b.with_extension("znippy");
        POOL_BUDGET_OVERRIDE.store(0, Ordering::SeqCst);

        // The starved run really did plan a smaller pool than the old constant —
        // otherwise the comparison below proves nothing.
        let workers = CONFIG.max_core_in_flight.max(1);
        let starved_plan = PoolPlan::plan(u64::MAX / 4, workers, STARVED);
        assert!(
            starved_plan.bytes() <= STARVED,
            "the starved plan reserved {} bytes, over its {STARVED}-byte budget",
            starved_plan.bytes()
        );
        assert!(
            starved_plan.bytes() * 100 < PoolPlan::default_for(workers).bytes(),
            "the starved plan ({} B) is not meaningfully smaller than the old 1.6 GiB",
            starved_plan.bytes()
        );

        // ── The archives are the same archive ───────────────────────────────
        assert_eq!(report_a.total_files, report_b.total_files, "file count differs");
        assert_eq!(report_a.chunks, report_b.chunks, "chunk count differs");
        assert_eq!(report_a.total_bytes_in, report_b.total_bytes_in, "input bytes differ");
        assert_eq!(
            chunk_signatures(&arc_a),
            chunk_signatures(&arc_b),
            "a starved slot-pool budget changed the archive — chunk boundaries or \
             checksums moved with the memory plan"
        );

        // ── …and both round-trip to the original bytes ──────────────────────
        let dec_a = scratch("mem_dec_a");
        let dec_b = scratch("mem_dec_b");
        let va = znippy_common::decompress_archive(&arc_a, true, &dec_a).expect("decompress A");
        let vb = znippy_common::decompress_archive(&arc_b, true, &dec_b).expect("decompress B");
        assert_eq!(va.corrupt_files, 0, "unconstrained archive had corrupt files");
        assert_eq!(vb.corrupt_files, 0, "starved archive had corrupt files");
        let tree_orig = read_tree(&input);
        assert_eq!(read_tree(&dec_a), tree_orig, "unconstrained round-trip != original");
        assert_eq!(read_tree(&dec_b), tree_orig, "starved round-trip != original");

        for d in [&input, &out_a, &out_b, &dec_a, &dec_b] {
            let _ = fs::remove_dir_all(d);
        }
    }

    // ─────────────────────────────────────────────────────────────────────────
    // FALSE-GREEN REGRESSION GUARDS (audit ★2 + ★8, false-green-hunt 2026-07-28)
    //
    // These prove the compress surface can no longer light GREEN
    // (`archive_written`) over work that did not happen: an input with zero
    // files, and an input where a file is silently dropped after an open/read
    // failure. Each asserts the EXACT `archive_written` gate that
    // `compress_reporting` uses, so a regression that re-inflates the count or
    // drops the empty-input guard turns these red.
    // ─────────────────────────────────────────────────────────────────────────

    /// The `archive_written` gate exactly as `znippy-cli`'s `compress_reporting`
    /// computes it. A green compress REQUIRES: files enumerated, at least one
    /// chunk committed, and NO file dropped after an open/read failure.
    fn archive_written_green(r: &CompressionReport) -> bool {
        r.total_files > 0 && r.chunks > 0 && r.files_failed == 0
    }

    /// ★2 — an EMPTY input directory must NOT return a green `Ok` report over
    /// "0 files, 0 chunks". Mirrors the append path's `ensure!(file_count > 0)`.
    /// Before the fix this returned `Ok(CompressionReport{total_files:0,…})` and
    /// the CLI lit `archive_written = true` — the walker-bug shape.
    #[test]
    fn empty_input_dir_refuses_green() {
        let input = scratch("empty_input");
        // Directory exists but contains no files at all.
        let out = scratch("empty_out");
        let arc = out.join("e.znippy");

        let res = compress_dir(&input, &arc, false, None, None, None);
        assert!(
            res.is_err(),
            "empty input dir must ERROR (nothing to pack), got a green report: {res:?}"
        );

        for d in [&input, &out] {
            let _ = fs::remove_dir_all(d);
        }
    }

    /// ★2 — an input tree of ONLY sub-directories (no regular files) is the same
    /// zero-work case: the walk yields 0 files → must ERROR, not green.
    #[test]
    fn only_dirs_no_files_refuses_green() {
        let input = scratch("only_dirs");
        fs::create_dir_all(input.join("a/b/c")).unwrap();
        fs::create_dir_all(input.join("d/e")).unwrap();
        let out = scratch("only_dirs_out");
        let arc = out.join("d.znippy");

        let res = compress_dir(&input, &arc, false, None, None, None);
        assert!(
            res.is_err(),
            "an all-directories input must ERROR (0 files), got: {res:?}"
        );

        for d in [&input, &out] {
            let _ = fs::remove_dir_all(d);
        }
    }

    /// ★8 (small pass) — a file that cannot be opened/read must NOT be counted as
    /// packed. We compress a dir of good small files plus ONE unreadable file
    /// (mode 000). The archive still has chunks (the good files packed), but:
    ///   * `files_failed >= 1`,
    ///   * `compressed_files + uncompressed_files == total_files - files_failed`
    ///     (the packed count excludes the dropped file), and
    ///   * the `archive_written` gate is RED — NOT green over an inflated count.
    /// Run under BOTH the io_uring path and the forced fallback, since each has
    /// its own open/read failure branch. Before the fix `uf/cf` were bumped up
    /// front, so `total_files` counted the dropped file and the surface was green.
    #[test]
    fn unreadable_small_file_not_counted_green() {
        use std::os::unix::fs::PermissionsExt;

        for &force_fallback in &[false, true] {
            let tag = if force_fallback { "unread_small_fb" } else { "unread_small_io" };
            let input = scratch(tag);

            // Good, readable small files → these DO pack and produce chunks.
            fs::write(input.join("good_a.txt"), b"alpha payload here").unwrap();
            fs::write(input.join("good_b.txt"), b"beta payload longer bytes").unwrap();
            for i in 0..8 {
                fs::write(input.join(format!("g_{i}.txt")), format!("file {i} bytes")).unwrap();
            }

            // The saboteur: real bytes (size > 0), then perms stripped so open
            // fails (EACCES) for us as owner → 0 bytes reach a blob.
            let bad = input.join("locked.bin");
            fs::write(&bad, vec![0x5Au8; 512]).unwrap();
            fs::set_permissions(&bad, fs::Permissions::from_mode(0o000)).unwrap();

            FORCE_SMALL_FALLBACK.store(force_fallback, Ordering::SeqCst);
            let out = scratch(&format!("{tag}_out"));
            let arc = out.join("a.znippy");
            let report = compress_dir(&input, &arc, false, None, None, None)
                .expect("compress should still succeed for the readable files");
            FORCE_SMALL_FALLBACK.store(false, Ordering::SeqCst);

            let packed = report.compressed_files + report.uncompressed_files;

            assert!(
                report.files_failed >= 1,
                "[fallback={force_fallback}] the unreadable file must count as failed, \
                 files_failed={}",
                report.files_failed
            );
            assert_eq!(
                packed,
                report.total_files - report.files_failed,
                "[fallback={force_fallback}] packed count must exclude the dropped file \
                 (total={}, failed={}, packed={})",
                report.total_files, report.files_failed, packed
            );
            assert!(
                report.chunks > 0,
                "[fallback={force_fallback}] the readable files should still have packed"
            );
            assert!(
                !archive_written_green(&report),
                "[fallback={force_fallback}] archive_written must be RED when a file was \
                 dropped — inflated green count regressed"
            );

            // Restore perms so cleanup can remove the file.
            let _ = fs::set_permissions(&bad, fs::Permissions::from_mode(0o644));
            for d in [&input, &out] {
                let _ = fs::remove_dir_all(d);
            }
        }
    }

    /// ★8 (big pass) — the same guarantee on the big-file path: an unreadable
    /// file larger than one slice must not inflate the packed count. Guarded to
    /// the small-slice hosts (like `starved_…`), so we never write a 100 MiB
    /// probe file on a low-core box; the small-pass test above covers the logic
    /// universally and `run_big_pass` is its structural mirror.
    #[test]
    fn unreadable_big_file_not_counted_green() {
        use std::os::unix::fs::PermissionsExt;

        let slice = PoolPlan::slice_size_for(CONFIG.max_core_in_flight.max(1));
        if slice > 16 * 1024 * 1024 {
            eprintln!("skipping big-pass unreadable test: slice_size {slice} too large here");
            return;
        }

        let input = scratch("unread_big");
        // A readable big file (> slice) → goes to the big pass and packs.
        let good: Vec<u8> = (0..(slice + 4096)).map(|i| (i * 7 % 251) as u8).collect();
        fs::write(input.join("good_big.bin"), &good).unwrap();
        // The saboteur: also > slice (so it is partitioned to the big pass), then
        // perms stripped so File::open fails and it is dropped from the archive.
        let bad = input.join("locked_big.bin");
        fs::write(&bad, vec![0x33u8; slice + 1]).unwrap();
        fs::set_permissions(&bad, fs::Permissions::from_mode(0o000)).unwrap();

        let out = scratch("unread_big_out");
        let arc = out.join("b.znippy");
        let report = compress_dir(&input, &arc, false, None, None, None)
            .expect("compress should still succeed for the readable big file");

        let packed = report.compressed_files + report.uncompressed_files;
        assert!(
            report.files_failed >= 1,
            "the unreadable big file must count as failed, files_failed={}",
            report.files_failed
        );
        assert_eq!(
            packed,
            report.total_files - report.files_failed,
            "big-pass packed count must exclude the dropped file (total={}, failed={})",
            report.total_files, report.files_failed
        );
        assert!(
            !archive_written_green(&report),
            "big-pass: archive_written must be RED when a big file was dropped"
        );

        let _ = fs::set_permissions(&bad, fs::Permissions::from_mode(0o644));
        for d in [&input, &out] {
            let _ = fs::remove_dir_all(d);
        }
    }

    /// Collect a sorted, order-independent signature of every chunk in an
    /// archive's index: (relative_path, chunk_seq, checksum, uncompressed_size,
    /// blob_size, compressed). Blob offsets are deliberately excluded — they are
    /// assigned in writer/thread completion order and so are legitimately
    /// non-deterministic, but the bytes (hence checksums) and sizes must not be.
    fn chunk_signatures(
        archive: &PathBuf,
    ) -> Vec<(String, u32, [u8; 32], u64, u64, bool)> {
        use arrow::array::{
            BooleanArray, FixedSizeBinaryArray, StringArray, UInt32Array, UInt64Array,
        };
        let (_schema, batches) = znippy_common::index::read_znippy_index_filtered(
            archive,
            &znippy_common::index::IndexFilter::default(),
        )
        .expect("read index");

        let mut sigs = Vec::new();
        for batch in &batches {
            let paths = batch
                .column_by_name("relative_path")
                .unwrap()
                .as_any()
                .downcast_ref::<StringArray>()
                .unwrap();
            let seqs = batch
                .column_by_name("chunk_seq")
                .unwrap()
                .as_any()
                .downcast_ref::<UInt32Array>()
                .unwrap();
            let sums = batch
                .column_by_name("checksum")
                .unwrap()
                .as_any()
                .downcast_ref::<FixedSizeBinaryArray>()
                .unwrap();
            let usz = batch
                .column_by_name("uncompressed_size")
                .unwrap()
                .as_any()
                .downcast_ref::<UInt64Array>()
                .unwrap();
            let bsz = batch
                .column_by_name("blob_size")
                .unwrap()
                .as_any()
                .downcast_ref::<UInt64Array>()
                .unwrap();
            let comp = batch
                .column_by_name("compressed")
                .unwrap()
                .as_any()
                .downcast_ref::<BooleanArray>()
                .unwrap();

            for i in 0..batch.num_rows() {
                let mut sum = [0u8; 32];
                sum.copy_from_slice(&sums.value(i)[..32]);
                sigs.push((
                    paths.value(i).to_string(),
                    seqs.value(i),
                    sum,
                    usz.value(i),
                    bsz.value(i),
                    comp.value(i),
                ));
            }
        }
        sigs.sort();
        sigs
    }
}