ldeflate 0.1.3

//! **Axis B — split within one stream.** The write-side mirror of
//! `linflate::deflate_scan`.
//!
//! `deflate_scan::find_all_flushes` *searches* a compressed stream for
//! `00 00 FF FF` full-flush markers so the read side can hand independent
//! segments to independent workers. This module *creates* them: one large object
//! is cut into fixed byte ranges of the **uncompressed** input, each range is
//! compressed by a different worker with a compressor that has never seen the
//! others, and the results are concatenated.
//!
//! ```text
//!   input:   [------- seg 0 -------][------- seg 1 -------][-- seg 2 --]
//!               worker 7               worker 2              worker 19
//!   output:  78 9C │ deflate·FULL     │ deflate·FULL        │ deflate·FIN │ adler32
//!                  └─ 00 00 FF FF ────┴─ 00 00 FF FF ───────┘
//! ```
//!
//! ## Why the concatenation is a legal zlib stream
//!
//! `Z_FULL_FLUSH` does two things and this design leans on both. It byte-aligns
//! the bit stream (so the next segment's first byte lands on a byte boundary and
//! can be appended verbatim), and it **empties the LZ77 window** (so no
//! back-reference in segment *k+1* can point into segment *k*, which is exactly
//! why segment *k+1* could be compressed by a worker that never saw segment
//! *k*). The final segment is `Z_FINISH`, which sets `BFINAL=1`. An RFC-1950
//! header goes in front and the whole-input Adler-32 — assembled from the
//! per-segment checksums via `adler32_combine` — goes behind. A stock decoder
//! (`zlib`, `git`, `flate2`, `linflate`) sees one ordinary stream of blocks.
//!
//! This is pigz's trick. It is not novel; what matters is that it costs a small,
//! *measurable* amount of ratio (each window reset forecloses long-range
//! matches) and that the ratio cost is reported honestly rather than hidden.
//!
//! ## Determinism
//!
//! Segment boundaries are a function of `(input length, segment size)` **only** —
//! never of the worker count, never of completion order. So a 4-thread run and a
//! 32-thread run emit byte-identical output. That is asserted, not assumed
//! ([`crate::tests`] / `tests/determinism.rs`).

use crate::container::Container;
use crate::unit::{DeflateError, Encoder};

/// One segment of a split stream: where it came from and what it produced.
#[derive(Debug, Clone)]
pub struct Segment {
    /// Byte range of the **uncompressed** input this segment covers.
    pub src_start: usize,
    pub src_end: usize,
    /// The segment's compressed bytes (bare DEFLATE, ending in a full-flush
    /// marker unless this is the final segment).
    pub bytes: Vec<u8>,
    /// Checksum of `input[src_start..src_end]` under the container's algorithm
    /// (Adler-32 seeded at 1 for zlib, CRC-32 seeded at 0 for gzip). Computed on
    /// the worker over that worker's own bytes, so it parallelises with the
    /// compression instead of becoming a serial tail.
    pub sum: u32,
}

/// Plan the segment boundaries for an input of `len` bytes at `segment` size.
///
/// Returns `(start, end, is_last)` triples. Deliberately a pure function of its
/// two arguments — this is what makes the compressor deterministic across
/// thread counts.
///
/// # Errors
///
/// [`DeflateError::TooManyPieces`] if `len / segment` exceeds
/// [`crate::limits::MAX_PIECES`]. A `segment` of 1 against a megabyte asks for
/// a million independently compressed, independently checksummed pieces; this
/// refuses **before** the `Vec` is built rather than trying. The ceiling errors
/// instead of clamping `segment` upward because the segment plan is exactly
/// what carries this crate's byte-identical-output promise — see
/// [`crate::limits`].
pub fn plan_segments(
    len: usize,
    segment: usize,
) -> Result<Vec<(usize, usize, bool)>, DeflateError> {
    let n = crate::limits::check_pieces("split segments", len, segment)?;
    if len == 0 {
        return Ok(vec![(0, 0, true)]);
    }
    let segment = segment.max(1);
    Ok((0..n)
        .map(|i| {
            let start = i * segment;
            let end = ((i + 1) * segment).min(len);
            (start, end, i + 1 == n)
        })
        .collect())
}

