ldeflate 0.1.0

//! 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 third and let the loop grow.
        out.reserve((src.len() / 3).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)
    }
}

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