ldeflate 0.1.3

//! **The two-queue planner** — work that does not earn splitting goes in one
//! queue, work that does goes in another.
//!
//! This is the scheduling core, and it is the part that is actually new. The
//! shape is borrowed from `ljar`, but the borrow needs stating precisely,
//! because ljar does *not* implement the thing this crate needs (see
//! `.nornir/gatling-tricks.md` for the full note):
//!
//! - ljar's **10 MB threshold** (`ljar/src/lib.rs:48`) picks between two *entry
//!   points* for a whole archive — an in-memory pass or a streaming reader. It
//!   is not a per-item size class.
//! - ljar's **batch of 64** (`ljar/src/reader.rs:20`) is a **RAM bound** — "only
//!   one batch of decompressed output is live at a time" — not dispatch
//!   amortisation.
//! - ljar dispatches **one gatling unit per entry, with claim size 1**
//!   (`ljar/src/parallel.rs:32`, `ljar/src/batch.rs:109`). It never coalesces.
//!
//! So ljar has the *silhouette* of the two-queue design and none of the
//! coalescing. For git objects that gap is the whole problem: dispatching a task
//! to deflate 200 bytes costs more than the deflate. What this module adds is
//! the small queue.
//!
//! ## The three unit kinds
//!
//! | kind | when | why |
//! |---|---|---|
//! | [`Unit::Batch`] | object < `coalesce_below` | one task drains many objects **reusing one `Encoder`**. Measured: this is the ENTIRE win. Reusing a compressor (`reset`, ~1.6 us) instead of building one (`Compress::new`, ~4.1 us) saves ~2.5 us per object; the gatling claim it also amortises is only ~62 ns and never mattered. |
//! | [`Unit::Whole`] | `coalesce_below` ≤ object < `split_above` | one task, one object — the classic axis |
//! | [`Unit::Segment`] | object ≥ `split_above` | one object across many tasks, so a single 400 MB blob cannot serialise a worker |
//!
//! ## Determinism
//!
//! The plan is a pure function of `(object sizes, Config)`. It does not read the
//! worker count, the core count, or anything about completion order. That is
//! what lets the compressor promise byte-identical output at 1 thread and at 32.

use crate::container::Container;
use crate::precompressed::SkipPolicy;
use crate::split::{self, Segment};
use crate::unit::{DeflateError, Encoder, EncoderPool};

/// Tuning for [`compress_objects`]. Every default here is **measured** on this
/// machine (AMD Threadripper PRO 3975WX, 32 cores, 2026-08-03) — see
/// `.nornir/ldeflate-design.md` for the sweeps that produced them, and for the
/// two hypotheses those sweeps FALSIFIED.
///
/// | field | default | what the measurement said |
/// |---|---|---|
/// | `coalesce_below` | 64 KiB | monotone win up to here, no knee: 1.15x fewer CPU-seconds on the sub-64 KiB majority, 1.02x over a whole real repository. Covers 93.8% of real git objects. |
/// | `batch_bytes` | 512 KiB | keeps ~15 units per worker on a 46 k-object corpus — big enough to amortise, small enough that LPT can still balance. |
/// | `claim` | 4 | measured **irrelevant** here (0.3% across 1/4/16). Kept at 4 because it costs nothing; the house note that claim 1 is a 2x regression is about far cheaper units than a deflate. |
/// | `split_above` / `segment` | 4 MiB / 1 MiB | 10.3x wall throughput on large compressible objects for a 2.2% ratio cost. `segment` must stay <= `split_above` or splitting never fires. |
#[derive(Debug, Clone, Copy)]
pub struct Config {
    /// DEFLATE level, 0-9.
    pub level: u32,
    /// Objects **strictly smaller** than this are coalesced into batches.
    pub coalesce_below: usize,
    /// A coalesced batch closes once it holds this many source bytes...
    pub batch_bytes: usize,
    /// ...or this many objects, whichever comes first.
    pub batch_count: usize,
    /// Objects **at or above** this are split within-stream.
    pub split_above: usize,
    /// Uncompressed bytes per segment on the split path.
    pub segment: usize,
    /// Worker threads. `0` ⇒ one per core.
    pub workers: usize,
    /// gatling claim size — indices taken per atomic `fetch_add`. `1` is a
    /// measured regression on cheap units; the planner already coalesces, so
    /// this is a second-order knob.
    pub claim: usize,
    /// LZ77 window, `2^window_bits` bytes (9..=15). See
    /// [`Encoder::new_with_window`] — for git-sized objects the default 15 is
    /// mostly a per-object state-reset tax, not compression.
    pub window_bits: u8,
    /// Which framing wraps each output stream: zlib (default), gzip, or bare
    /// DEFLATE. The scheduler is identical for all three — see
    /// [`crate::container`].
    pub container: Container,
    /// Whether to skip the entropy coder for content that is already
    /// compressed. **On by default** — see [`crate::precompressed`].
    ///
    /// Applies to all three containers at once because it is applied where the
    /// bytes are compressed ([`run_unit`]), not in any one container's writer.
    pub skip: SkipPolicy,
}

