ldeflate 0.1.3

//! **gzip (RFC 1952)** — framing over the same engine, in two shapes.
//!
//! There is almost no new code here and that is the point (LAW 5). A gzip member
//! is a 10-byte header, a bare DEFLATE body, a CRC-32 and a length. The
//! scheduling — the two-queue planner, the coalescing, the split at
//! `Z_FULL_FLUSH`, the no-barrier pass — is `ldeflate`'s and is reused verbatim
//! by setting [`Container::Gzip`].
//!
//! # Two shapes, and they are not interchangeable
//!
//! **[`gzip_compress`] — one member, split internally.** The DEFLATE body is cut
//! at full-flush boundaries and the pieces are compressed on different workers,
//! then concatenated under a single header and a single combined CRC-32. This is
//! what `pigz` emits by default. Every gzip tool reads it, but **no gzip tool can
//! decompress it in parallel**, because the segment boundaries are invisible
//! from outside without scanning for them.
//!
//! **[`gzip_compress_members`] — many complete members, concatenated.** RFC 1952
//! §2.2 makes a gzip file "a series of members", and stock `gzip` / `zcat`
//! decompress the concatenation transparently. Each member is independently
//! decodable, so this is the shape `lgz` — the workspace's own parallel gzip
//! *reader* — can fan out over.
//!
//! That second shape closes a loop worth naming. The repo's bench harness builds
//! its multi-member corpus with a shell pipeline:
//!
//! ```text
//! split -b 8M … | xargs -P N gzip -6 | cat > corpus.gz
//! ```
//!
//! `gzip_compress_members` is that pipeline, in-process, without `split`,
//! `xargs`, `gzip` or a temporary directory.
//!
//! # Cost of the member shape
//!
//! Independent members cannot share an LZ77 window, so the ratio is worse than a
//! single stream — the same trade the split path makes, in a form the reader can
//! see. Measured numbers are in `.nornir/ldeflate-design.md`; pick the member
//! size from there rather than guessing.

use crate::container::Container;
use crate::plan::{Config, compress_objects};
use crate::unit::DeflateError;

/// Compress `src` into **one** gzip member, splitting the DEFLATE body across
/// workers at full-flush boundaries.
///
/// Reads back under `gzip -d`, `zcat`, `pigz -d`, `flate2` and `lgz`. Output is
/// deterministic: `MTIME` is written as zero (see [`Container::header`]) and the
/// segment plan is a pure function of the input length and `cfg.segment`, so the
/// bytes do not depend on the worker count.
pub fn gzip_compress(src: &[u8], cfg: &Config) -> Result<Vec<u8>, DeflateError> {
    let one: [&[u8]; 1] = [src];
    // `split_above: 1` forces the split queue even for a small input; the plan
    // then produces one segment per `cfg.segment` bytes.
    let c = Config {
        container: Container::Gzip,
        split_above: 1,
        coalesce_below: 0,
        ..*cfg
    };
    Ok(compress_objects(&one, &c)?
        .pop()
        .expect("one input, one output"))
}

/// Compress `src` into a **concatenation of independent gzip members** of about
/// `member` uncompressed bytes each.
///
/// Stock `gzip` and `zcat` read the concatenation as one stream. Unlike
/// [`gzip_compress`], each member here is independently decodable, so a parallel
/// gzip *reader* (`lgz`) can fan out over it.
///
/// `member` is treated as at least 1; a zero-length input still produces one
/// (empty) member, because a gzip file with no members is not a gzip file.
///
/// # Errors
///
/// [`DeflateError::TooManyPieces`] if `src.len() / member` exceeds
/// [`crate::limits::MAX_PIECES`]. Each member carries a 10-byte header, its own
/// DEFLATE state and an 8-byte trailer, so a `member` of 1 against a few
/// megabytes is a request for millions of independent compressors — refused
/// here **before** the first allocation. The ceiling errors rather than
/// enlarging `member` for you, because a silently different member size is a
/// silently different output file (see [`crate::limits`]).
pub fn gzip_compress_members(
    src: &[u8],
    cfg: &Config,
    member: usize,
) -> Result<Vec<u8>, DeflateError> {
    crate::limits::check_pieces("gzip members", src.len(), member)?;
    let member = member.max(1);
    // Slice the input into member-sized views. Zero copy: these are borrows into
    // the caller's buffer, and the planner never copies them into the pool.
    let chunks: Vec<&[u8]> = if src.is_empty() {
        vec![&src[..]]
    } else {
        src.chunks(member).collect()
    };
    // Each member is one whole object to the planner. Splitting is disabled
    // because a member IS the unit of parallelism here; coalescing is disabled
    // because merging two members into one task would be pointless work-hiding
    // (they are already large by construction).
    // Resolve the already-compressed probe against the WHOLE input, once, and
    // hand every member the answer. Probing per member would be wrong in the
    // only direction that matters: the magic bytes live at offset 0 of the file,
    // so member 0 would be stored and members 1..n — which are slices out of the
    // middle of the same PNG or packfile — would each be handed to the entropy
    // coder that this exists to avoid. Same reasoning as the split path in
    // `crate::split::compress_segment_skipping`, and it keeps the output
    // independent of `member`.
    let resolved = if cfg.skip.skip(src) {
        crate::precompressed::SkipPolicy::already_compressed()
    } else {
        crate::precompressed::SkipPolicy::compress_everything()
    };
    let c = Config {
        container: Container::Gzip,
        coalesce_below: 0,
        split_above: usize::MAX,
        skip: resolved,
        ..*cfg
    };
    let members = compress_objects(&chunks, &c)?;
    let total: usize = members.iter().map(|m| m.len()).sum();
    let mut out = Vec::with_capacity(total);
    for m in &members {
        out.extend_from_slice(m);
    }
    Ok(out)
}

