znippy-compress/src/slot_packer.rs
//! 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 `ChunkRevolver` coalescing slots (io_uring batched for small files) and
//! yields `PackedSlice`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 `PackedSlice` 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::poolplan::PoolPlan;
use znippy_zoomies::chunk_revolver::{ChunkRevolver, Filler, PackedSlice, Returner};
use znippy_common::CompressionReport;
use znippy_common::{ArchiveMetaSink, ArrowIpcSink, GroupKey};
// The pool geometry is no longer a pair of hardcoded consts here: a
// `ChunkRevolver::with_coalescing(8, 200 MiB, _)` would reserve 1.6 GiB before a 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(PackedSlice),
}
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: Returner,
}
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<PackedSlice>,
file: &Arc<File>,
out_cursor: &Arc<AtomicU64>,
returner: Returner,
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: PackedSlice| -> 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;
// The tree's SHAPE, collected on the walk that is already stat-ing everything.
// A directory owns no chunk, so it has no row in the data index and used to be
// counted into `total_dirs` and then thrown away — which lost every EMPTY
// directory on extract. Recording it here costs the walk nothing (walkdir has
// already stat-ed the entry) and is what `crate::fsmeta` restores. Sized
// lazily: a tree with no subdirectories seals no section at all.
let mut shape: Vec<znippy_common::fsmeta::FsEntry> = Vec::new();
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;
// The root itself has no relative path and needs no row — the
// caller chose the output directory.
if let Ok(rel) =
e.path().strip_prefix(input_dir).map_err(|_| ()).and_then(|r| {
if r.as_os_str().is_empty() { Err(()) } else { Ok(r) }
})
{
shape.push(znippy_common::fsmeta::FsEntry {
relative_path: rel.to_string_lossy().into_owned(),
kind: znippy_common::fsmeta::FsKind::Dir,
mode: e
.metadata()
.map(|m| znippy_common::fsmeta::dir_mode(&m))
.unwrap_or(0),
});
}
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;
// PERF (Law 2): one allocation for the whole tree's modes, sized from the walk.
let mut file_modes: Vec<(String, u32)> = Vec::with_capacity(all_files.len());
for (i, path) in all_files.iter().enumerate() {
// One `metadata()` for both jobs: the partition needs the length, and the
// shape section needs the mode. Splitting them would double the stat count
// on a tree of thousands of manifests for nothing.
let md = path.metadata();
let size = md.as_ref().map(|m| m.len()).unwrap_or(0);
// Collect the raw mode here — one `metadata()` serves both jobs, so the
// shape costs the walk nothing. `summarise_file_modes` then reduces the
// whole list to ONE archive-scoped default row plus the deviations, which
// is what keeps the section O(distinct modes) instead of paying a second
// copy of every path the lookup sub-index already holds.
if let (Some(mode), Ok(rel)) = (
md.as_ref().ok().and_then(znippy_common::fsmeta::file_mode),
path.strip_prefix(input_dir),
) {
file_modes.push((rel.to_string_lossy().into_owned(), mode));
}
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(),
},
)?;
// Seal the tree's shape as a RESERVED sub-index, so every data reader
// (`list`, `verify`, `get`, the typed views, DuckDB over the Arrow file) skips
// it by construction and an older znippy opens the archive unchanged. Pushed
// through the same `ArchiveMetaSink` the data index went through, so there is
// no second write path and the iceberg backend gets it too.
//
shape.extend(znippy_common::fsmeta::summarise_file_modes(&file_modes));
if !shape.is_empty() {
sink.push_subindex(
znippy_common::fsmeta::fsmeta_schema().as_ref(),
&[znippy_common::fsmeta::build_fsmeta_batch(&shape)?],
GroupKey {
pkg_type: znippy_common::RESERVED_PKG_TYPE,
repo: String::new(),
module_name: znippy_common::index::ZNIPPY_FSMETA_MODULE.to_string(),
},
)?;
}
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 = ChunkRevolver::with_coalescing(plan.num_slots, plan.slot_size, plan.slice_size);
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(); }
}
// `cur` is empty (taken above), but `Option<Filler>` now owns a
// DESTRUCTOR — an unpublished `Filler` returns its slot to the free
// list, where the old `Clip` simply leaked it. That destructor keeps
// `cur`'s borrow of `pool` alive to the end of the scope, so the
// binding has to go before the pool can. This is the one place the
// upstream pool's strictly-safer drop behaviour is visible here.
drop(cur);
// Drain: reclaim every slot to prove the workers and the writer are
// done with slot memory before the backing buffers go away. The
// reclaimed fillers are HELD until all of them are in hand, then
// dropped together — and that holding is load-bearing, not tidiness.
//
// `Filler` has a destructor that returns its slot to the free list
// (the old `Clip` had none and leaked the slot instead). So
//
// for _ in 0..pool.num_slots() { let _ = pool.claim_slot(); }
//
// does NOT drain the pool: each claim drops at the end of its own
// statement and pushes the slot straight back, so the loop can pop
// the SAME free slot `num_slots` times while every other slot is
// still in flight. It then falls through to `drop(pool)`, freeing the
// memory blocks while workers still hold `PackedSlice` pointers into
// them — a use-after-free.
//
// MEASURED 2026-08-25, release build, one 80 MB incompressible file
// (4 slices of 25 MiB): wrote exactly 26 214 400 bytes — the first
// slice — and then died with SIGSEGV (exit 139). Holding the fillers
// makes each iteration consume a DISTINCT slot, so the loop cannot
// finish until every slot has genuinely come back.
let mut reclaimed = Vec::with_capacity(pool.num_slots());
for _ in 0..pool.num_slots() {
reclaimed.push(pool.claim_slot());
}
drop(reclaimed);
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 = ChunkRevolver::with_coalescing(plan.num_slots, plan.slot_size, plan.slice_size);
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_slot(); }
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_slot();
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, mut skip) = batch[i];
let got = read_results[i];
// Refine the enumeration-time verdict with the bytes now in
// the slot — the SAME predicate `drive_pass` applies, so the
// archive is byte-identical either way. What changes is the
// REPORT: `skip` here was `skip_by_path` alone, which cannot
// see an entry carrying no extension, so a tree of bare-hex
// gzip blobs (a container-registry mirror: every filename is
// a sha256 with no suffix) reported `uncompressed_files: 0`
// and a 0.00 % ratio while the index correctly recorded every
// one of them `compressed = false`. MEASURED on a synthetic
// `talosctl image cache-create` tree: 3 gzip blobs, report
// said 3 compressed / 0 stored, on-disk said the opposite.
// A win nobody can see is a win nobody defends.
//
// A small file is exactly one `Round`, so the head here IS the
// file head and needs no `carries_file_wide` narrowing (that
// exists only because the big pass probes every chunk).
if !skip && got > 0 {
let head = unsafe {
std::slice::from_raw_parts(
slot_ptr.add(offsets[i]),
got.min(SNIFF_PREFIX_LEN),
)
};
skip = policy.skip_by_bytes(head);
}
// 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(); }
}
// `cur` is empty (taken above), but `Option<Filler>` now owns a
// DESTRUCTOR — an unpublished `Filler` returns its slot to the free
// list, where the old `Clip` simply leaked it. That destructor keeps
// `cur`'s borrow of `pool` alive to the end of the scope, so the
// binding has to go before the pool can. This is the one place the
// upstream pool's strictly-safer drop behaviour is visible here.
drop(cur);
// Drain: reclaim every slot to prove the workers and the writer are
// done with slot memory before the backing buffers go away. The
// reclaimed fillers are HELD until all of them are in hand, then
// dropped together — and that holding is load-bearing, not tidiness.
//
// `Filler` has a destructor that returns its slot to the free list
// (the old `Clip` had none and leaked the slot instead). So
//
// for _ in 0..pool.num_slots() { let _ = pool.claim_slot(); }
//
// does NOT drain the pool: each claim drops at the end of its own
// statement and pushes the slot straight back, so the loop can pop
// the SAME free slot `num_slots` times while every other slot is
// still in flight. It then falls through to `drop(pool)`, freeing the
// memory blocks while workers still hold `PackedSlice` pointers into
// them — a use-after-free.
//
// MEASURED 2026-08-25, release build, one 80 MB incompressible file
// (4 slices of 25 MiB): wrote exactly 26 214 400 bytes — the first
// slice — and then died with SIGSEGV (exit 139). Holding the fillers
// makes each iteration consume a DISTINCT slot, so the loop cannot
// finish until every slot has genuinely come back.
let mut reclaimed = Vec::with_capacity(pool.num_slots());
for _ in 0..pool.num_slots() {
reclaimed.push(pool.claim_slot());
}
drop(reclaimed);
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 ChunkRevolver,
tx_slice: &crossbeam_channel::Sender<PackedSlice>,
cur: &mut Option<Filler<'p>>,
need: usize,
) {
loop {
if cur.is_none() {
*cur = pool.claim_slot();
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.
//
// Both bounds here are stated against the planner's documented contract
// rather than against `STARVED` directly, because `STARVED` is not
// reachable on every host. `PoolPlan::plan` has a floor of ONE SLOT OF ONE
// SLICE — the pipeline must be able to hold the largest cut it can
// produce or it cannot make progress — and `slice_size` is
// `DEFAULT_SLOT_SIZE / max_core_in_flight`, which is a property of the
// HOST, not of the budget. It cannot follow the budget down: it is the
// big/small partition threshold and the big-file cut length, so moving it
// would move every chunk boundary and break the very byte-identity this
// test exists to assert.
//
// MEASURED 2026-08-25: on an 8-physical-core box `max_core_in_flight` is
// `ceil(8 × 0.90)` = 8, so `slice_size` is 200 MiB / 8 = 25 MiB and the
// floor is 26 214 400 B — over the 8 MiB budget. The old assertions read
// `bytes() <= STARVED` and `bytes() * 100 < default`, and BOTH were false
// here (25 MiB > 8 MiB; 2.6 GiB > 1.6 GiB) while the planner was behaving
// exactly as documented. They passed only on a box wide enough to make
// `slice_size` smaller than 8 MiB — 25 physical cores or more. A test that
// encodes the author's core count is not an invariant.
let workers = CONFIG.max_core_in_flight.max(1);
let starved_plan = PoolPlan::plan(u64::MAX / 4, workers, STARVED);
let floor = PoolPlan::slice_size_for(workers) as u64;
assert!(
starved_plan.bytes() <= STARVED.max(floor),
"the starved plan reserved {} bytes, over BOTH its {STARVED}-byte \
budget and the one-slice floor of {floor} bytes",
starved_plan.bytes()
);
// A budget this far under the floor must land exactly ON the floor: one
// slot, sized to one slice. Nothing smaller is representable.
assert_eq!(starved_plan.num_slots, 1, "a starved plan keeps one slot");
assert_eq!(
starved_plan.slot_size, starved_plan.slice_size,
"the one slot is exactly one slice wide"
);
// And it is meaningfully smaller than the old unconditional 1.6 GiB. The
// bound is 8× rather than the old 100×: dropping to one slot is worth
// exactly `DEFAULT_NUM_SLOTS` on its own, and the further `workers` factor
// from `slot_size = slice_size` is host-dependent, so 8× is the part that
// is true everywhere. Stated as an inequality, not an identity, because
// `DEFAULT_SLOT_SIZE / workers` truncates whenever `workers` does not
// divide 200 MiB (29 workers on a 32-core box loses 18 bytes).
assert!(
starved_plan.bytes() * 8 <= PoolPlan::default_for(workers).bytes(),
"the starved plan ({} B) is not at least 8× 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
}
/// **The pipeline terminates when a pass publishes NOTHING.**
///
/// Required by the `Magazine` → `ChunkRevolver` collapse and by nothing
/// before it. The fork's `claim` read a crossbeam receiver, so it returned
/// `None` once every sender was dropped and a drain loop could break out on
/// that sentinel. `ChunkRevolver::claim_slot` has no sentinel — its free list
/// is a `Mutex<Vec<u32>>` + `Condvar` and it returns `Some` for the pool's
/// whole life, blocking while every slot is in flight. If any drain or reader
/// loop had been relying on that `None` to stop, the pass would HANG rather
/// than fail, and a hang is invisible to an `is_err()`/`is_ok()` assertion —
/// the test harness would simply never finish.
///
/// So this is asserted with a watchdog rather than by falling through. The
/// two existing zero-work tests (`empty_input_dir_refuses_green`,
/// `only_dirs_no_files_refuses_green`) do NOT cover it: with no files at all
/// both `big_indices` and `small_indices` are empty, the passes are skipped
/// at `compress_dir`, and the pool is never constructed. The case that
/// reaches a live pool and still publishes nothing is a pass whose every file
/// FAILS TO OPEN — the reader claims no slot, commits no slice, sends no
/// slice, then drops `tx_slice` and runs the drain loop over a pool whose
/// slots are all still free.
///
/// Both small-pass paths are exercised: io_uring and the fallback.
#[test]
fn a_pass_that_publishes_no_slice_still_terminates() {
use std::os::unix::fs::PermissionsExt;
use std::sync::mpsc;
for &force_fallback in &[false, true] {
let tag = if force_fallback { "noslice_fb" } else { "noslice_io" };
let input = scratch(tag);
// EVERY file unreadable: the pass is entered (the walk found files,
// so `small_indices` is non-empty and the pool IS built) and then not
// one byte reaches a slot.
for i in 0..8 {
let p = input.join(format!("locked_{i}.bin"));
fs::write(&p, vec![0x5Au8; 512]).unwrap();
fs::set_permissions(&p, fs::Permissions::from_mode(0o000)).unwrap();
}
let out = scratch(&format!("{tag}_out"));
let arc = out.join("a.znippy");
// The watchdog. A hang in the drain loop would otherwise stall the
// whole test binary with no failure message.
let (tx, rx) = mpsc::channel();
let t = std::thread::spawn(move || {
FORCE_SMALL_FALLBACK.store(force_fallback, Ordering::SeqCst);
let r = compress_dir(&input, &arc, false, None, None, None);
FORCE_SMALL_FALLBACK.store(false, Ordering::SeqCst);
let _ = tx.send(r.is_ok());
let _ = fs::remove_dir_all(&input);
});
match rx.recv_timeout(std::time::Duration::from_secs(60)) {
Ok(_) => {}
Err(_) => panic!(
"compress_dir did not return within 60 s on a pass that publishes no \
slice (fallback={force_fallback}) — the drain loop is waiting on a slot \
that will never come back, which is what losing `claim`'s `None` \
sentinel costs if a loop depended on it"
),
}
t.join().expect("compress thread panicked");
let _ = fs::remove_dir_all(&out);
}
}
}