impl Default for Config {
    fn default() -> Self {
        Config {
            level: 6,
            coalesce_below: 64 * 1024,
            batch_bytes: 512 * 1024,
            batch_count: 256,
            split_above: 4 << 20,
            segment: 1 << 20,
            workers: 0,
            claim: 4,
            window_bits: crate::unit::WINDOW_BITS_DEFAULT,
            container: Container::Zlib,
            skip: SkipPolicy::resolve(),
        }
    }
}

impl Config {
    /// Level 6 — git's own default, and the level `git repack` uses unless told
    /// otherwise.
    pub fn level(mut self, level: u32) -> Self {
        self.level = level;
        self
    }
    /// Force a worker count (`0` ⇒ one per core). Used by the scaling bench.
    pub fn workers(mut self, w: usize) -> Self {
        self.workers = w;
        self
    }
    /// Disable coalescing — the "one task per object" arm the bench compares
    /// against. Sets the small threshold to zero so no object is ever batched.
    pub fn without_coalescing(mut self) -> Self {
        self.coalesce_below = 0;
        self
    }
    /// Disable within-stream splitting — the "objects only" arm.
    pub fn without_splitting(mut self) -> Self {
        self.split_above = usize::MAX;
        self
    }
    /// Emit each stream in `container` instead of zlib.
    pub fn container(mut self, container: Container) -> Self {
        self.container = container;
        self
    }
    /// Set the already-compressed skip policy (default: probe the bytes).
    pub fn skip(mut self, skip: SkipPolicy) -> Self {
        self.skip = skip;
        self
    }
    /// Run the entropy coder over everything, probing nothing — the pre-probe
    /// behaviour, by name. The arm the bench compares against.
    pub fn compress_everything(mut self) -> Self {
        self.skip = SkipPolicy::compress_everything();
        self
    }
}

/// One dispatchable piece of work. Holds only indices and ranges — never bytes.
/// The objects stay where the caller put them and every worker reads them
/// through a shared `&[&[u8]]`; nothing is cloned into the pool.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Unit {
    /// Objects `[lo, hi)`, each compressed to its own stream, all by one worker.
    Batch { lo: usize, hi: usize },
    /// One whole object.
    Whole { idx: usize },
    /// Segment `seg` of object `idx`, covering source bytes `[start, end)`.
    Segment {
        idx: usize,
        seg: usize,
        start: usize,
        end: usize,
        is_last: bool,
    },
}

