haematite 0.4.0

// API-003: sweep actor tests.

use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};

use beamr::module::ModuleRegistry;
use beamr::scheduler::{Scheduler, SchedulerConfig};

use super::recover::{collect_tree, recover_view};
use super::{SweepError, SweepHandle, SweepStats};
use crate::shard::actor::{ShardActor, ShardHandle};
use crate::store::DiskStore;
use crate::tree::{LeafNode, Node};
use crate::wal::{DurableWal, FsyncPolicy, Mutation};

const TIMEOUT: Duration = Duration::from_secs(5);

#[test]
fn sweep_stats_default_is_zero() {
    assert_eq!(SweepStats::default().scanned, 0);
    assert_eq!(SweepStats::default().expired, 0);
    assert_eq!(SweepStats::default().deleted, 0);
}

/// Reconstruct the shard's PHYSICAL state from its on-disk store + WAL, exactly
/// as `run_sweep`'s `recover_view` does, and report whether `key` physically
/// resolves to a live entry.
///
/// Crucially this performs NO TTL evaluation: it does not call
/// `is_expired_at`/`visible_value`. It answers only "are these bytes still in
/// the merged store+WAL?", so it is a pure physical-presence probe that
/// read-time expiry filtering can never satisfy.
fn physically_present(store_dir: &Path, wal_path: &Path, key: &[u8]) -> Result<bool, SweepError> {
    let (store, root, buffer) = recover_view(store_dir, wal_path)?;
    let mut merged = BTreeMap::new();
    if let Some(root) = root {
        collect_tree(&store, root, &mut merged)?;
    }
    for mutation in &buffer {
        match mutation {
            Mutation::Put { key, value } => {
                merged.insert(key.clone(), value.clone());
            }
            Mutation::Delete { key } => {
                merged.remove(key);
            }
        }
    }
    Ok(merged.contains_key(key))
}

/// FALSIFIABLE: the periodic self-tick must PHYSICALLY remove an expired entry
/// from the shard's store/WAL, NOT merely read-filter it.
///
/// `physically_present` reconstructs the raw merged store+WAL with no TTL
/// evaluation at all, so this assertion cannot be satisfied by read-time
/// filtering (which leaves the bytes in place and only hides them on read). The
/// entry physically disappears ONLY when a real sweep pass issues a
/// `shard.delete`, writing a `Delete` mutation that shadows the entry.
///
/// What makes it falsifiable: the test NEVER calls `sweep_once`. The only thing
/// that can run a sweep here is the actor's own delayed self-tick armed by
/// `schedule_next_tick`. Revert the wiring — drop the first-slice arm, or turn
/// `ctx.schedule` back into the old immediate `ctx.send`, or stop re-arming
/// after each tick — and no `tick` atom is ever delivered on the interval,
/// `run_sweep` never runs, the entry stays physically present, and the bounded
/// wait below expires with `physically_present` still true: the final assertion
/// fails. (Confirmed: removing the wiring makes this fail.)
///
/// A multi-thread scheduler is required: `run_sweep` blocks waiting for the
/// shard actor's `delete` reply, so the shard must run on a sibling thread.
#[test]
fn periodic_tick_physically_deletes_expired_entry() -> Result<(), Box<dyn std::error::Error>> {
    let scheduler = Arc::new(Scheduler::new(
        SchedulerConfig::default(),
        Arc::new(ModuleRegistry::new()),
    )?);

    let dir = tempfile::tempdir()?;
    let store_dir = dir.path().join("sweep.store");
    let wal_path = dir.path().join("sweep.wal");
    // Seed an empty committed root so the store dir exists with a tree.
    let mut store = DiskStore::new(&store_dir)?;
    let _root = store.put(&Node::Leaf(LeafNode::new(Vec::new())?))?;
    drop(store);

    let shard = ShardHandle::spawn(Arc::clone(&scheduler), store_dir.clone(), wal_path.clone())?;

    // Write an entry with a short, non-zero TTL so it is live (and therefore
    // physically present and NOT yet a sweep target) at the instant we record
    // the precondition, then expires shortly after.
    let key = b"ttl-victim".to_vec();
    shard.put_with_ttl(
        key.clone(),
        b"doomed".to_vec(),
        Some(Duration::from_millis(100)),
        TIMEOUT,
    )?;

    // Precondition: the entry is PHYSICALLY present in the raw store+WAL.
    assert!(
        physically_present(&store_dir, &wal_path, &key)?,
        "entry must be physically present before any sweep runs"
    );

    // Small explicit interval (no implicit default — R4) so the periodic tick
    // fires quickly. Spawned AFTER the precondition check so the sweep cannot
    // race ahead of it.
    let interval = Duration::from_millis(20);
    let sweep = SweepHandle::spawn(
        Arc::clone(&scheduler),
        store_dir.clone(),
        wal_path.clone(),
        shard,
        interval,
        TIMEOUT,
    )?;

    // Drive the scheduler by waiting (bounded) for the self-scheduled tick to
    // fire and physically delete the entry. No manual sweep is triggered.
    let deadline = Instant::now() + Duration::from_secs(10);
    while physically_present(&store_dir, &wal_path, &key)? && Instant::now() < deadline {
        std::thread::sleep(Duration::from_millis(20));
    }

    assert!(
        !physically_present(&store_dir, &wal_path, &key)?,
        "periodic sweep must physically remove the expired entry from the store/WAL"
    );

    assert!(sweep.shutdown(TIMEOUT).is_ok());
    scheduler.shutdown();
    Ok(())
}

