zesven 3.0.0

//! A large entry has nothing to be compressed alongside, so the codec splits
//! it into blocks and compresses those across cores instead. Three things have
//! to hold for that to be worth doing, and each is a property rather than a
//! case:
//!
//! - what comes back out is what went in, at every size around a block edge;
//! - the archive does not depend on how many cores produced it, so the same
//!   input is the same file on every machine;
//! - memory follows the window and not the entry, which is the whole reason
//!   this path exists and the one thing a naive implementation gets wrong.

#![cfg(all(feature = "lzma2", feature = "parallel"))]

use std::io::{Cursor, Read, Write};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use zesven::read::Archive;
use zesven::write::{EntryMeta, WriteOptions, Writer};
use zesven::{ArchivePath, Threads};

/// Entries at or above this go through the write-through path.
///
/// Mirrors `write::streaming_entry::STREAMING_THRESHOLD`, which is internal.
/// A test that quietly stopped exercising that path would still pass, so the
/// sizes below are stated relative to this rather than as bare numbers.
const STREAMING_THRESHOLD: usize = 64 * 1024 * 1024;

/// Data that compresses a little, as a video container does.
///
/// Neither incompressible nor repetitive: the first would hide a matcher that
/// silently produced nothing, and the second compresses so far that a block
/// boundary stops being visible in the output at all.
fn payload(len: usize) -> Vec<u8> {
    let mut data = Vec::with_capacity(len);
    let mut state = 0x243F_6A88_85A3_08D3u64;
    while data.len() < len {
        state ^= state << 13;
        state ^= state >> 7;
        state ^= state << 17;
        data.extend_from_slice(&state.to_le_bytes());
        if state % 8 == 0 {
            // Occasional structure, so matches exist to be found.
            data.extend_from_slice(b"........................");
        }
    }
    data.truncate(len);
    data
}

fn write_archive(data: &[u8], threads: Threads) -> Vec<u8> {
    let options = WriteOptions::new()
        .level(1)
        .expect("level")
        .threads(threads);
    let mut writer = Writer::create(Cursor::new(Vec::new()))
        .expect("writer")
        .options(options);
    writer
        .add_bytes(ArchivePath::new("big.bin").expect("path"), data)
        .expect("adds");
    let (_result, cursor) = writer.finish_into_inner().expect("finishes");
    cursor.into_inner()
}

fn read_back(archive: &[u8]) -> Vec<u8> {
    let mut archive = Archive::open(Cursor::new(archive.to_vec())).expect("opens");
    archive.extract_to_vec("big.bin").expect("extracts")
}

/// Every size around a block edge has to survive the trip.
///
/// Blocks start at one dictionary and grow to four, and a block is handed to a
/// worker only once a quarter of the next one has arrived behind it. At level 1
/// the dictionary is 1 MiB, so the first eight blocks account for 12 MiB and
/// every one after them is 4 MiB: 64 MiB is exactly a boundary, and 69 MiB is
/// exactly where the block after it becomes worth dispatching. Off-by-one in
/// the split shows up at those two sizes and nowhere else.
#[test]
fn test_large_entries_round_trip_around_block_edges() {
    let mib = 1024 * 1024;

    for len in [
        STREAMING_THRESHOLD - 1,
        STREAMING_THRESHOLD,
        STREAMING_THRESHOLD + 1,
        STREAMING_THRESHOLD + 4 * mib,
        STREAMING_THRESHOLD + 5 * mib - 1,
        STREAMING_THRESHOLD + 5 * mib,
        STREAMING_THRESHOLD + 5 * mib + 1,
    ] {
        let data = payload(len);
        let archive = write_archive(&data, Threads::count_or_single(4));
        assert_eq!(read_back(&archive), data, "len={len}");
    }
}