/// Build the work plan for `sizes` under `cfg`.
///
/// Pure: same inputs ⇒ same plan, always, on any machine with any core count.
///
/// # Errors
///
/// [`DeflateError::TooManyPieces`] if any one object would be cut into more
/// than [`crate::limits::MAX_PIECES`] segments at `cfg.segment`. The object
/// *count* is deliberately not capped — those buffers are the caller's and
/// already resident; only counts this crate amplifies out of a size parameter
/// are checked. See [`crate::limits`].
pub fn plan(sizes: &[usize], cfg: &Config) -> Result<Vec<Unit>, DeflateError> {
    let mut units = Vec::new();
    let mut batch_lo: Option<usize> = None;
    let mut batch_bytes = 0usize;
    let mut batch_count = 0usize;

    // Close the currently-open coalesced batch, if any.
    let flush = |units: &mut Vec<Unit>,
                 batch_lo: &mut Option<usize>,
                 batch_bytes: &mut usize,
                 batch_count: &mut usize,
                 hi: usize| {
        if let Some(lo) = batch_lo.take() {
            if hi > lo {
                units.push(Unit::Batch { lo, hi });
            }
        }
        *batch_bytes = 0;
        *batch_count = 0;
    };

    for (idx, &size) in sizes.iter().enumerate() {
        if size < cfg.coalesce_below {
            // ── small queue: accumulate ─────────────────────────────────────
            if batch_lo.is_none() {
                batch_lo = Some(idx);
            }
            batch_bytes += size;
            batch_count += 1;
            if batch_bytes >= cfg.batch_bytes || batch_count >= cfg.batch_count {
                flush(
                    &mut units,
                    &mut batch_lo,
                    &mut batch_bytes,
                    &mut batch_count,
                    idx + 1,
                );
            }
            continue;
        }
        // A non-small object ends any open batch (batches stay contiguous, so
        // reassembly is a straight walk with no index bookkeeping).
        flush(
            &mut units,
            &mut batch_lo,
            &mut batch_bytes,
            &mut batch_count,
            idx,
        );

        if size >= cfg.split_above {
            // ── large queue: split within the stream ────────────────────────
            for (seg, (start, end, is_last)) in split::plan_segments(size, cfg.segment)?
                .into_iter()
                .enumerate()
            {
                units.push(Unit::Segment {
                    idx,
                    seg,
                    start,
                    end,
                    is_last,
                });
            }
        } else {
            units.push(Unit::Whole { idx });
        }
    }
    flush(
        &mut units,
        &mut batch_lo,
        &mut batch_bytes,
        &mut batch_count,
        sizes.len(),
    );
    Ok(units)
}

/// LPT weight for one unit — the source bytes it covers. `gatling_map_balanced`
/// schedules heaviest-first, so the 400 MB blob's segments are claimed before
/// the 200-byte trees and the cheap tail fills the gaps.
#[inline]
pub fn weight(u: &Unit, sizes: &[usize]) -> u64 {
    match *u {
        Unit::Batch { lo, hi } => sizes[lo..hi].iter().map(|&s| s as u64).sum(),
        Unit::Whole { idx } => sizes[idx] as u64,
        Unit::Segment { start, end, .. } => (end - start) as u64,
    }
}

/// What one unit produced. The `u32` beside each finished stream is the
/// whole-object checksum under the container's algorithm, computed **on the
/// worker** — and only when the caller asked for it (`want_sum`), so the git
/// path never pays for a checksum it does not use.
enum UnitOut {
    Batch(Vec<(Vec<u8>, u32)>),
    Whole((Vec<u8>, u32)),
    Segment(Segment),
}

