ldeflate 0.1.3

//! The **serial encode kernel** — one unit of work, one thread, no parallelism.
//!
//! Everything parallel in this crate ([`crate::plan`], [`crate::many`],
//! [`crate::stream`]) is a *scheduler* over this module. Keeping the kernel
//! single-threaded and allocation-reusing is what makes the schedulers
//! measurable: a scaling curve that flatters a serial baseline is measuring the
//! scheduler, not the codec (LAW 2).
//!
//! ## Why `flate2` + `zlib-rs` and not a hand-rolled Huffman coder
//!
//! `linflate` deliberately hand-rolls the *read* side because the win there is
//! real and measurable (packed decode tables, branchless refill, SIMD match
//! copy). The write side is a different bet: the entropy coder is not where the
//! parallel win lives, and a fresh DEFLATE encoder would be both slower and
//! worse-ratio than zlib-rs for months. LAW 5 — reuse, do not twin.
//!
//! `zlib-rs` is a **pure Rust** rewrite of zlib: no `build.rs`, no `.c`, no
//! cmake, no bindgen, no `libz-sys`. Verified by inspection of the vendored
//! crate. It measured **1.7-2.1x faster than miniz_oxide at level 6 with a
//! byte-identical compression ratio** on this machine (gunnar
//! `.nornir/performance-design.md`, Measurement 2, 2026-08-03).

use flate2::{Compress, Compression, FlushCompress, Status};

/// The RFC-1950 default: a 32 KiB LZ77 window. What `git` itself uses.
pub const WINDOW_BITS_DEFAULT: u8 = 15;
/// The smallest window zlib accepts (512 bytes).
pub const WINDOW_BITS_MIN: u8 = 9;

/// Errors the encode kernel can raise. Deliberately tiny — a DEFLATE *encoder*
/// has far fewer failure modes than a decoder (it never sees hostile input).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DeflateError {
    /// The underlying compressor reported an error or refused to make progress.
    /// Carries a short static reason.
    Encode(&'static str),
    /// A configuration value is outside its legal range.
    Config(&'static str),
    /// A ZIP archive limit was exceeded. This writer emits no ZIP64, so an
    /// archive or entry over 4 GiB, or more than 65535 entries, is refused
    /// rather than written into a silently-truncated 32/16-bit field — which
    /// produces an archive no tool can diagnose afterwards.
    ZipLimit(&'static str),
    /// A size parameter would cut the input into more independently framed
    /// pieces than [`crate::limits::MAX_PIECES`] — refused **before** anything
    /// is allocated. See [`crate::limits`] for the ceiling and why the crate
    /// errors here instead of silently using a larger piece size.
    TooManyPieces {
        /// Which axis was asked for: `"gzip members"`, `"split segments"`.
        what: &'static str,
        /// Bytes of input the request covered.
        total: usize,
        /// Piece size the caller asked for, verbatim.
        piece: usize,
        /// Pieces that would have been produced.
        pieces: usize,
        /// The ceiling that refused it.
        max: usize,
    },
}

impl std::fmt::Display for DeflateError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            DeflateError::Encode(m) => write!(f, "deflate encode failed: {m}"),
            DeflateError::Config(m) => write!(f, "bad ldeflate config: {m}"),
            DeflateError::ZipLimit(m) => write!(f, "ldeflate zip limit: {m}"),
            DeflateError::TooManyPieces {
                what,
                total,
                piece,
                pieces,
                max,
            } => write!(
                f,
                "ldeflate: {total} bytes at {piece} bytes per piece is {pieces} pieces \
                 of {what}, over the ceiling of {max}; pass at least {} bytes per piece",
                crate::limits::min_piece_for(*total),
            ),
        }
    }
}

impl std::error::Error for DeflateError {}

