ldeflate 0.1.1

//! **The 31+1 split** — N−1 workers compress, one dedicated thread streams the
//! result down to the device.
//!
//! [`crate::plan::compress_objects`] materialises every output stream in RAM and
//! hands back a `Vec<Vec<u8>>`. For a repack of a large repository that is the
//! wrong shape: peak RSS becomes the size of the whole compressed pack, and the
//! device sits idle for the entire compress phase and then has to absorb
//! everything at once.
//!
//! This module is the streaming entry point. It uses
//! `gatling::gatling::ordered::run_ordered_sink`, whose **collector runs on the
//! calling thread** — so the caller's thread *is* the one dedicated writer, and
//! `n_workers` compressors run beside it. Ask for `cores - 1` workers and the
//! topology is literally 31 compute + 1 I/O on this box.
//!
//! ```text
//!   producer (lazy, back-pressured)
//!        │  ≤ cap objects alive at any instant
//!        ▼
//!   [31 workers]  compress  ──►  reorder buffer  ──►  writer (calling thread)
//!    self-dispatch, no barrier      bounded by cap        one uncontended pwrite
//! ```
//!
//! ## Why one writer and not 32
//!
//! Because on local NVMe a write is a memcpy-class syscall with no round-trip
//! latency to hide, one uncontended writer already saturates the device, and
//! adding writers buys lock, seek and allocator contention. That is not a guess
//! — it is skade's documented rationale for keeping its local path sequential
//! (`skade/.nornir/s3-ingest-concurrency.md`, "Why local stays sequential (and
//! must)"). The inverse holds for object storage, where the cost is per-op
//! latency and only concurrency hides it; that case wants `gatling::io`, not
//! this.

use std::io::Write;

use crate::plan::Config;
use crate::unit::{DeflateError, EncoderPool};

/// Where one object landed in the output stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Placed {
    /// Byte offset of this object's zlib stream in the output.
    pub offset: u64,
    /// Length of the zlib stream in bytes.
    pub len: u64,
}

/// Default worker count for the streaming path: **one fewer than the box has**,
/// because the calling thread is the writer and wants a core to itself.
pub fn compute_workers() -> usize {
    std::thread::available_parallelism()
        .map(|n| n.get())
        .unwrap_or(2)
        .saturating_sub(1)
        .max(1)
}