/// Run one unit on a worker thread. `pool_wrapped` / `pool_raw` hand out that
/// worker's reusable encoders (zlib-wrapped for whole objects, bare for
/// segments).
///
/// **Both checkouts are lazy and each happens at most once per unit.** A
/// coalesced batch of ~95 git objects therefore pays one pool round-trip, and a
/// unit that never reaches the bare encoder never constructs one. Before the
/// pool this function's caller built BOTH encoders unconditionally, so the git
/// path — for which the bare one is dead code — paid `Compress::new` (4082 ns)
/// per unit for an encoder that was never touched.
fn run_unit(
    u: &Unit,
    objects: &[&[u8]],
    pool_wrapped: &EncoderPool,
    pool_raw: &EncoderPool,
    level: u32,
    container: Container,
    want_sum: bool,
    skip: SkipPolicy,
) -> Result<UnitOut, DeflateError> {
    // For zlib the encoder writes the wrapper itself (fast path, and it computes
    // Adler-32 internally). For gzip and raw the body is compressed bare and
    // framed here, so there is exactly one framing code path.
    // The checksum is an extra pass over the input, so it is computed only when
    // a caller needs it out of band (ZIP headers). For zlib and gzip the trailer
    // already carries it and `want_sum` is false, so the git path is unchanged.
    //
    // There is only ONE encoder in play here, not two: `pool_wrapped` is built
    // with `zlib_wrapped = container.encoder_is_zlib_wrapped()`, so on the
    // gzip/raw containers it *already is* the bare encoder the `else` arm wanted.
    // The old two-encoder signature made that look like a choice; it never was,
    // and the second encoder was pure construction cost. `pool_raw` is reached
    // only by [`Unit::Segment`].
    let one = |src: &[u8], ew: &mut Encoder| -> Result<(Vec<u8>, u32), DeflateError> {
        let sum = if want_sum { container.checksum(src) } else { 0 };
        // ── the skip, for every container at once ───────────────────────────
        // Deciding here rather than in gzip.rs/zip.rs is the whole point: this
        // closure is the only place in the crate where an object's bytes meet
        // the entropy coder, so one branch covers zlib, gzip, ZIP and JAR and
        // there is no per-format copy to keep in agreement. `stored_deflate` is
        // a length-prefixed memcpy, and `Container::wrap` already knows how to
        // frame a bare body for each container.
        if skip.skip(src) {
            let body = crate::precompressed::stored_deflate(src, true);
            return Ok((container.wrap(level, &body, src), sum));
        }
        let mut o = Vec::new();
        ew.compress_into(src, &mut o)?;
        if container.encoder_is_zlib_wrapped() {
            Ok((o, sum))
        } else {
            Ok((container.wrap(level, &o, src), sum))
        }
    };
    match *u {
        Unit::Batch { lo, hi } => {
            // ONE checkout for the whole batch — ~95 objects on the measured
            // corpus — so the pool lock is touched twice per ~95 deflates.
            let mut ew = pool_wrapped.checkout()?;
            let mut outs = Vec::with_capacity(hi - lo);
            for obj in &objects[lo..hi] {
                outs.push(one(obj, &mut ew)?);
            }
            Ok(UnitOut::Batch(outs))
        }
        Unit::Whole { idx } => {
            let mut ew = pool_wrapped.checkout()?;
            Ok(UnitOut::Whole(one(objects[idx], &mut ew)?))
        }
        Unit::Segment {
            idx,
            start,
            end,
            is_last,
            ..
        } => {
            // The probe is on the WHOLE object, not on this segment's slice, so
            // every segment of one object agrees — a segment from the middle of
            // a PNG has no magic bytes of its own, and a plan that stored some
            // segments and deflated others would still be correct but would
            // depend on where the cuts landed.
            let mut er = pool_raw.checkout()?;
            Ok(UnitOut::Segment(split::compress_segment_skipping(
                &mut er,
                objects[idx],
                start,
                end,
                is_last,
                container,
                skip.skip(objects[idx]),
            )?))
        }
    }
}

/// Compress every object in `objects` into its own complete zlib (RFC 1950)
/// stream, in input order.
///
/// This is the entry point Gunnar's repack/pack-serve path wants: thousands of
/// independent small streams, plus a thin tail of large ones, all fanned out
/// across one no-barrier gatling pass.
///
/// **Zero copy:** `objects` is borrowed for the whole call. Workers read the
/// caller's bytes in place; the only allocation on the hot path is each
/// object's output buffer, which has to exist anyway.
///
/// **Zero wait:** every unit is claimed off one shared atomic cursor. A worker
/// that finishes a 200-byte tree grabs the next unit immediately; it never waits
/// on a barrier, a scheduler, or a sibling that drew a 400 MB blob.
pub fn compress_objects(objects: &[&[u8]], cfg: &Config) -> Result<Vec<Vec<u8>>, DeflateError> {
    Ok(compress_objects_inner(objects, cfg, false)?
        .into_iter()
        .map(|(b, _)| b)
        .collect())
}

