ldeflate/src/unit.rs
//! 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"
);
}
}
}
}
}