znippy-compress 0.9.6

//! 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};
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, should_skip_compression,
};
use znippy_common::meta::{BlobMeta, ChunkMeta};
use znippy_common::slotpool::Magazine;
use znippy_common::CompressionReport;
use znippy_common::{ArchiveMetaSink, ArrowIpcSink, GroupKey};

const SLOT_SIZE: usize = 200 * 1024 * 1024;
const NUM_SLOTS: usize = 8;

/// 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);

/// 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,
) -> 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);

        if round.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)
}

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> {
    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;

    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);
    let slice_size = SLOT_SIZE / num_workers.max(1);

    // ── PARTITION ────────────────────────────────────────────────────────────
    let mut big_indices: Vec<usize> = Vec::new();
    let mut small_indices: Vec<usize> = Vec::new();
    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);
        } else {
            small_indices.push(i);
        }
    }

    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 (uf, ub, cf, cb, blobs, meta) = run_big_pass(
            &all_files, input_dir, &big_indices, no_skip, plugin,
            &file, &out_cursor, num_workers,
        )?;
        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 (uf, ub, cf, cb, blobs, meta) = run_small_pass(
            &all_files, input_dir, &small_indices, no_skip, plugin,
            &file, &out_cursor, num_workers, slice_size,
        )?;
        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()?;

    Ok(CompressionReport {
        total_files,
        compressed_files,
        uncompressed_files,
        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
        },
    })
}

// ─────────────────────────────────────────────────────────────────────────────
// 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],
    no_skip: bool,
    plugin: Option<&znippy_common::plugin::PluginRegistry>,
    file: &Arc<File>,
    out_cursor: &Arc<AtomicU64>,
    num_workers: usize,
) -> Result<(u64, u64, u64, u64, Vec<BlobMeta>, Vec<(usize, FileExtMeta)>)> {
    let pool = Magazine::new(NUM_SLOTS, SLOT_SIZE, num_workers);
    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 = !no_skip && should_skip_compression(path);
                if skip { uf += 1; ub += file_size; } else { cf += 1; cb += file_size; }

                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);
                    continue;
                }

                let f = match File::open(path) {
                    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;

                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; }
                    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; }
                }

                // 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..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)?;

    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],
    no_skip: bool,
    plugin: Option<&znippy_common::plugin::PluginRegistry>,
    file: &Arc<File>,
    out_cursor: &Arc<AtomicU64>,
    num_workers: usize,
    _slice_size: usize,
) -> Result<(u64, u64, u64, u64, Vec<BlobMeta>, Vec<(usize, FileExtMeta)>)> {
    let pool = Magazine::new(NUM_SLOTS, SLOT_SIZE, num_workers);
    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 = !no_skip && should_skip_compression(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
                    }

                    if skip { uf += 1; ub += file_size as u64; }
                    else { cf += 1; cb += file_size as u64; }
                    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];

                    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..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)?;

    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);
        }
    }

    /// 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
    }
}