/// Compress one segment. Called on a worker thread, with `enc` being that
/// worker's reusable encoder.
///
/// `is_last` selects `Z_FINISH` (BFINAL=1) over `Z_FULL_FLUSH`.
pub fn compress_segment(
    enc: &mut Encoder,
    src: &[u8],
    start: usize,
    end: usize,
    is_last: bool,
    container: Container,
) -> Result<Segment, DeflateError> {
    compress_segment_skipping(enc, src, start, end, is_last, container, false)
}

/// [`compress_segment`], optionally storing the segment instead of deflating it.
///
/// `store` is decided by the caller from the **whole** object (see
/// [`crate::precompressed`]), never from this segment's slice: a segment cut out
/// of the middle of a PNG carries no magic bytes of its own, and a decision made
/// per slice would make the output depend on where the cuts landed.
///
/// A stored non-final segment ends on a block boundary, which is the same seam
/// `Z_FULL_FLUSH` leaves, so stored and deflated segments concatenate freely.
pub fn compress_segment_skipping(
    enc: &mut Encoder,
    src: &[u8],
    start: usize,
    end: usize,
    is_last: bool,
    container: Container,
    store: bool,
) -> Result<Segment, DeflateError> {
    debug_assert!(!enc.is_zlib_wrapped(), "segments must be bare DEFLATE");
    let chunk = &src[start..end];
    let mut bytes = Vec::new();
    if store {
        bytes = crate::precompressed::stored_deflate(chunk, is_last);
    } else if is_last {
        enc.compress_into(chunk, &mut bytes)?;
    } else {
        enc.compress_segment_into(chunk, &mut bytes)?;
    }
    let sum = container.checksum(chunk);
    Ok(Segment {
        src_start: start,
        src_end: end,
        bytes,
        sum,
    })
}

/// Assemble planned segments (in stream order) into one complete zlib stream.
///
/// Panics in debug if the segments are not contiguous and in order — that would
/// be a scheduler bug, and a silently mis-ordered assembly produces a stream
/// that inflates to *wrong bytes* rather than failing, which is the worst
/// possible failure mode.
pub fn assemble(level: u32, segments: &[Segment], container: Container) -> Vec<u8> {
    assemble_checksummed(level, segments, container).0
}

/// [`assemble`], also returning the folded whole-stream checksum.
///
/// ZIP stores each entry's CRC-32 in its local and central headers rather than
/// in a trailer, so it needs the folded value that [`assemble`] would otherwise
/// consume and discard.
pub fn assemble_checksummed(
    level: u32,
    segments: &[Segment],
    container: Container,
) -> (Vec<u8>, u32) {
    debug_assert!(!segments.is_empty(), "assemble needs at least one segment");
    debug_assert!(
        segments.windows(2).all(|w| w[0].src_end == w[1].src_start),
        "segments must be contiguous and in stream order"
    );

    let body: usize = segments.iter().map(|s| s.bytes.len()).sum();
    let total: u64 = segments
        .iter()
        .map(|s| (s.src_end - s.src_start) as u64)
        .sum();
    let mut out = Vec::with_capacity(12 + body + 8);
    out.extend_from_slice(&container.header(level));

    // Fold the per-segment checksums into the whole-stream one. This is the
    // only reason `adler32_combine` / `crc32_combine` exist in this crate.
    let mut sum = container.checksum_init();
    for s in segments {
        out.extend_from_slice(&s.bytes);
        let n = (s.src_end - s.src_start) as u64;
        sum = container.checksum_combine(sum, s.sum, n);
    }
    out.extend_from_slice(&container.trailer(sum, total));
    (out, sum)
}