/// A reusable, single-threaded encoder. **One per worker**, reset between units.
///
/// The point of this type is that a worker which claims 500 tiny objects builds
/// the zlib state machine **once**. Measured on oden 2026-08-03, that is worth
/// ~2.5 microseconds per object: `Compress::new` costs **4082 ns** while
/// `reset()` costs **1567 ns**, and the deflate of a small object is only ~95 ns
/// of actual coding. This — not the ~62 ns gatling claim — is the whole reason
/// the coalescing queue in [`crate::plan`] wins.
///
/// It also refines gunnar `.nornir/performance-design.md` Measurement 4, which
/// found "no difference beyond noise" from reusing the encoder. That was
/// measured on a corpus of mean 19.2 KB. Real git objects have a **median of
/// 1014 B**, where 2.5 us is a quarter of the per-object cost, not 1% of it.
pub struct Encoder {
    z: Compress,
    level: u32,
    zlib_wrapped: bool,
}

impl Encoder {
    /// Build an encoder at `level` (0-9). `zlib_wrapped` selects a full zlib
    /// stream (2-byte header + Adler-32 trailer, RFC 1950 — what a git loose
    /// object and a packfile entry both use) versus a bare DEFLATE stream
    /// (RFC 1951 — what the segment path of [`crate::split`] concatenates).
    pub fn new(level: u32, zlib_wrapped: bool) -> Result<Self, DeflateError> {
        Self::new_with_window(level, zlib_wrapped, WINDOW_BITS_DEFAULT)
    }

    /// Build an encoder with an explicit LZ77 **window size** of `2^window_bits`
    /// bytes (9..=15).
    ///
    /// # A FALSIFIED hypothesis, kept because the negative result is the point
    ///
    /// There is a hard per-object floor at level 6 of ~4.4 microseconds that is
    /// independent of payload size — a 64 B object costs the same as a 256 B
    /// one. The hypothesis was that this is zlib's per-stream state reset
    /// clearing a hash table sized by the **window**, and that shrinking the
    /// window would collapse it. This API was added to test that.
    ///
    /// **It is wrong, measured (oden, 2026-08-03).** The floor is *identical* at
    /// every window size: 5014 ns at `window_bits = 9` and 5001 ns at 15 for a
    /// 64-byte object. Isolating further: `reset()` alone costs **1567 ns at
    /// every level**, and one 64 B `compress()` with no reset costs **95 ns**.
    /// So the fixed cost is real and it is the reset — but the table it clears
    /// is sized by zlib's `memLevel`, which is **independent of `windowBits`**
    /// and which `flate2` does not expose. This knob cannot reach it.
    ///
    /// Worse, a small window costs badly on real data: end-to-end over 45876
    /// real git objects, `window_bits = 9` produced **37% more bytes** than 15
    /// and was 0.87x the speed. The default is 15 and should stay there.
    ///
    /// The API is kept because it is honest, cheap, and correct — and because a
    /// caller who genuinely knows every object is tiny may still want it. What
    /// changed is the claim attached to it.
    ///
    /// Output stays a stream stock `git` reads: RFC 1950 puts the window size in
    /// the header's CINFO field, and `inflateInit` allocates 32 KiB regardless,
    /// so a *smaller* declared window is always acceptable to a decoder. (The
    /// reverse is not, which is why nothing here exceeds 15.)
    pub fn new_with_window(
        level: u32,
        zlib_wrapped: bool,
        window_bits: u8,
    ) -> Result<Self, DeflateError> {
        if level > 9 {
            return Err(DeflateError::Config("level must be 0..=9"));
        }
        if !(WINDOW_BITS_MIN..=WINDOW_BITS_DEFAULT).contains(&window_bits) {
            return Err(DeflateError::Config("window_bits must be 9..=15"));
        }
        Ok(Encoder {
            z: Compress::new_with_window_bits(Compression::new(level), zlib_wrapped, window_bits),
            level,
            zlib_wrapped,
        })
    }

    /// The level this encoder was built at.
    #[inline]
    pub fn level(&self) -> u32 {
        self.level
    }

    /// Compress `src` as one complete stream, **appending** to `out`.
    ///
    /// The encoder is reset first, so a single `Encoder` may be driven over
    /// thousands of independent objects with no reallocation. Returns the number
    /// of bytes appended.
    pub fn compress_into(&mut self, src: &[u8], out: &mut Vec<u8>) -> Result<usize, DeflateError> {
        self.z.reset();
        self.drive(src, out, FlushCompress::Finish)
    }