/// What an entry claims to be must not reach the bytes of the archive.
///
/// The length in `EntryMeta` is whatever the caller had to hand: a `stat` taken
/// before the file was finished being written, a guess about a pipe, or nothing
/// at all. It decides which entries are worth reporting progress for and
/// nothing else - if it also decided the dictionary, the block size, or which
/// path the entry took, then the same bytes described differently would produce
/// different archives, and a length declared too small would send an entry far
/// larger than memory down the path that holds it whole.
#[test]
fn test_the_declared_size_does_not_reach_the_archive() {
    let data = payload(STREAMING_THRESHOLD + 6 * 1024 * 1024);

    let write = |declared: u64| {
        let options = WriteOptions::new()
            .level(1)
            .expect("level")
            .threads(Threads::count_or_single(4));
        let mut writer = Writer::create(Cursor::new(Vec::new()))
            .expect("writer")
            .options(options);
        writer
            .add_stream(
                ArchivePath::new("big.bin").expect("path"),
                &mut Cursor::new(data.clone()),
                EntryMeta::file(declared),
            )
            .expect("adds");
        let (_result, cursor) = writer.finish_into_inner().expect("finishes");
        cursor.into_inner()
    };

    let truthful = write(data.len() as u64);
    for declared in [0, 1, (STREAMING_THRESHOLD - 1) as u64, 4 << 30] {
        assert_eq!(
            write(declared),
            truthful,
            "a declared size of {declared} changed the archive",
        );
    }
    assert_eq!(read_back(&truthful), data);
}

/// The archive must not depend on how many cores wrote it.
///
/// Block boundaries come from the dictionary alone, so a two-core laptop and a
/// sixty-four-core server have to produce the same file. Without this the
/// worker count would leak into the output and an archive would stop being
/// reproducible off the machine that made it.
#[test]
fn test_output_does_not_depend_on_worker_count() {
    let data = payload(STREAMING_THRESHOLD + 3 * 1024 * 1024);

    let reference = write_archive(&data, Threads::count_or_single(2));
    for threads in [4, 8, 16] {
        assert_eq!(
            write_archive(&data, Threads::count_or_single(threads)),
            reference,
            "{threads} workers changed the archive",
        );
    }
    assert_eq!(read_back(&reference), data);
}

/// A single thread writes one unbroken stream, which still has to read back.
///
/// That output is deliberately different - it is the smallest the level can
/// produce - so the guarantee here is the round trip, not the bytes.
#[test]
fn test_single_thread_round_trips() {
    let data = payload(STREAMING_THRESHOLD + 1024 * 1024);
    let archive = write_archive(&data, Threads::Single);
    assert_eq!(read_back(&archive), data);
}

/// A sink that records how much has reached it.
struct CountingSink {
    inner: Cursor<Vec<u8>>,
    written: Arc<AtomicU64>,
}

impl Write for CountingSink {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let n = self.inner.write(buf)?;
        self.written.fetch_add(n as u64, Ordering::Release);
        Ok(n)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

impl std::io::Seek for CountingSink {
    fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
        self.inner.seek(pos)
    }
}

/// A source that records how much has been taken from it, and notes the
/// largest gap between what it has given out and what has reached the sink.
struct GapReader<'a> {
    data: &'a [u8],
    position: usize,
    written: Arc<AtomicU64>,
    largest_gap: u64,
}

impl Read for GapReader<'_> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let n = buf.len().min(self.data.len() - self.position);
        buf[..n].copy_from_slice(&self.data[self.position..self.position + n]);
        self.position += n;

        let gap = (self.position as u64).saturating_sub(self.written.load(Ordering::Acquire));
        self.largest_gap = self.largest_gap.max(gap);

        Ok(n)
    }
}

/// Data with nothing to find in it, so that what is written tracks what was
/// read.
///
/// The measurement below is read-position minus bytes-written. On data that
/// compresses, most of that difference is the compression itself and grows
/// with the entry whatever the writer holds, which would make the test pass
/// for the wrong reason. Incompressible data makes the difference the
/// writer's own backlog and nothing else.
fn incompressible(len: usize) -> Vec<u8> {
    let mut data = Vec::with_capacity(len);
    let mut state = 0x2545_F491_4F6C_DD1Du64;
    while data.len() < len {
        state ^= state << 13;
        state ^= state >> 7;
        state ^= state << 17;
        data.extend_from_slice(&state.to_le_bytes());
    }
    data.truncate(len);
    data
}

