zipatch-rs 1.5.0

//! Tests that pin the actual flush and sync_all call counts for each
//! `CheckpointPolicy` variant, including the mid-block emission contract.
//!
//! The earlier `checkpoint_emission.rs` cadence test only inspects the
//! *ordering* of records as seen by the sink; it cannot observe whether
//! flush/fsync actually fired because both `record_checkpoint` and
//! `record_checkpoint_mid_block` call `sink.record(...)` the same way. These
//! tests use the `#[cfg(test)]` counters on `ApplyContext` directly.

mod common;

use std::io::Cursor;
use std::sync::{Arc, Mutex};

use common::{FileBlock, adir_body, sqpk_addfile_body, wrap_patch};
use zipatch_rs::test_utils::make_chunk;
use zipatch_rs::{ApplyContext, Checkpoint, CheckpointPolicy, CheckpointSink, ZiPatchReader};

#[derive(Clone)]
struct PolicySink(CheckpointPolicy);

impl CheckpointSink for PolicySink {
    fn record(&mut self, _: &Checkpoint) -> std::io::Result<()> {
        Ok(())
    }
    fn policy(&self) -> CheckpointPolicy {
        self.0
    }
}

/// Build [ADIR, SqpkFile(AddFile with N blocks), ADIR].
fn build_patch_with_n_blocks(n: usize) -> Vec<u8> {
    let blocks: Vec<FileBlock> = (0..n)
        .map(|i| FileBlock {
            is_compressed: false,
            decompressed: vec![0x10 + i as u8; 64],
        })
        .collect();
    wrap_patch(vec![
        make_chunk(b"ADIR", &adir_body("alpha")),
        make_chunk(b"SQPK", &sqpk_addfile_body("data/target.dat", 0, &blocks)),
        make_chunk(b"ADIR", &adir_body("bravo")),
    ])
}

// -------- Flush policy: every chunk-boundary record flushes; mid-block never --------

#[test]
fn flush_policy_flushes_on_every_boundary_not_on_mid_block() {
    // Catches: flush() called on mid-block records, or NOT called on boundary records.
    //
    // Patch: [ADIR, AddFile(2 blocks), ADIR] → 3 boundary records + 2 mid-block.
    // Under Flush: flush on every boundary, never on mid-block.
    // Expected flush count = 3 (one per boundary) + 1 (post-loop flush from apply_to).
    // Expected sync count = 0.
    let patch = build_patch_with_n_blocks(2);
    let tmp = tempfile::tempdir().unwrap();
    let mut ctx =
        ApplyContext::new(tmp.path()).with_checkpoint_sink(PolicySink(CheckpointPolicy::Flush));
    ZiPatchReader::new(Cursor::new(patch))
        .unwrap()
        .apply_to(&mut ctx)
        .unwrap();

    // 3 boundary records each call flush(); plus 1 from the post-loop apply_to flush.
    assert_eq!(
        ctx.test_flush_count, 4,
        "flush called on wrong count (3 boundary + 1 post-loop)"
    );
    assert_eq!(
        ctx.test_sync_count, 0,
        "sync_all must not fire under Flush policy"
    );
}

// -------- Fsync policy: every boundary record syncs; mid-block never --------

#[test]
fn fsync_policy_syncs_on_every_boundary_not_on_mid_block() {
    // Catches: sync_all() called on mid-block records, or NOT called on boundary records.
    //
    // Patch: [ADIR, AddFile(2 blocks), ADIR] → 3 boundary records + 2 mid-block.
    // Under Fsync: sync_all on every boundary, never on mid-block.
    // Expected sync count = 3.
    let patch = build_patch_with_n_blocks(2);
    let tmp = tempfile::tempdir().unwrap();
    let mut ctx =
        ApplyContext::new(tmp.path()).with_checkpoint_sink(PolicySink(CheckpointPolicy::Fsync));
    ZiPatchReader::new(Cursor::new(patch))
        .unwrap()
        .apply_to(&mut ctx)
        .unwrap();

    assert_eq!(
        ctx.test_sync_count, 3,
        "sync_all must fire once per boundary record (3 boundaries)"
    );
    // Mid-block records (2) must not contribute to the sync count.
}

// -------- FsyncEveryN: boundary records count toward cadence; mid-block excluded --------