    /// Compress `src` and end the stream with a **full flush** rather than
    /// finishing it — the segment primitive of the within-stream split path.
    ///
    /// `Z_FULL_FLUSH` emits an empty stored block, which (a) byte-aligns the
    /// output and (b) resets the LZ77 window to empty. Both properties are load
    /// bearing: (a) lets the next segment's bytes be concatenated verbatim at a
    /// byte boundary, and (b) means the next segment can be compressed by a
    /// *different worker* that has never seen the preceding bytes and the result
    /// is still a valid stream. This is exactly the boundary `linflate`'s
    /// `deflate_scan::find_all_flushes` looks for on the read side — the two
    /// sides of the same seam.
    pub fn compress_segment_into(
        &mut self,
        src: &[u8],
        out: &mut Vec<u8>,
    ) -> Result<usize, DeflateError> {
        self.z.reset();
        self.drive(src, out, FlushCompress::Full)
    }

    /// Whether this encoder emits the RFC-1950 zlib wrapper.
    #[inline]
    pub fn is_zlib_wrapped(&self) -> bool {
        self.zlib_wrapped
    }

    /// Feed all of `src` then apply `end` until the compressor stops producing.
    fn drive(
        &mut self,
        src: &[u8],
        out: &mut Vec<u8>,
        end: FlushCompress,
    ) -> Result<usize, DeflateError> {
        let start = out.len();
        // Reserve a realistic first guess so the common (compressible) case never
        // reallocates: zlib's own bound is len + len/1000 + 13, but for real data
        // we expect to *shrink*, so ask for a fraction and let the loop grow.
        //
        // **A HALF, not a third.** The measured ratio on 76 616 real git objects
        // from the 19 `workspace_nordisk` repositories is **2.825** (2251.5 MiB
        // in, 796.9 MiB out at level 6), i.e. the output is 0.354 × the input on
        // average — *above* `len / 3`. A third was therefore a systematic
        // under-guess: every object whose ratio beat 3.0, which is the majority
        // of the bytes, forced a `reserve` + memcpy inside the drive loop. Half
        // clears the measured mean with headroom and still asks for far less
        // than zlib's own worst-case bound.
        out.reserve((src.len() / 2).max(64) + 64);

        let mut consumed = 0usize;
        loop {
            let in_before = self.z.total_in();
            let out_before = self.z.total_out();

            // Hand the compressor the whole remaining spare capacity of `out`.
            // Growing here rather than using a fixed scratch keeps the copy count
            // at zero: the compressor writes straight into the destination.
            if out.len() == out.capacity() {
                out.reserve(out.capacity().max(1024));
            }
            let spare = out.capacity() - out.len();
            let write_at = out.len();
            // SAFETY: `spare` bytes past `out.len()` are allocated (inside
            // capacity) and are only READ back after `set_len` extends over the
            // prefix the compressor reports it wrote.
            let dst =
                unsafe { std::slice::from_raw_parts_mut(out.as_mut_ptr().add(write_at), spare) };

            let remaining = &src[consumed..];
            let flush = if consumed == src.len() {
                end
            } else {
                FlushCompress::None
            };
            let status = self
                .z
                .compress(remaining, dst, flush)
                .map_err(|_| DeflateError::Encode("compressor returned an error"))?;

            let produced = (self.z.total_out() - out_before) as usize;
            let took = (self.z.total_in() - in_before) as usize;
            // SAFETY: the compressor reported `produced` bytes written into `dst`.
            unsafe { out.set_len(write_at + produced) };
            consumed += took;

            match status {
                Status::StreamEnd => break,
                Status::Ok | Status::BufError => {
                    if consumed == src.len() && produced == 0 && took == 0 {
                        // Input drained, nothing more produced. For `Finish` that
                        // means we should have seen StreamEnd; for `Full` it is
                        // the normal termination.
                        if matches!(end, FlushCompress::Finish) {
                            return Err(DeflateError::Encode(
                                "compressor stalled before StreamEnd",
                            ));
                        }
                        break;
                    }
                    // Otherwise loop: either more input to feed, or the output
                    // buffer filled and needs growing (next iteration reserves).
                    if produced == spare {
                        out.reserve(spare.max(1024));
                    }
                }
            }
        }
        Ok(out.len() - start)
    }
}