// ---------------------------------------------------------------------------
// Sweep crash-recovery: a crash before a sweep delete is committed must leave NO
// TTL entry half-deleted. `recover_view` (the FIRST thing `run_sweep` does) must
// reconstruct a consistent durable view — every key is fully present or fully
// absent, never torn.
// ---------------------------------------------------------------------------

/// Reconstruct the merged physical state (committed tree + uncommitted WAL
/// buffer) exactly as `run_sweep` does, returning the live `(key -> value)` map.
fn merged_view(
    store_dir: &Path,
    wal_path: &Path,
) -> Result<BTreeMap<Vec<u8>, Vec<u8>>, Box<dyn std::error::Error>> {
    let (store, root, buffer) = recover_view(store_dir, wal_path)?;
    let mut merged = BTreeMap::new();
    if let Some(root) = root {
        collect_tree(&store, root, &mut merged)?;
    }
    for mutation in &buffer {
        match mutation {
            Mutation::Put { key, value } => {
                merged.insert(key.clone(), value.clone());
            }
            Mutation::Delete { key } => {
                merged.remove(key);
            }
        }
    }
    Ok(merged)
}

/// CRASH BEFORE ANY SWEEP DELETE: durably commit a live entry and an expired
/// entry, then "crash" (drop the actor) WITHOUT running a sweep. `recover_view`
/// must reconstruct BOTH entries intact — a crash before the sweep commits leaves
/// nothing half-deleted, so the next sweep starts from the complete durable set.
#[test]
fn recover_view_after_crash_before_sweep_is_intact() -> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let store_dir = dir.path().join("sweep.store");
    let wal_path = dir.path().join("sweep.wal");

    // Seed durable committed state: one live entry and one already-expired entry.
    let mut store = DiskStore::new(&store_dir)?;
    let wal = DurableWal::new(&wal_path, FsyncPolicy::CommitOnly)?;
    let mut actor = ShardActor::new(wal);
    actor.put_with_ttl(b"live".to_vec(), b"keep".to_vec(), None)?;
    // A 1ns TTL is already expired by the time the sweep runs.
    actor.put_with_ttl(
        b"doomed".to_vec(),
        b"gone".to_vec(),
        Some(Duration::from_nanos(1)),
    )?;
    let _root = actor.commit(&mut store)?;
    drop(actor); // crash: close the WAL with NO sweep delete issued.

    // recover_view (the start of run_sweep) reconstructs BOTH keys physically —
    // the expired one is NOT half-deleted just because a sweep was pending.
    let merged = merged_view(&store_dir, &wal_path)?;
    assert!(
        merged.contains_key(b"live".as_slice()),
        "the live entry must survive the crash"
    );
    assert!(
        merged.contains_key(b"doomed".as_slice()),
        "an expired entry not yet swept must be fully present after a crash, never torn"
    );
    Ok(())
}

/// CRASH AFTER A SWEEP DELETE WAS STAGED BUT BEFORE THE NEXT COMMIT: the sweep's
/// `delete_if_expired` appends a bare-removal to the WAL (durable) and buffers it,
/// but the physical tree commit has not happened. On recovery the WAL Delete is
/// replayed over the committed root, so the swept key is consistently ABSENT and
/// the live key is fully intact — never a torn half-state.
#[test]
fn recover_view_after_staged_sweep_delete_is_consistent() -> Result<(), Box<dyn std::error::Error>>
{
    let dir = tempfile::tempdir()?;
    let store_dir = dir.path().join("sweep.store");
    let wal_path = dir.path().join("sweep.wal");

    let mut store = DiskStore::new(&store_dir)?;
    let wal = DurableWal::new(&wal_path, FsyncPolicy::CommitOnly)?;
    let mut actor = ShardActor::new(wal);
    actor.put_with_ttl(b"live".to_vec(), b"keep".to_vec(), None)?;
    actor.put_with_ttl(
        b"doomed".to_vec(),
        b"gone".to_vec(),
        Some(Duration::from_nanos(1)),
    )?;
    actor.commit(&mut store)?;

    // The sweep stages the expired delete (WAL append + buffer) but the process
    // crashes BEFORE the next commit folds it into the tree.
    let removed = actor.delete_if_expired(b"doomed", &store)?;
    assert!(removed, "the expired key must be a sweep candidate");
    drop(actor); // crash with the delete staged in the WAL, not yet committed.

    // Recovery replays the staged WAL Delete: the swept key is consistently gone,
    // the live key is fully present. No entry is half-deleted.
    let merged = merged_view(&store_dir, &wal_path)?;
    assert!(
        merged.contains_key(b"live".as_slice()),
        "the live entry must remain fully intact"
    );
    assert!(
        !merged.contains_key(b"doomed".as_slice()),
        "a staged sweep delete must recover as fully absent, never torn"
    );
    Ok(())
}