/// Runs one entry through the writer and returns the largest backlog seen.
fn largest_backlog(data: &[u8]) -> u64 {
    let written = Arc::new(AtomicU64::new(0));
    let sink = CountingSink {
        inner: Cursor::new(Vec::new()),
        written: Arc::clone(&written),
    };
    let options = WriteOptions::new()
        .level(1)
        .expect("level")
        .threads(Threads::count_or_single(4));
    let mut writer = Writer::create(sink).expect("writer").options(options);

    let mut source = GapReader {
        data,
        position: 0,
        written: Arc::clone(&written),
        largest_gap: 0,
    };

    writer
        .add_stream(
            ArchivePath::new("big.bin").expect("path"),
            &mut source,
            EntryMeta::file(data.len() as u64),
        )
        .expect("adds");
    let gap = source.largest_gap;
    let _ = writer.finish_into_inner().expect("finishes");
    gap
}

/// Memory has to follow the window, not the size of the entry.
///
/// What the writer is holding is the gap between what it has read and what it
/// has written. The invariant is that this does not grow with the entry:
/// doubling the input must not double the backlog, which is what makes a file
/// larger than memory archivable at all. A multi-threaded encoder without
/// backpressure queues the whole entry, and that is what this catches.
///
/// Stated as a comparison between two sizes rather than as a fraction of one.
/// A fraction is satisfied by anything that buffers a constant share of the
/// input - including buffering all of it - so it would have passed for a
/// writer with no bound at all.
#[test]
fn test_memory_follows_the_window_not_the_entry() {
    let small = incompressible(STREAMING_THRESHOLD + 8 * 1024 * 1024);
    let large = incompressible(STREAMING_THRESHOLD + 136 * 1024 * 1024);

    let small_backlog = largest_backlog(&small);
    let large_backlog = largest_backlog(&large);

    // The entry grew by 128 MiB. The backlog is allowed to differ by a little
    // - it is sampled, and the last blocks of a run land differently - but not
    // to follow the entry.
    let slack = 16 * 1024 * 1024;
    assert!(
        large_backlog <= small_backlog + slack,
        "backlog went from {small_backlog} bytes on a {} byte entry to \
         {large_backlog} on a {} byte one: it is following the entry rather \
         than the window",
        small.len(),
        large.len(),
    );

    // And it really is bounded, not merely growing slowly. The bound is the
    // threshold - which is read before the entry can be recognised as large -
    // plus the blocks in flight behind it. What it is not is a share of the
    // entry: on the larger entry here that would allow four times as much.
    let bound = (STREAMING_THRESHOLD + 64 * 1024 * 1024) as u64;
    for (backlog, data) in [(small_backlog, &small), (large_backlog, &large)] {
        assert!(
            backlog < bound,
            "backlog {backlog} on a {} byte entry, over a bound of {bound}",
            data.len(),
        );
    }
}

/// A wrong declared size must not cost memory either.
///
/// The bound above holds when the caller knows how long the entry is. An entry
/// that claims to be small and is not has to be recognised by reading, or the
/// writer holds a file it was told would fit and does not.
#[test]
fn test_a_size_declared_too_small_still_bounds_memory() {
    let data = incompressible(STREAMING_THRESHOLD + 136 * 1024 * 1024);

    let written = Arc::new(AtomicU64::new(0));
    let sink = CountingSink {
        inner: Cursor::new(Vec::new()),
        written: Arc::clone(&written),
    };
    let options = WriteOptions::new()
        .level(1)
        .expect("level")
        .threads(Threads::count_or_single(4));
    let mut writer = Writer::create(sink).expect("writer").options(options);

    let mut source = GapReader {
        data: &data,
        position: 0,
        written: Arc::clone(&written),
        largest_gap: 0,
    };

    writer
        .add_stream(
            ArchivePath::new("big.bin").expect("path"),
            &mut source,
            // A tenth of the truth.
            EntryMeta::file(data.len() as u64 / 10),
        )
        .expect("adds");
    let backlog = source.largest_gap;
    let _ = writer.finish_into_inner().expect("finishes");

    assert!(
        backlog < (STREAMING_THRESHOLD + 64 * 1024 * 1024) as u64,
        "backlog {backlog} on a {} byte entry declared as {}: the declared size \
         decided how the entry was held",
        data.len(),
        data.len() / 10,
    );
}