// ── The encoder pool ────────────────────────────────────────────────────────

/// A shared pool of reusable [`Encoder`]s, one borrowed per unit of work.
///
/// # Why this type exists
///
/// [`Encoder`] already amortises `Compress::new` (**4082 ns**) down to
/// `reset()` (**1567 ns**) *within* one unit of work. It could not amortise it
/// **across** units, because every scheduler in this crate hands its `map`
/// closure a `Fn`, not an `FnMut` — there is nowhere to keep a `&mut Encoder`.
/// So both schedulers built fresh encoders inside the closure:
///
/// * [`crate::plan::compress_objects`] built **two** per unit (a zlib-wrapped
///   one and a bare one), and for the git path the bare one was *never touched*
///   — 8.2 µs of `Compress::new` per unit, half of it for an encoder that did
///   no work at all.
/// * [`crate::stream::compress_objects_to_writer`] — the 31+1 path the design
///   doc names as what "a real `repack()` should use" — built one **per
///   object**. On a git-median 1 KiB object the whole deflate is ~5 µs, so it
///   was paying 4.1 µs of construction for 0.9 µs of work.
///
/// # Why a pool and not a `thread_local!`
///
/// A thread-local would make the crate non-reentrant: a caller who invoked
/// `compress_objects` from inside another `compress_objects`' map closure on the
/// same thread would hit a `RefCell` double-borrow panic. The pool is owned by
/// one call, so nesting simply builds a second pool.
///
/// # Cost of the pool itself
///
/// Two uncontended `Mutex` operations per **checkout**, ~20 ns each on this box,
/// against the 4082 ns it removes. Contention is bounded by the worker count,
/// and a worker only touches the lock at the start and end of a unit — never
/// per object inside a coalesced batch.
pub struct EncoderPool {
    free: std::sync::Mutex<Vec<Encoder>>,
    /// How many `Compress::new` calls this pool has made. **This, not the length
    /// of the free list, is the thing the reuse guard asserts on** — see
    /// [`EncoderPool::constructed`].
    built: std::sync::atomic::AtomicUsize,
    level: u32,
    zlib_wrapped: bool,
    window_bits: u8,
}

impl EncoderPool {
    /// A pool that hands out encoders at `level` / `zlib_wrapped` /
    /// `window_bits`. Builds nothing up front — the first `checkout` on each
    /// worker pays one `Compress::new` and every later one is free.
    pub fn new(level: u32, zlib_wrapped: bool, window_bits: u8) -> Self {
        EncoderPool {
            free: std::sync::Mutex::new(Vec::new()),
            built: std::sync::atomic::AtomicUsize::new(0),
            level,
            zlib_wrapped,
            window_bits,
        }
    }

    /// Borrow an encoder. Returns it to the pool when the guard drops.
    ///
    /// # Errors
    ///
    /// Only on the very first checkout that has to construct one, and only for
    /// the same reasons [`Encoder::new_with_window`] fails: an out-of-range
    /// level or window.
    pub fn checkout(&self) -> Result<PooledEncoder<'_>, DeflateError> {
        let taken = self
            .free
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .pop();
        let enc = match taken {
            Some(e) => e,
            None => {
                self.built
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                Encoder::new_with_window(self.level, self.zlib_wrapped, self.window_bits)?
            }
        };
        Ok(PooledEncoder {
            pool: self,
            enc: Some(enc),
        })
    }

    /// How many encoders are currently parked in the pool.
    ///
    /// **Do not assert reuse on this.** It is a state round-trip, and it reads
    /// *identically* whether `checkout` reuses the encoder it popped or throws
    /// it away and builds a fresh one — the first version of the reuse guard
    /// asserted on exactly this and passed with reuse deliberately broken
    /// (LAW 2, "your own new guard can be blind"). Use [`constructed`].
    ///
    /// [`constructed`]: EncoderPool::constructed
    pub fn parked(&self) -> usize {
        self.free
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .len()
    }

    /// How many `Compress::new` calls this pool has made over its lifetime.
    ///
    /// This is the **applied output** of the pool: the whole type exists to make
    /// this number small, and it is the only observable that can tell reuse from
    /// reconstruction. A pool that popped an encoder and then built a new one
    /// anyway would leave `parked()` unchanged and this counter climbing.
    pub fn constructed(&self) -> usize {
        self.built.load(std::sync::atomic::Ordering::Relaxed)
    }
}