/// Compress `objects` and stream every result to `out` in input order, with
/// N−1 compressors running beside the single writer.
///
/// Returns one [`Placed`] per object, so the caller can build a pack index
/// without a second pass.
///
/// `cfg.workers == 0` resolves to [`compute_workers`] (i.e. cores − 1), not to
/// "one per core" — the difference is the whole point of this entry point.
///
/// Note this path compresses each object as **one whole stream**; it does not
/// split large objects across workers, because the ordered sink emits per item.
/// For a batch job dominated by a few huge blobs use
/// [`crate::plan::compress_objects`], which does split. For a pack of thousands
/// of small-to-medium objects — the git case — this is the right shape.
pub fn compress_objects_to_writer<W: Write>(
    objects: &[&[u8]],
    cfg: &Config,
    out: &mut W,
) -> Result<Vec<Placed>, DeflateError> {
    let workers = if cfg.workers == 0 {
        compute_workers()
    } else {
        cfg.workers
    };
    let cap = (workers * 4).max(8);
    let level = cfg.level;
    let wb = cfg.window_bits;
    let ct = cfg.container;

    let mut next = 0usize;
    let producer = move || -> Option<(usize, &[u8])> {
        if next < objects.len() {
            let i = next;
            next += 1;
            // Borrowed, never cloned: the worker compresses the caller's bytes.
            Some((i, objects[i]))
        } else {
            None
        }
    };

    // Lazy: the first checkout on each worker constructs, every later one is a
    // `reset`. Shared with `plan.rs` rather than twinned (LAW 5).
    let pool = EncoderPool::new(level, ct.encoder_is_zlib_wrapped(), wb);

    let mut placed: Vec<Placed> = Vec::with_capacity(objects.len());
    let mut offset: u64 = 0;
    let mut sink_err: Option<DeflateError> = None;

    // The collector — and therefore this closure — runs on THIS thread. It is
    // the one dedicated I/O core.
    let mut sink = |_seq: u64, r: Result<Vec<u8>, DeflateError>| -> anyhow::Result<()> {
        match r {
            Ok(bytes) => {
                out.write_all(&bytes)
                    .map_err(|e| anyhow::anyhow!("ldeflate: write failed: {e}"))?;
                placed.push(Placed {
                    offset,
                    len: bytes.len() as u64,
                });
                offset += bytes.len() as u64;
                Ok(())
            }
            Err(e) => {
                sink_err.get_or_insert(e.clone());
                Err(anyhow::anyhow!("{e}"))
            }
        }
    };

    let run = gatling::gatling::ordered::run_ordered_sink(
        producer,
        workers,
        cap,
        |_label: usize, input: &[u8]| -> Result<Vec<u8>, DeflateError> {
            // ONE encoder per WORKER for the whole call, not one per object.
            // This path called `Encoder::new_with_window` inside the map closure,
            // i.e. `Compress::new` (**4082 ns**) for every object — against a
            // total per-object cost of about 5 us on a git-median 1130 B object.
            // See [`EncoderPool`] for why a pool and not a `thread_local!`.
            let mut enc = pool.checkout()?;
            let mut o = Vec::new();
            enc.compress_into(input, &mut o)?;
            if ct.encoder_is_zlib_wrapped() {
                Ok(o)
            } else {
                Ok(ct.wrap(level, &o, input))
            }
        },
        &mut sink,
    );

    if let Some(e) = sink_err {
        return Err(e);
    }
    run.map_err(|_| DeflateError::Encode("ordered sink aborted"))?;

    gatling::functional_status(
        "ldeflate",
        "compress_objects_to_writer",
        true,
        &"N-1 compressors + 1 dedicated writer on the calling thread",
    );
    Ok(placed)
}

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

    fn objs(n: usize) -> Vec<Vec<u8>> {
        (0..n)
            .map(|i| {
                let mut v = Vec::new();
                let tag = format!("obj{i}:");
                // Wildly uneven sizes, so a barrier or a static carve-up would
                // show up as a slow tail.
                let target = if i % 97 == 0 {
                    400_000
                } else {
                    40 + (i * 7) % 3_000
                };
                while v.len() < target {
                    v.extend_from_slice(tag.as_bytes());
                    v.extend_from_slice(b"the quick brown fox ");
                }
                v.truncate(target);
                v
            })
            .collect()
    }

    /// The streamed bytes must be the exact concatenation of the per-object
    /// streams, **in input order**, with correct offsets — at every worker
    /// count. An out-of-order emit corrupts a pack silently, so this is checked
    /// by decoding each recorded range back to its object rather than by
    /// comparing a hash of the whole file.
    #[test]
    fn streamed_output_is_ordered_and_offsets_are_right() {
        let data = objs(600);
        let r: Vec<&[u8]> = data.iter().map(|v| v.as_slice()).collect();
        for w in [1usize, 2, 7, 31] {
            let mut buf: Vec<u8> = Vec::new();
            let placed =
                compress_objects_to_writer(&r, &Config::default().workers(w), &mut buf).unwrap();
            assert_eq!(placed.len(), data.len(), "workers={w}");
            assert_eq!(
                placed.last().unwrap().offset + placed.last().unwrap().len,
                buf.len() as u64,
                "workers={w}: recorded extents must cover the whole output"
            );
            for (i, p) in placed.iter().enumerate() {
                let slice = &buf[p.offset as usize..(p.offset + p.len) as usize];
                let mut got = Vec::new();
                flate2::read::ZlibDecoder::new(slice)
                    .read_to_end(&mut got)
                    .unwrap_or_else(|e| panic!("workers={w} obj {i}: {e}"));
                assert_eq!(
                    got, data[i],
                    "workers={w}: object {i} landed at the wrong offset"
                );
            }
        }
    }

    /// RED-when-broken: the ordering assertion above must be able to FAIL.
    /// Shuffle the recorded extents and prove the check trips — otherwise
    /// "decoded to the right object" would be true for any permutation.
    #[test]
    fn the_ordering_check_can_actually_fail() {
        let data = objs(50);
        let r: Vec<&[u8]> = data.iter().map(|v| v.as_slice()).collect();
        let mut buf: Vec<u8> = Vec::new();
        let mut placed = compress_objects_to_writer(&r, &Config::default(), &mut buf).unwrap();
        // Swap two extents of DIFFERENT objects; the decode-and-compare must now
        // disagree for at least one index.
        placed.swap(3, 17);
        let mut mismatches = 0;
        for (i, p) in placed.iter().enumerate() {
            let slice = &buf[p.offset as usize..(p.offset + p.len) as usize];
            let mut got = Vec::new();
            if flate2::read::ZlibDecoder::new(slice)
                .read_to_end(&mut got)
                .is_err()
                || got != data[i]
            {
                mismatches += 1;
            }
        }
        assert!(
            mismatches > 0,
            "a deliberately shuffled extent table still passed the per-object decode \
             check — that check cannot detect mis-ordering and proves nothing"
        );
    }

    /// The 31+1 default must actually leave a core for the writer.
    #[test]
    fn default_worker_count_reserves_one_core() {
        let cores = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(2);
        let w = compute_workers();
        assert_eq!(
            w,
            (cores - 1).max(1),
            "expected cores-1 workers, got {w} on {cores} cores"
        );
        assert!(
            w < cores || cores == 1,
            "the writer must not be sharing a core with a compressor"
        );
    }
}