/// [`compress_objects`], also returning each object's whole-object checksum
/// under `cfg.container`.
///
/// For a container that stores its checksum **out of band** — ZIP keeps each
/// entry's CRC-32 in the local and central headers, not in a trailer — this
/// avoids a second serial pass over every entry. The checksum is computed on the
/// worker that compressed the object, or folded from the per-segment checksums
/// when the object took the split path.
pub fn compress_objects_checksummed(
    objects: &[&[u8]],
    cfg: &Config,
) -> Result<Vec<(Vec<u8>, u32)>, DeflateError> {
    compress_objects_inner(objects, cfg, true)
}

fn compress_objects_inner(
    objects: &[&[u8]],
    cfg: &Config,
    want_sum: bool,
) -> Result<Vec<(Vec<u8>, u32)>, DeflateError> {
    let sizes: Vec<usize> = objects.iter().map(|o| o.len()).collect();
    let units = plan(&sizes, cfg)?;
    if units.is_empty() {
        return Ok(Vec::new());
    }

    let level = cfg.level;
    let wb = cfg.window_bits;
    let ct = cfg.container;
    // Two pools, both LAZY: nothing is constructed until a unit asks for it, so
    // the git path — which never reaches `pool_r` — constructs from it ZERO
    // times. Owned by this call, not a `thread_local!`, so the crate stays
    // reentrant.
    let pool_z = EncoderPool::new(level, ct.encoder_is_zlib_wrapped(), wb);
    let pool_r = EncoderPool::new(level, false, wb);
    // The ONE fan-out. LPT (heaviest first) + batched atomic claims; results
    // come back in unit order, so reassembly below is a single forward walk.
    let results: Vec<Result<UnitOut, DeflateError>> =
        gatling::gatling_forkjoin::gatling_map_balanced(
            &units,
            cfg.workers,
            cfg.claim,
            |u| weight(u, &sizes),
            |_i, u| {
                // Encoders come from the pool, so `Compress::new` is paid once
                // per WORKER for the whole call instead of twice per unit.
                run_unit(u, objects, &pool_z, &pool_r, level, ct, want_sum, cfg.skip)
            },
        );

    // ── Reassemble, in order, on one thread ─────────────────────────────────
    let mut out: Vec<(Vec<u8>, u32)> = Vec::with_capacity(objects.len());
    let mut pending: Vec<Segment> = Vec::new();
    let mut pending_idx: Option<usize> = None;

    for (u, r) in units.iter().zip(results) {
        match (u, r?) {
            (Unit::Batch { .. }, UnitOut::Batch(v)) => out.extend(v),
            (Unit::Whole { .. }, UnitOut::Whole(v)) => out.push(v),
            (Unit::Segment { idx, is_last, .. }, UnitOut::Segment(s)) => {
                debug_assert!(pending_idx.is_none_or(|p| p == *idx));
                pending_idx = Some(*idx);
                pending.push(s);
                if *is_last {
                    out.push(split::assemble_checksummed(level, &pending, ct));
                    pending.clear();
                    pending_idx = None;
                }
            }
            _ => unreachable!("unit kind and output kind cannot disagree"),
        }
    }
    debug_assert!(pending.is_empty(), "a split object was left unassembled");
    debug_assert_eq!(
        out.len(),
        objects.len(),
        "one output stream per input object"
    );

    gatling::functional_status(
        "ldeflate",
        "compress_objects",
        true,
        &"objects fanned out over one no-barrier gatling pass",
    );
    Ok(out)
}