/// An [`Encoder`] borrowed from an [`EncoderPool`], returned on drop.
pub struct PooledEncoder<'a> {
    pool: &'a EncoderPool,
    // `Option` only so `Drop` can move the encoder back out.
    enc: Option<Encoder>,
}

impl std::ops::Deref for PooledEncoder<'_> {
    type Target = Encoder;
    fn deref(&self) -> &Encoder {
        self.enc.as_ref().expect("live until drop")
    }
}

impl std::ops::DerefMut for PooledEncoder<'_> {
    fn deref_mut(&mut self) -> &mut Encoder {
        self.enc.as_mut().expect("live until drop")
    }
}

impl Drop for PooledEncoder<'_> {
    fn drop(&mut self) {
        if let Some(e) = self.enc.take() {
            self.pool
                .free
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .push(e);
        }
    }
}

/// One-shot: compress `src` into a complete zlib (RFC 1950) stream.
///
/// This is the exact byte sequence a git loose object body and a packfile entry
/// payload carry. Convenience over [`Encoder`]; in a hot loop reuse an
/// `Encoder`.
pub fn zlib_compress(src: &[u8], level: u32) -> Result<Vec<u8>, DeflateError> {
    let mut out = Vec::new();
    Encoder::new(level, true)?.compress_into(src, &mut out)?;
    Ok(out)
}

/// One-shot: compress `src` into a bare DEFLATE (RFC 1951) stream.
pub fn deflate_compress(src: &[u8], level: u32) -> Result<Vec<u8>, DeflateError> {
    let mut out = Vec::new();
    Encoder::new(level, false)?.compress_into(src, &mut out)?;
    Ok(out)
}

// ── Adler-32, and why this crate needs its own ──────────────────────────────
//
// The within-stream split path compresses N segments on N different workers,
// each with its own compressor that has never seen the other segments. No one
// of them can compute the whole-stream Adler-32 the RFC-1950 trailer requires.
// So we compute it per segment and COMBINE — the same `adler32_combine` zlib
// itself exposes for exactly this case. ~40 lines, pure arithmetic, no dep.

/// Largest number of bytes that can be summed before the 32-bit accumulators
/// can overflow: the classic zlib NMAX.
const NMAX: usize = 5552;
const BASE: u32 = 65521;

/// Adler-32 of `data`, seeded with `adler` (use `1` for a fresh checksum).
pub fn adler32(adler: u32, data: &[u8]) -> u32 {
    let mut s1 = adler & 0xffff;
    let mut s2 = (adler >> 16) & 0xffff;
    for block in data.chunks(NMAX) {
        for &b in block {
            s1 += b as u32;
            s2 += s1;
        }
        s1 %= BASE;
        s2 %= BASE;
    }
    (s2 << 16) | s1
}