/// Serial reference implementation of the split path — used by tests as the
/// oracle the parallel scheduler must match byte-for-byte, and by
/// [`crate::plan`] when a caller asks for one worker.
pub fn split_compress_serial(
    src: &[u8],
    level: u32,
    segment: usize,
) -> Result<Vec<u8>, DeflateError> {
    split_compress_serial_in(src, level, segment, Container::Zlib)
}

/// [`split_compress_serial`] in an explicit container.
pub fn split_compress_serial_in(
    src: &[u8],
    level: u32,
    segment: usize,
    container: Container,
) -> Result<Vec<u8>, DeflateError> {
    let mut enc = Encoder::new(level, false)?;
    let plan = plan_segments(src.len(), segment)?;
    let mut segs = Vec::with_capacity(plan.len());
    for (start, end, last) in plan {
        segs.push(compress_segment(
            &mut enc, src, start, end, last, container,
        )?);
    }
    Ok(assemble(level, &segs, container))
}

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

    fn corpus(n: usize) -> Vec<u8> {
        // Compressible, but with enough entropy that segmentation actually
        // costs ratio — a corpus of all-zeros would hide any boundary bug
        // because every segment compresses to nothing (LAW 2: identity value).
        let words: &[&[u8]] = &[
            b"fn main() { println!(\"hello\"); }\n",
            b"use std::collections::HashMap;\n",
            b"    let mut total = 0usize;\n",
            b"// TODO: revisit this once the planner lands\n",
        ];
        let mut out = Vec::with_capacity(n + 64);
        let mut s: u64 = 0x9E3779B97F4A7C15;
        while out.len() < n {
            s = s
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            out.extend_from_slice(words[(s >> 33) as usize % words.len()]);
        }
        out.truncate(n);
        out
    }

    /// Assert one (len, seg) pair round-trips through a stock decoder.
    fn round_trips(len: usize, seg: usize) {
        let src = corpus(len);
        let z = split_compress_serial(&src, 6, seg).unwrap();
        let mut got = Vec::new();
        flate2::read::ZlibDecoder::new(&z[..])
            .read_to_end(&mut got)
            .unwrap_or_else(|e| panic!("len={len} seg={seg}: inflate failed: {e}"));
        assert_eq!(got, src, "len={len} seg={seg}: bytes differ");
    }

    /// **Degenerate segment sizes, against SMALL inputs.** `seg = 1` and
    /// `seg = 2` are exactly the boundary cases worth testing — an empty final
    /// segment, a one-byte final segment, a segment count that is not a divisor
    /// — and they are cheap here because the largest input is 8 KiB, i.e. at
    /// most ~8 k pieces.
    ///
    /// **Do not re-cross these with the large inputs below.** `seg = 1` against
    /// a megabyte is a million independently compressed, framed and checksummed
    /// pieces; that cross is what OOM-killed this box on 2026-08-03 (the
    /// original loop ran `len` up to 1_000_000 against `seg = 1` and took over
    /// 6 minutes even under an address-space cap). The two loops together cover
    /// strictly more shapes than the cross did and finish in under a second.
    /// The library now refuses the absurd end of that cross outright — see
    /// `a_degenerate_segment_size_is_refused_not_attempted`.
    #[test]
    fn degenerate_segment_sizes_round_trip_on_small_inputs() {
        for &len in &[0usize, 1, 2, 3, 7, 8, 4095, 4096, 4097, 8192] {
            for &seg in &[1usize, 2, 3, 7, 4096] {
                round_trips(len, seg);
            }
        }
    }

    /// **Large inputs, against REALISTIC segment sizes.** The size axis, kept
    /// separate from the degenerate-segment axis above for the reason stated
    /// there. Every pair here stays in the low hundreds of pieces.
    #[test]
    fn large_inputs_round_trip_at_realistic_segment_sizes() {
        for &len in &[100_000usize, 1_000_000] {
            for &seg in &[4096usize, 65_536, 1 << 20] {
                round_trips(len, seg);
            }
        }
    }

    /// **The OOM guard, split side.** A `segment` size that would cut the input
    /// into an absurd number of pieces must come back as an `Err` — from the
    /// planner, from the serial path, and from the public `compress_split`
    /// entry point, because all three route through `plan_segments`.
    ///
    /// Seen red on 2026-08-03 with the ceiling check removed from
    /// `plan_segments`: every case below returned `Ok`, building a
    /// million-element plan, and this test failed on the first `expect_err`.
    #[test]
    fn a_degenerate_segment_size_is_refused_not_attempted() {
        use crate::limits::MAX_PIECES;

        // The planner itself — the one writer all three paths share.
        let just_over = MAX_PIECES + 1;
        let e = plan_segments(just_over, 1).expect_err("one segment per byte must be refused");
        assert!(
            matches!(e, DeflateError::TooManyPieces { pieces, max, .. }
                     if pieces == just_over && max == MAX_PIECES),
            "wrong error: {e}"
        );
        assert!(
            e.to_string().contains("split segments"),
            "message does not name the axis: {e}"
        );

        // The shape the old cross-product test ran: 1_000_000 bytes at seg = 1.
        assert!(
            plan_segments(1_000_000, 1).is_err(),
            "a million one-byte segments were planned"
        );
        assert!(
            plan_segments(1_000_000, 0).is_err(),
            "segment=0 escaped the ceiling"
        );
        assert!(
            split_compress_serial(&vec![b'x'; 1_000_000], 6, 1).is_err(),
            "the serial split path attempted a million one-byte segments"
        );
        assert!(
            crate::compress_split(
                &vec![b'x'; 1_000_000],
                &Config {
                    segment: 1,
                    ..Config::default()
                }
            )
            .is_err(),
            "the public compress_split entry point bypassed the ceiling"
        );
        assert!(
            crate::compress_objects(
                &[&vec![b'x'; 1_000_000][..]],
                &Config {
                    split_above: 1,
                    segment: 1,
                    ..Config::default()
                },
            )
            .is_err(),
            "compress_objects bypassed the ceiling"
        );

        // Not vacuous: at the ceiling and at every realistic segment size the
        // planner still says yes.
        assert_eq!(plan_segments(MAX_PIECES, 1).unwrap().len(), MAX_PIECES);
        assert!(
            plan_segments(256 << 30, 1 << 20).is_ok(),
            "256 GiB at 1 MiB segments"
        );
        assert!(split_compress_serial(&vec![b'x'; 1_000_000], 6, 4096).is_ok());
    }

    /// RED-when-broken: the Adler-32 trailer must be *checked* by the decoder.
    /// `flate2`'s `ZlibDecoder` validates it — so corrupting one trailer byte
    /// must make the read above FAIL. Without this, the round-trip test above
    /// would pass even if `adler32_combine` were returning garbage, because the
    /// payload bytes would still be right. This is the guard on the guard.
    #[test]
    fn a_wrong_adler_trailer_is_actually_rejected() {
        let src = corpus(200_000);
        let mut z = split_compress_serial(&src, 6, 16_384).unwrap();
        // Sanity: it reads clean first.
        let mut ok = Vec::new();
        flate2::read::ZlibDecoder::new(&z[..])
            .read_to_end(&mut ok)
            .unwrap();
        assert_eq!(ok, src);

        // Now break the trailer. If the decoder did not verify it, this would
        // still succeed and the test above would be worthless.
        let n = z.len();
        z[n - 1] ^= 0xff;
        let mut bad = Vec::new();
        let r = flate2::read::ZlibDecoder::new(&z[..]).read_to_end(&mut bad);
        assert!(
            r.is_err(),
            "the decoder did NOT verify the Adler-32 trailer — the round-trip \
             tests in this module prove nothing about adler32_combine"
        );
    }

    /// RED-when-broken: if the final segment were emitted with FULL_FLUSH
    /// instead of FINISH, no `BFINAL=1` block would ever be written. Prove a
    /// decoder actually notices — otherwise `compress_segment`'s `is_last`
    /// branch is untested.
    #[test]
    fn a_stream_with_no_final_block_is_rejected() {
        let src = corpus(50_000);
        let mut enc = Encoder::new(6, false).unwrap();
        let plan = plan_segments(src.len(), 8192).unwrap();
        let mut segs = Vec::new();
        for (start, end, _last) in plan {
            // Deliberately pass is_last = false for EVERY segment.
            segs.push(
                compress_segment(&mut enc, &src, start, end, false, Container::Zlib).unwrap(),
            );
        }
        let z = assemble(6, &segs, Container::Zlib);
        let mut got = Vec::new();
        let r = flate2::read::ZlibDecoder::new(&z[..]).read_to_end(&mut got);
        assert!(
            r.is_err() || got != src,
            "a stream whose last block is not BFINAL was accepted as complete — \
             the `is_last` branch of compress_segment is not being tested"
        );
    }

    /// The plan is a pure function of (len, segment): contiguous, covering, in
    /// order, exactly one `is_last`.
    ///
    /// Same rule as the round-trip tests above: degenerate `seg` is paired with
    /// small `len`, big `len` with realistic `seg`. Crossing them asks the
    /// planner for a piece count it now (correctly) refuses, and asked the old
    /// planner for a million-element `Vec`.
    #[test]
    fn segment_plan_is_contiguous_and_covering() {
        let check = |len: usize, seg: usize| {
            let p = plan_segments(len, seg)
                .unwrap_or_else(|e| panic!("len={len} seg={seg}: refused: {e}"));
            assert!(!p.is_empty());
            assert_eq!(p[0].0, 0, "len={len} seg={seg}: must start at 0");
            assert_eq!(
                p.last().unwrap().1,
                len,
                "len={len} seg={seg}: must cover to len"
            );
            assert_eq!(
                p.iter().filter(|(_, _, l)| *l).count(),
                1,
                "len={len} seg={seg}: exactly one final segment"
            );
            for w in p.windows(2) {
                assert_eq!(w[0].1, w[1].0, "len={len} seg={seg}: gap or overlap");
                assert!(
                    !w[0].2,
                    "len={len} seg={seg}: is_last set on a non-final segment"
                );
            }
        };
        for &len in &[0usize, 1, 999, 1000, 1001, 9999] {
            for &seg in &[1usize, 7, 1000] {
                check(len, seg);
            }
        }
        for &len in &[1usize << 20, (1 << 20) + 1, 64 << 20] {
            for &seg in &[1000usize, 1 << 16, 1 << 20] {
                check(len, seg);
            }
        }
    }

    /// Splitting costs ratio. Measure it here so the cost is a documented
    /// number rather than a surprise — and so a future change that silently
    /// stops resetting the window (which would improve ratio but break
    /// segment independence) is visible.
    #[test]
    fn splitting_costs_ratio_and_the_cost_is_bounded() {
        let src = corpus(4 << 20);
        let whole = crate::unit::zlib_compress(&src, 6).unwrap();
        for &seg in &[1 << 20, 256 << 10, 64 << 10] {
            let split = split_compress_serial(&src, 6, seg).unwrap();
            let overhead = split.len() as f64 / whole.len() as f64;
            assert!(
                overhead >= 1.0,
                "seg={seg}: split output ({}) SMALLER than unsplit ({}) — impossible \
                 unless the window is not actually being reset",
                split.len(),
                whole.len()
            );
            assert!(
                overhead < 1.25,
                "seg={seg}: split costs {:.1}% more than unsplit — far beyond the \
                 expected few percent; a boundary is being emitted per block, not per segment",
                (overhead - 1.0) * 100.0
            );
        }
    }
}