/// The serial reference. Same output, one thread, no gatling — the baseline
/// every scaling number in the bench is divided by, and the oracle the parallel
/// path is asserted equal to.
pub fn compress_objects_serial(
    objects: &[&[u8]],
    cfg: &Config,
) -> Result<Vec<Vec<u8>>, DeflateError> {
    let sizes: Vec<usize> = objects.iter().map(|o| o.len()).collect();
    let units = plan(&sizes, cfg)?;
    // The same two pools the parallel path uses, so the serial oracle and the
    // scheduler run byte-identical code (LAW 5). With one thread a pool is a
    // one-element free list: `Compress::new` once, `reset` thereafter.
    let pool_z = EncoderPool::new(
        cfg.level,
        cfg.container.encoder_is_zlib_wrapped(),
        cfg.window_bits,
    );
    let pool_r = EncoderPool::new(cfg.level, false, cfg.window_bits);

    let mut out: Vec<Vec<u8>> = Vec::with_capacity(objects.len());
    let mut pending: Vec<Segment> = Vec::new();
    for u in &units {
        match run_unit(
            u,
            objects,
            &pool_z,
            &pool_r,
            cfg.level,
            cfg.container,
            false,
            cfg.skip,
        )? {
            UnitOut::Batch(v) => out.extend(v.into_iter().map(|(b, _)| b)),
            UnitOut::Whole((b, _)) => out.push(b),
            UnitOut::Segment(s) => {
                let last = matches!(u, Unit::Segment { is_last: true, .. });
                pending.push(s);
                if last {
                    out.push(split::assemble(cfg.level, &pending, cfg.container));
                    pending.clear();
                }
            }
        }
    }
    Ok(out)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn objs(sizes: &[usize]) -> Vec<Vec<u8>> {
        sizes
            .iter()
            .enumerate()
            .map(|(i, &n)| {
                // Distinct, compressible-but-not-trivial content per object, so a
                // mis-ordered reassembly cannot pass by luck.
                let tag = format!("object-{i}-");
                let mut v = Vec::with_capacity(n);
                while v.len() < n {
                    v.extend_from_slice(tag.as_bytes());
                    v.extend_from_slice(b"payload payload payload ");
                }
                v.truncate(n);
                v
            })
            .collect()
    }

    fn refs(v: &[Vec<u8>]) -> Vec<&[u8]> {
        v.iter().map(|x| x.as_slice()).collect()
    }

    /// The plan must cover every object exactly once, in order — whatever the
    /// mix of sizes and thresholds.
    #[test]
    fn plan_covers_every_object_exactly_once() {
        let cfg = Config::default();
        let sizes: Vec<usize> = (0..500)
            .map(|i| match i % 7 {
                0 => 120,
                1 => 900,
                2 => 4_000,
                3 => 70_000,
                4 => 300_000,
                5 => 3_000_000,
                _ => 40,
            })
            .collect();
        let units = plan(&sizes, &cfg).unwrap();

        let mut seen = vec![0usize; sizes.len()];
        let mut order = Vec::new();
        for u in &units {
            match *u {
                Unit::Batch { lo, hi } => {
                    for i in lo..hi {
                        seen[i] += 1;
                        order.push(i);
                    }
                }
                Unit::Whole { idx } => {
                    seen[idx] += 1;
                    order.push(idx);
                }
                Unit::Segment { idx, is_last, .. } => {
                    if is_last {
                        seen[idx] += 1;
                        order.push(idx);
                    }
                }
            }
        }
        assert!(
            seen.iter().all(|&c| c == 1),
            "every object exactly once: {seen:?}"
        );
        assert!(
            order.windows(2).all(|w| w[0] < w[1]),
            "objects must stay in order"
        );
    }

    /// **The scaling-bench guard, seen red.** The parallel path and the serial
    /// path must produce byte-identical output at every worker count. If they
    /// did not, the bench would be timing two different computations and its
    /// speedup number would be meaningless.
    #[test]
    fn parallel_equals_serial_at_every_worker_count() {
        let data = objs(&[80, 300, 900, 64_000, 200_000, 12 << 20, 40, 5, 1_500_000]);
        let r = refs(&data);
        let base = compress_objects_serial(&r, &Config::default()).unwrap();
        for w in [1usize, 2, 3, 8, 32] {
            let got = compress_objects(&r, &Config::default().workers(w)).unwrap();
            assert_eq!(
                got, base,
                "workers={w}: output differs from the serial oracle"
            );
        }
    }

    /// Determinism across worker counts is only meaningful if the output would
    /// actually *change* under a thread-count-dependent plan. Prove the plan is
    /// the thing carrying determinism: a plan built for a different segment size
    /// must produce different bytes, so "identical" above is not vacuous.
    #[test]
    fn segment_size_actually_changes_the_bytes() {
        // Must exceed `split_above` (4 MiB) or nothing is split and the two
        // configs trivially agree — the failure mode this test exists to catch.
        let data = objs(&[24 << 20]);
        let r = refs(&data);
        assert!(
            data[0].len() >= Config::default().split_above,
            "fixture must reach the split queue"
        );
        let a = compress_objects(
            &r,
            &Config {
                segment: 256 << 10,
                ..Config::default()
            },
        )
        .unwrap();
        let b = compress_objects(
            &r,
            &Config {
                segment: 1 << 20,
                ..Config::default()
            },
        )
        .unwrap();
        assert_ne!(
            a, b,
            "two different segment sizes produced identical bytes — the split path \
             is not actually segmenting, so the determinism test above proves nothing"
        );
    }

    /// Every stream this crate emits, whichever queue produced it, must inflate
    /// back to its object.
    #[test]
    fn every_queue_round_trips() {
        use std::io::Read;
        // Sizes chosen to hit all three queues under the default config.
        let data = objs(&[10, 200, 65_000, 300_000, 9 << 20, 7, 1_100_000]);
        let r = refs(&data);
        let out = compress_objects(&r, &Config::default()).unwrap();
        assert_eq!(out.len(), data.len());
        for (i, (z, src)) in out.iter().zip(&data).enumerate() {
            let mut got = Vec::new();
            flate2::read::ZlibDecoder::new(&z[..])
                .read_to_end(&mut got)
                .unwrap_or_else(|e| panic!("object {i} ({} B) failed to inflate: {e}", src.len()));
            assert_eq!(&got, src, "object {i} ({} B) round-trip differs", src.len());
        }
    }

    /// The coalescing threshold must actually change the plan shape — otherwise
    /// `without_coalescing()` (the bench's A/B arm) is a no-op and the batching
    /// measurement is fake.
    #[test]
    fn coalescing_actually_changes_the_unit_count() {
        let sizes = vec![200usize; 10_000];
        let batched = plan(&sizes, &Config::default()).unwrap();
        let unbatched = plan(&sizes, &Config::default().without_coalescing()).unwrap();
        assert_eq!(
            unbatched.len(),
            10_000,
            "no-coalesce must be one unit per object"
        );
        assert!(
            batched.len() * 8 < unbatched.len(),
            "coalescing produced {} units vs {} — barely fewer, so the small queue \
             is not doing anything and the A/B arm is measuring noise",
            batched.len(),
            unbatched.len()
        );
        // And the two must still agree byte-for-byte.
        let data = objs(&vec![200usize; 400]);
        let r = refs(&data);
        assert_eq!(
            compress_objects(&r, &Config::default()).unwrap(),
            compress_objects(&r, &Config::default().without_coalescing()).unwrap(),
            "coalescing must not change a single output byte"
        );
    }

    /// Empty input, empty objects, one object — the shapes that break planners.
    #[test]
    fn degenerate_inputs() {
        let cfg = Config::default();
        assert!(compress_objects(&[], &cfg).unwrap().is_empty());

        let empties: Vec<Vec<u8>> = vec![Vec::new(); 5];
        let r = refs(&empties);
        let out = compress_objects(&r, &cfg).unwrap();
        assert_eq!(out.len(), 5);
        use std::io::Read;
        for z in &out {
            let mut got = Vec::new();
            flate2::read::ZlibDecoder::new(&z[..])
                .read_to_end(&mut got)
                .unwrap();
            assert!(got.is_empty());
        }
    }
}