/// Split `data` at gzip member boundaries, returning one slice per member.
///
/// The inverse of [`gzip_compress_members`], and the check that the members it
/// wrote really are independent: each returned slice must be a complete,
/// self-contained gzip stream. Returns `None` if `data` is not a concatenation
/// of well-formed members (a truncated tail, or a header this does not
/// understand).
///
/// Deliberately conservative — it only handles the headers this crate writes
/// (`FLG == 0`, no FEXTRA/FNAME/FCOMMENT/FHCRC), because its job is to verify our
/// own output, not to be a general gzip parser. `lgz` is the general reader.
pub fn gzip_members_of(data: &[u8]) -> Option<Vec<&[u8]>> {
    let mut out = Vec::new();
    let mut i = 0usize;
    while i < data.len() {
        let rest = &data[i..];
        if rest.len() < 18 || rest[0] != 0x1f || rest[1] != 0x8b || rest[2] != 0x08 {
            return None;
        }
        if rest[3] != 0 {
            return None; // optional fields present; not something we write
        }
        // Inflate the body to find where it ends. `flate2` reports how many
        // input bytes it consumed, which is the only reliable way to find a
        // member boundary without a full DEFLATE parser.
        let body = &rest[10..];
        let mut d = flate2::Decompress::new(false);
        let mut sink = Vec::new();
        let mut scratch = vec![0u8; 64 * 1024];
        loop {
            let before_in = d.total_in();
            let before_out = d.total_out();
            let status = d
                .decompress(
                    &body[(before_in as usize).min(body.len())..],
                    &mut scratch,
                    flate2::FlushDecompress::None,
                )
                .ok()?;
            let produced = (d.total_out() - before_out) as usize;
            sink.extend_from_slice(&scratch[..produced]);
            match status {
                flate2::Status::StreamEnd => break,
                _ => {
                    if d.total_in() == before_in && produced == 0 {
                        return None; // stalled: truncated member
                    }
                }
            }
        }
        let consumed = 10 + d.total_in() as usize + 8; // header + body + trailer
        if i + consumed > data.len() {
            return None;
        }
        out.push(&data[i..i + consumed]);
        i += consumed;
    }
    if out.is_empty() { None } else { Some(out) }
}

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

    fn corpus(n: usize) -> Vec<u8> {
        let words: [&[u8]; 4] = [
            b"2026-08-03T10:15:00Z INFO request id=af39c0 status=200\n",
            b"pub fn run() -> usize { let x = 1; x + 2 }\n",
            b"the quick brown fox jumps over the lazy dog\n",
            b"use std::collections::HashMap;\n",
        ];
        let mut v = Vec::with_capacity(n + 64);
        let mut s: u64 = 0x9E3779B97F4A7C15;
        while v.len() < n {
            s = s
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            v.extend_from_slice(words[(s >> 33) as usize % words.len()]);
        }
        v.truncate(n);
        v
    }

    fn ungzip(z: &[u8]) -> Vec<u8> {
        let mut out = Vec::new();
        flate2::read::MultiGzDecoder::new(z)
            .read_to_end(&mut out)
            .unwrap();
        out
    }

    /// The single-member shape at many sizes, through a decoder that verifies
    /// the CRC-32 and ISIZE trailer.
    #[test]
    fn the_single_member_shape_round_trips_at_every_size() {
        for &len in &[0usize, 1, 4095, 4096, 100_000, 3_000_000] {
            let src = corpus(len);
            let single = gzip_compress(&src, &Config::default()).unwrap();
            assert_eq!(ungzip(&single), src, "single-member len={len}");
            assert_eq!(
                &single[..3],
                &[0x1f, 0x8b, 0x08],
                "len={len}: not a gzip header"
            );
        }
    }

    /// **Degenerate member sizes, against SMALL inputs.** `m = 1` — one member
    /// per byte — is exactly the boundary worth testing, and it is cheap here
    /// because the largest input is 8 KiB.
    ///
    /// **Do not re-cross this with the large inputs below.** `m = 1` against
    /// `len = 3_000_000` is three million gzip members, each with its own
    /// 10-byte header, DEFLATE state and 8-byte trailer. That single cross —
    /// the line this test replaces — allocated ~497 GB on 2026-08-03, the OOM
    /// killer fired, and it took three unrelated tmux sessions down with it.
    /// Measured peak RSS for that one test under a 24 GB address-space cap:
    /// 20.9 GB. The library now refuses that request outright — see
    /// `a_degenerate_member_size_is_refused_not_attempted`.
    #[test]
    fn degenerate_member_sizes_round_trip_on_small_inputs() {
        let cfg = Config::default();
        for &len in &[0usize, 1, 2, 3, 4095, 4096, 8192] {
            let src = corpus(len);
            for &m in &[1usize, 2, 3, 4096] {
                let multi = gzip_compress_members(&src, &cfg, m).unwrap();
                assert_eq!(ungzip(&multi), src, "multi-member len={len} member={m}");
            }
        }
    }

    /// **Large inputs, against REALISTIC member sizes.** The size axis, kept
    /// separate from the degenerate-member axis above for the reason stated
    /// there.
    #[test]
    fn large_inputs_round_trip_at_realistic_member_sizes() {
        let cfg = Config::default();
        for &len in &[100_000usize, 3_000_000] {
            let src = corpus(len);
            for &m in &[64 << 10, 1 << 20] {
                let multi = gzip_compress_members(&src, &cfg, m).unwrap();
                assert_eq!(ungzip(&multi), src, "multi-member len={len} member={m}");
            }
        }
    }

    /// **The OOM guard.** A `member` size that would produce an absurd number
    /// of members must come back as an `Err` **without allocating**, not as a
    /// 497 GB attempt.
    ///
    /// Seen red on 2026-08-03 with the ceiling check removed from
    /// `gzip_compress_members`: the first case below returned `Ok` after ~1 s
    /// and ~0.9 GB of RSS, and this test failed on `is_err()`. That is the whole
    /// point — the assertion catches a library that says yes.
    ///
    /// The cases are ordered cheapest-first deliberately, so a regression trips
    /// on the 256 KB case and the test never reaches the 3 MB one.
    #[test]
    fn a_degenerate_member_size_is_refused_not_attempted() {
        let cfg = Config::default();

        // Just over the ceiling: one member per byte of a 256 KiB buffer.
        let just_over = crate::limits::MAX_PIECES + 1;
        let e = gzip_compress_members(&vec![b'x'; just_over], &cfg, 1)
            .expect_err("one member per byte over the ceiling must be refused");
        assert!(
            matches!(e, DeflateError::TooManyPieces { pieces, .. } if pieces == just_over),
            "wrong error: {e}"
        );
        // The message has to name the request and the ceiling, or a caller
        // cannot tell what to pass instead.
        let m = e.to_string();
        assert!(
            m.contains("gzip members"),
            "message does not name the axis: {m}"
        );
        assert!(
            m.contains(&just_over.to_string()),
            "message does not name the request: {m}"
        );
        assert!(
            m.contains(&crate::limits::MAX_PIECES.to_string()),
            "message does not name the ceiling: {m}"
        );

        // The exact shape that took the box down: 3 MB at one byte per member.
        assert!(
            gzip_compress_members(&vec![b'x'; 3_000_000], &cfg, 1).is_err(),
            "3 million one-byte gzip members were accepted — this is the 2026-08-03 OOM"
        );
        // `member = 0` must not be a way around it (it floors to 1).
        assert!(
            gzip_compress_members(&vec![b'x'; 3_000_000], &cfg, 0).is_err(),
            "member=0 escaped"
        );

        // ...and the guard is not simply refusing everything: one under the
        // ceiling, and every realistic member size, still work.
        assert!(
            gzip_compress_members(&vec![b'x'; 4096], &cfg, 1).is_ok(),
            "4096 members at m=1"
        );
        assert!(gzip_compress_members(&vec![b'x'; 3_000_000], &cfg, 64 << 10).is_ok());
    }

    /// **The property the member shape exists for**: every member must be
    /// independently decodable. Split the concatenation and decompress each
    /// piece ALONE — if any piece needs its predecessors, the shape is a lie and
    /// `lgz` could not fan out over it.
    #[test]
    fn every_member_decodes_alone() {
        let src = corpus(2_000_000);
        let z = gzip_compress_members(&src, &Config::default(), 256 * 1024).unwrap();
        let members = gzip_members_of(&z).expect("output must parse as a member sequence");
        assert!(
            members.len() >= 7,
            "expected ~8 members, got {}",
            members.len()
        );

        let mut rebuilt = Vec::new();
        for (i, m) in members.iter().enumerate() {
            let mut got = Vec::new();
            flate2::read::GzDecoder::new(*m)
                .read_to_end(&mut got)
                .unwrap_or_else(|e| panic!("member {i} does not decode standalone: {e}"));
            rebuilt.extend_from_slice(&got);
        }
        assert_eq!(
            rebuilt, src,
            "members decoded alone must concatenate to the input"
        );
    }

    /// RED-when-broken: the single-member shape is NOT a member sequence, so
    /// `every_member_decodes_alone` is testing something real. If
    /// `gzip_members_of` returned several pieces for a split single member, the
    /// test above would pass trivially.
    #[test]
    fn a_split_single_member_is_still_exactly_one_member() {
        let src = corpus(2_000_000);
        let z = gzip_compress(
            &src,
            &Config {
                segment: 64 << 10,
                ..Config::default()
            },
        )
        .unwrap();
        let members = gzip_members_of(&z).expect("must parse");
        assert_eq!(
            members.len(),
            1,
            "a full-flush-split body must remain ONE gzip member; got {} — the split \
             path is emitting member headers and the two shapes are not distinct",
            members.len()
        );
        assert_eq!(ungzip(&z), src);
    }

    /// The CRC-32 trailer must actually be checked by the decoder, or every
    /// round-trip test above proves nothing about `crc32_combine`.
    #[test]
    fn a_wrong_crc_trailer_is_rejected() {
        let src = corpus(300_000);
        let mut z = gzip_compress(&src, &Config::default()).unwrap();
        assert_eq!(ungzip(&z), src, "must read clean before corruption");
        let n = z.len();
        z[n - 5] ^= 0xff; // last byte of the CRC field
        let mut out = Vec::new();
        let r = flate2::read::MultiGzDecoder::new(&z[..]).read_to_end(&mut out);
        assert!(
            r.is_err(),
            "the decoder did NOT verify the gzip CRC-32 — the round-trip tests in \
             this module say nothing about crc32_combine"
        );
    }

    /// Deterministic across worker counts, and MTIME really is zero (a
    /// timestamp would make every run differ and silently break that promise).
    #[test]
    fn deterministic_and_no_timestamp() {
        let src = corpus(4_000_000);
        let base = gzip_compress(&src, &Config::default().workers(1)).unwrap();
        for w in [2usize, 8, 32] {
            assert_eq!(
                gzip_compress(&src, &Config::default().workers(w)).unwrap(),
                base,
                "workers={w}: gzip output differs across worker counts"
            );
        }
        assert_eq!(
            &base[4..8],
            &[0, 0, 0, 0],
            "MTIME must be zero for reproducibility"
        );
        let m1 = gzip_compress_members(&src, &Config::default().workers(1), 1 << 20).unwrap();
        let m2 = gzip_compress_members(&src, &Config::default().workers(16), 1 << 20).unwrap();
        assert_eq!(m1, m2, "multi-member output differs across worker counts");
    }

    /// The member shape costs ratio (no shared window across members) and the
    /// cost must be visible and bounded. If members were sharing a window they
    /// would not be independent.
    #[test]
    fn member_shape_costs_ratio_and_that_proves_independence() {
        let src = corpus(8 << 20);
        let cfg = Config::default();
        let single = gzip_compress(
            &src,
            &Config {
                segment: 8 << 20,
                ..cfg
            },
        )
        .unwrap();
        let multi = gzip_compress_members(&src, &cfg, 256 * 1024).unwrap();
        assert!(
            multi.len() > single.len(),
            "independent members ({} B) must be LARGER than one stream ({} B); if not, \
             they are sharing an LZ77 window and are not independent",
            multi.len(),
            single.len()
        );
        let overhead = multi.len() as f64 / single.len() as f64;
        assert!(
            overhead < 1.5,
            "member overhead {:.1}% is implausibly large",
            (overhead - 1.0) * 100.0
        );
    }
}