/// Combine two Adler-32 checksums: `a` over some prefix, `b` over the `len_b`
/// bytes that follow it, giving the checksum of the concatenation.
///
/// This is `adler32_combine` from zlib, reimplemented (it is pure modular
/// arithmetic and pulling a C zlib for 15 lines would violate the no-C rule).
pub fn adler32_combine(a: u32, b: u32, len_b: u64) -> u32 {
    let rem = (len_b % BASE as u64) as u32;
    let a1 = a & 0xffff;
    let a2 = (a >> 16) & 0xffff;
    let b1 = b & 0xffff;
    let b2 = (b >> 16) & 0xffff;

    let mut sum1 = a1 as u64;
    let mut sum2 = (rem as u64 * a1 as u64) % BASE as u64;
    sum1 += b1 as u64 + BASE as u64 - 1;
    sum2 += a2 as u64 + b2 as u64 + BASE as u64 - rem as u64;

    if sum1 >= BASE as u64 {
        sum1 -= BASE as u64;
    }
    if sum1 >= BASE as u64 {
        sum1 -= BASE as u64;
    }
    if sum2 >= (BASE as u64) << 1 {
        sum2 -= (BASE as u64) << 1;
    }
    if sum2 >= BASE as u64 {
        sum2 -= BASE as u64;
    }
    (sum1 as u32) | ((sum2 as u32) << 16)
}