#[test]
fn fsync_every_n_counts_only_boundary_records() {
    // Catches: mid-block records bumping the cadence counter, causing early syncs.
    //
    // Patch: [ADIR, AddFile(4 blocks), ADIR] → 3 boundary + 4 mid-block.
    // Under FsyncEveryN(3): cadence fires on the 3rd boundary record only.
    // Expected sync count = 1.  Mid-block records must not move the counter.
    let patch = build_patch_with_n_blocks(4);
    let tmp = tempfile::tempdir().unwrap();
    let mut ctx = ApplyContext::new(tmp.path())
        .with_checkpoint_sink(PolicySink(CheckpointPolicy::FsyncEveryN(3)));
    ZiPatchReader::new(Cursor::new(patch))
        .unwrap()
        .apply_to(&mut ctx)
        .unwrap();

    assert_eq!(
        ctx.test_sync_count, 1,
        "FsyncEveryN(3) with 3 boundary records must fire exactly 1 sync_all"
    );
}

#[test]
fn fsync_every_2_fires_once_per_two_boundary_records() {
    // Catches: off-by-one in the `checkpoints_since_fsync >= n` check.
    //
    // Patch: [ADIR, AddFile(2 blocks), ADIR] → 3 boundary records.
    // Under FsyncEveryN(2): fires at boundary 2 (count=2 ≥ 2), then counter
    // resets, boundary 3 does not trigger another sync.
    // Expected sync count = 1.
    let patch = build_patch_with_n_blocks(2);
    let tmp = tempfile::tempdir().unwrap();
    let mut ctx = ApplyContext::new(tmp.path())
        .with_checkpoint_sink(PolicySink(CheckpointPolicy::FsyncEveryN(2)));
    ZiPatchReader::new(Cursor::new(patch))
        .unwrap()
        .apply_to(&mut ctx)
        .unwrap();

    assert_eq!(
        ctx.test_sync_count, 1,
        "FsyncEveryN(2) with 3 boundary records must fire 1 sync_all"
    );
}

#[test]
fn fsync_every_1_fires_on_every_boundary_record() {
    // Catches: FsyncEveryN(1) not behaving like per-record Fsync.
    //
    // Patch: [ADIR, AddFile(2 blocks), ADIR] → 3 boundary records.
    // Under FsyncEveryN(1): fires at every boundary record.
    // Expected sync count = 3.
    let patch = build_patch_with_n_blocks(2);
    let tmp = tempfile::tempdir().unwrap();
    let mut ctx = ApplyContext::new(tmp.path())
        .with_checkpoint_sink(PolicySink(CheckpointPolicy::FsyncEveryN(1)));
    ZiPatchReader::new(Cursor::new(patch))
        .unwrap()
        .apply_to(&mut ctx)
        .unwrap();

    assert_eq!(
        ctx.test_sync_count, 3,
        "FsyncEveryN(1) must sync on every boundary record (same as Fsync)"
    );
}

// -------- No sink installed: flush still called for post-loop durability --------

#[test]
fn no_sink_installed_does_not_panic_and_apply_succeeds() {
    // Catches: NoopCheckpointSink causing a panic or unexpected error.
    let patch = build_patch_with_n_blocks(2);
    let tmp = tempfile::tempdir().unwrap();
    let mut ctx = ApplyContext::new(tmp.path()); // default noop sink
    ZiPatchReader::new(Cursor::new(patch))
        .unwrap()
        .apply_to(&mut ctx)
        .unwrap();
    assert!(tmp.path().join("alpha").is_dir());
    assert!(tmp.path().join("bravo").is_dir());
}

// -------- Closure sink baseline: mid-block records visible; boundary records visible --------

#[test]
fn closure_sink_receives_both_mid_block_and_boundary_records() {
    // Catches: records dropped before reaching the sink.
    let patch = build_patch_with_n_blocks(3);
    let records: Arc<Mutex<Vec<Checkpoint>>> = Arc::new(Mutex::new(Vec::new()));
    let records2 = records.clone();

    let tmp = tempfile::tempdir().unwrap();
    let mut ctx = ApplyContext::new(tmp.path()).with_checkpoint_sink(
        move |c: &Checkpoint| -> std::io::Result<()> {
            records2.lock().unwrap().push(c.clone());
            Ok(())
        },
    );
    ZiPatchReader::new(Cursor::new(patch))
        .unwrap()
        .apply_to(&mut ctx)
        .unwrap();

    let got = records.lock().unwrap();
    // [ADIR, AddFile(3 blocks), ADIR] → 3 boundary + 3 mid-block = 6 records.
    assert_eq!(got.len(), 6, "expected 3 boundary + 3 mid-block records");
    let mid_count = got
        .iter()
        .filter(|c| matches!(c, Checkpoint::Sequential(s) if s.in_flight.is_some()))
        .count();
    assert_eq!(mid_count, 3, "exactly 3 mid-block records");
}