/// The two-byte RFC-1950 zlib header zlib itself emits for `level`.
///
/// CMF = 0x78 (deflate, 32 KiB window). FLG carries FLEVEL in its top two bits
/// and check bits chosen so `(CMF << 8 | FLG) % 31 == 0`. Produces the familiar
/// `78 01` / `78 5E` / `78 9C` / `78 DA` — byte-identical to what a single
/// unsplit `Compress::new(level, true)` writes, so the split and unsplit paths
/// are indistinguishable at the header.
pub fn zlib_header(level: u32) -> [u8; 2] {
    let cmf: u32 = 0x78;
    let flevel: u32 = match level {
        0..=1 => 0,
        2..=5 => 1,
        6 => 2,
        _ => 3,
    };
    let mut flg = flevel << 6;
    flg += 31 - ((cmf << 8 | flg) % 31);
    [cmf as u8, flg as u8]
}

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

    /// RED-when-broken: the header must match the byte-for-byte header a real
    /// `Compress::new(level, true)` emits. A wrong FLEVEL or a wrong check byte
    /// yields a stream `git` rejects with "invalid zlib header" — and this is
    /// the one place the split path hand-writes bytes zlib would otherwise own.
    #[test]
    fn zlib_header_matches_the_real_encoder_at_every_level() {
        for level in 0..=9u32 {
            let real = zlib_compress(b"the quick brown fox", level).unwrap();
            let ours = zlib_header(level);
            assert_eq!(
                &real[..2],
                &ours[..],
                "level {level}: hand-written zlib header {ours:02x?} != encoder's {:02x?}",
                &real[..2]
            );
            // The RFC-1950 check itself.
            let word = (ours[0] as u32) << 8 | ours[1] as u32;
            assert_eq!(word % 31, 0, "level {level}: header fails the %31 check");
        }
    }

    /// RED-when-broken: `adler32_combine` must equal a straight `adler32` over
    /// the concatenation, for many split points including the degenerate ones.
    /// A wrong `rem`, a missing `BASE` wrap, or swapped operands all diverge
    /// here — and would silently corrupt the trailer of every split stream.
    #[test]
    fn adler_combine_equals_whole_buffer_adler() {
        let data: Vec<u8> = (0..40_000u32)
            .map(|i| (i.wrapping_mul(2654435761) >> 13) as u8)
            .collect();
        let whole = adler32(1, &data);
        for &cut in &[0usize, 1, 2, 5551, 5552, 5553, 12_345, 39_999, 40_000] {
            let (a, b) = data.split_at(cut);
            let combined = adler32_combine(adler32(1, a), adler32(1, b), b.len() as u64);
            assert_eq!(
                combined, whole,
                "combine at cut={cut} gave {combined:08x}, whole is {whole:08x}"
            );
        }
        // Three-way, left-folded — the shape the split path actually uses.
        let mut acc = adler32(1, &data[..1000]);
        acc = adler32_combine(acc, adler32(1, &data[1000..25_000]), 24_000);
        acc = adler32_combine(acc, adler32(1, &data[25_000..]), 15_000);
        assert_eq!(acc, whole, "left-folded 3-way combine must equal the whole");
    }

    /// The pool must **reuse**, not merely hand out encoders. A pool that built
    /// a fresh `Compress` on every checkout would pass every correctness test in
    /// this crate while buying nothing at all — it is the whole point of the
    /// type, and the only observable difference is that the free list is
    /// non-empty after a guard drops.
    #[test]
    fn the_pool_actually_reuses_and_the_check_can_see_the_difference() {
        let pool = EncoderPool::new(6, true, WINDOW_BITS_DEFAULT);
        // Nothing built up front. Eager construction is exactly what the git
        // path must not pay for the bare encoder it never touches.
        assert_eq!(
            pool.constructed(),
            0,
            "the pool must build nothing until asked"
        );
        assert_eq!(pool.parked(), 0);

        let mut out = Vec::new();
        {
            let mut e = pool.checkout().unwrap();
            e.compress_into(b"the quick brown fox", &mut out).unwrap();
        }
        assert_eq!(
            pool.constructed(),
            1,
            "the first checkout must build exactly one"
        );
        assert_eq!(
            pool.parked(),
            1,
            "dropping the guard must return the encoder"
        );

        // ── THE assertion. Twenty more sequential checkouts must build NOTHING.
        //
        // This is asserted on the construction counter and NOT on `parked()`,
        // because `parked()` is blind here: a `checkout` that pops an encoder,
        // discards it and builds a fresh one leaves the free-list length
        // identical at every point above. The first version of this test did
        // exactly that and PASSED with reuse deliberately broken.
        for i in 0..20 {
            let mut e = pool.checkout().unwrap();
            out.clear();
            e.compress_into(b"another object entirely", &mut out)
                .unwrap();
            drop(e);
            assert_eq!(
                pool.constructed(),
                1,
                "checkout {i} constructed a new encoder instead of reusing the parked one — \
                 the pool is buying nothing"
            );
        }

        // Two guards live at once must be two DISTINCT encoders — otherwise a
        // coalesced batch and a segment on the same worker would fight over one
        // state machine — and both must come back.
        {
            let _a = pool.checkout().unwrap();
            let _b = pool.checkout().unwrap();
            assert_eq!(
                pool.constructed(),
                2,
                "the second concurrent guard needs its own"
            );
            assert_eq!(pool.parked(), 0);
        }
        assert_eq!(pool.parked(), 2, "both guards must return their encoder");
    }

    /// The pool's whole purpose, asserted end-to-end on the real scheduler: a
    /// `compress_objects` call over many objects must construct **at most one
    /// encoder per worker**, not one per unit and certainly not one per object.
    ///
    /// Before the pool, [`crate::plan::compress_objects_serial`] built two
    /// encoders per *unit* — including a bare one the zlib path never touched.
    #[test]
    fn the_serial_scheduler_builds_one_encoder_for_the_whole_call() {
        use crate::plan::{Config, compress_objects_serial};
        let data: Vec<Vec<u8>> = (0..4000)
            .map(|i| format!("object {i} ").repeat(30).into_bytes())
            .collect();
        let r: Vec<&[u8]> = data.iter().map(|v| v.as_slice()).collect();
        // Force one unit per object, so a per-UNIT construction would be 4000.
        let cfg = Config::default().without_coalescing();
        let out = compress_objects_serial(&r, &cfg).unwrap();
        assert_eq!(out.len(), data.len());
        // The construction count is not reachable from outside the call, so this
        // asserts the consequence instead: the same work with coalescing OFF
        // must still produce byte-identical output to coalescing ON. If the
        // scheduler were building fresh encoders per unit the bytes would still
        // match — so this test alone is NOT the reuse evidence; the counter test
        // above is. Kept because it pins the shape the counter test asserts on.
        let coalesced = compress_objects_serial(&r, &Config::default()).unwrap();
        assert_eq!(out, coalesced, "coalescing must not change one output byte");
    }

    /// RED-when-broken, the other half: a pooled encoder that was *not* reset
    /// between objects would carry the previous stream's state and produce
    /// garbage. Drive one pooled encoder over many different objects, checking
    /// out and returning it each time, and inflate every result back.
    #[test]
    fn a_recycled_encoder_still_starts_a_clean_stream_every_time() {
        use std::io::Read;
        let pool = EncoderPool::new(6, true, WINDOW_BITS_DEFAULT);
        let corpus: Vec<Vec<u8>> = (0..64)
            .map(|i| {
                let tag = format!("object-{i}-");
                let mut v = Vec::new();
                while v.len() < 40 + i * 137 {
                    v.extend_from_slice(tag.as_bytes());
                }
                v
            })
            .collect();
        for (i, src) in corpus.iter().enumerate() {
            let mut z = Vec::new();
            pool.checkout().unwrap().compress_into(src, &mut z).unwrap();
            // Byte-identical to a fresh encoder's output — the strong form. A
            // leaked LZ77 window would still inflate (zlib would just have a
            // different history) but would NOT match, so this catches the
            // silent-corruption case a round-trip alone would miss.
            assert_eq!(
                z,
                zlib_compress(src, 6).unwrap(),
                "object {i}: a recycled encoder produced different bytes from a fresh one"
            );
            let mut got = Vec::new();
            flate2::read::ZlibDecoder::new(&z[..])
                .read_to_end(&mut got)
                .unwrap();
            assert_eq!(&got, src, "object {i} round-trip differs");
        }
    }

    /// The known-answer test for adler32 itself, so a broken `adler32` cannot
    /// make the (self-consistent) combine test above pass vacuously — the
    /// identity-value trap from LAW 2.
    #[test]
    fn adler32_known_answers() {
        assert_eq!(adler32(1, b""), 1);
        assert_eq!(adler32(1, b"a"), 0x00620062);
        assert_eq!(adler32(1, b"abc"), 0x024d0127);
        assert_eq!(adler32(1, b"Wikipedia"), 0x11E60398);
        // Crosses NMAX so the modular reduction between blocks is exercised.
        let big = vec![0xffu8; 20_000];
        let mut ref_s1: u64 = 1;
        let mut ref_s2: u64 = 0;
        for &b in &big {
            ref_s1 = (ref_s1 + b as u64) % BASE as u64;
            ref_s2 = (ref_s2 + ref_s1) % BASE as u64;
        }
        assert_eq!(adler32(1, &big), ((ref_s2 as u32) << 16) | ref_s1 as u32);
    }

    /// The kernel must round-trip through BOTH decoders — `flate2` (which shares
    /// zlib-rs with the encoder, so it could in principle agree with a private
    /// bug) and `linflate` (an entirely independent, hand-rolled decoder). Two
    /// implementations agreeing is real evidence; one agreeing with itself is
    /// not.
    #[test]
    fn kernel_roundtrips_through_both_decoders() {
        let corpus: Vec<Vec<u8>> = vec![
            Vec::new(),
            b"x".to_vec(),
            b"the quick brown fox jumps over the lazy dog".repeat(200),
            (0..70_000u32).map(|i| (i % 251) as u8).collect(),
            (0..200_000u32)
                .map(|i| (i.wrapping_mul(2654435761) >> 11) as u8)
                .collect(),
        ];
        for (n, src) in corpus.iter().enumerate() {
            for level in [0u32, 1, 6, 9] {
                let z = zlib_compress(src, level).unwrap();
                // flate2 side.
                let mut got = Vec::new();
                use std::io::Read;
                flate2::read::ZlibDecoder::new(&z[..])
                    .read_to_end(&mut got)
                    .unwrap();
                assert_eq!(
                    &got, src,
                    "corpus {n} level {level}: flate2 round-trip differs"
                );

                // linflate side — raw DEFLATE, so strip the 2-byte header and
                // 4-byte trailer ourselves.
                if !src.is_empty() {
                    let raw = &z[2..z.len() - 4];
                    let mut buf = vec![0u8; src.len() + linflate::OVERWRITE_HEADROOM];
                    let wrote = linflate::inflate_into(raw, &mut buf).unwrap();
                    assert_eq!(
                        &buf[..wrote],
                        &src[..],
                        "corpus {n} level {level}: linflate differs"
                    );
                }
            }
        }
    }
}