haematite 0.7.0

//! COMMIT-COLLAPSE §11 (deferred item 2) — TTL-sweep dirtiness, standalone.
//!
//! The §2.1 transition table has a row `TTL sweep bare delete → dirty` whose
//! defining property is that NO public write is in sight: a sweep that buffers a
//! removal must itself dirty the shard via the commit-state cell, so the next
//! global commit persists the removal. This pins that row DIRECTLY through the
//! REAL native command loop (`ShardNativeHandler`, the same body production runs)
//! driven on beamr's cooperative scheduler — deterministic, single-threaded, no
//! sleeps: the dirtying event is the `DeleteIfExpired` sweep alone, issued after
//! the shard is already clean.
//!
//! Determinism note: the key is stored with a `Duration::ZERO` TTL (the sweep's
//! established already-expired shape, per `expiry_mutation_tests`), so the
//! monotonic clock has already passed its deadline by the time the sweep runs —
//! no wall-clock wait is used as synchronisation.

use std::collections::VecDeque;
use std::error::Error;
use std::sync::mpsc::{self, Receiver, TryRecvError};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use beamr::atom::{Atom, AtomTable};
use beamr::ets::OwnedTerm;
use beamr::module::ModuleRegistry;
use beamr::native::BifRegistryImpl;
use beamr::scheduler::WasmScheduler;
use beamr::term::Term;

use super::handle::{CommandQueue, ShardCommand, ShardCommandKind, ShardError};
use super::native::{self, ShardNativeHandler};
use crate::db::root_advance::RootAdvanceSeam;
use crate::shard::commit_state::CommitSnapshot;

type TestResult = Result<(), Box<dyn Error>>;

const MAX_TURNS_PER_REPLY: usize = 16;

fn cooperative_scheduler() -> WasmScheduler {
    let atom_table = Arc::new(AtomTable::with_common_atoms());
    let modules = Arc::new(ModuleRegistry::new());
    let bifs = Arc::new(BifRegistryImpl::new());
    WasmScheduler::new(atom_table, modules, bifs)
}

fn wake() -> OwnedTerm {
    OwnedTerm::immediate(Term::atom(Atom::OK))
}

/// The §5 classification predicate over a detached cell snapshot.
fn is_dirty(snapshot: &CommitSnapshot) -> bool {
    snapshot.recovering
        || snapshot.unreconciled
        || snapshot.committed_gen != snapshot.dirty_gen
        || snapshot.root.is_none()
}

/// Push one command, wake the shard, and pump cooperative turns until its reply
/// lands (the cooperative analogue of `ShardHandle::enqueue` + blocking `recv`).
fn enqueue_wake_and_pump<T>(
    scheduler: &mut WasmScheduler,
    pid: u64,
    commands: &CommandQueue,
    command: ShardCommand,
    reply_rx: &Receiver<T>,
) -> Result<T, Box<dyn Error>> {
    native::lock_queue(commands).push_back(command);
    scheduler.send_owned(pid, &wake())?;
    for _ in 0..MAX_TURNS_PER_REPLY {
        let _exited = scheduler.run_native_until_idle();
        match reply_rx.try_recv() {
            Ok(value) => return Ok(value),
            Err(TryRecvError::Empty) => {}
            Err(TryRecvError::Disconnected) => {
                return Err("shard reply channel disconnected before a reply".into());
            }
        }
    }
    Err("shard produced no reply within the cooperative turn budget".into())
}

/// A TTL-expiry sweep that buffers a removal dirties the shard through the cell,
/// with no public write in sight at the dirtying step. The put + commit are setup
/// (they leave the shard CLEAN); the `DeleteIfExpired` sweep alone re-dirties it.
#[test]
fn ttl_sweep_removal_dirties_the_shard_via_the_cell() -> TestResult {
    let mut scheduler = cooperative_scheduler();
    let dir = tempfile::tempdir()?;
    let store_dir = dir.path().join("store");
    let wal_path = dir.path().join("shard.wal");

    let commands: CommandQueue = Arc::new(Mutex::new(VecDeque::new()));
    let seam = RootAdvanceSeam::new();
    let cell = seam.commit_state(0);
    let factory = ShardNativeHandler::make_factory(
        store_dir,
        wal_path,
        Arc::clone(&commands),
        crate::tree::TreePolicy::V1_DEFAULT,
        seam.commit_state(0),
    );
    let pid = scheduler.spawn_native_root(factory);
    let _exited = scheduler.run_native_until_idle(); // boot turn

    // Setup: store a key whose TTL is already elapsed (Duration::ZERO), then commit
    // so the shard is CLEAN and the key is persisted with its expiry stamp.
    let (put_tx, put_rx) = mpsc::sync_channel(1);
    let put: Result<(), ShardError> = enqueue_wake_and_pump(
        &mut scheduler,
        pid,
        &commands,
        ShardCommand {
            id: 1,
            kind: ShardCommandKind::Put {
                key: b"ephemeral".to_vec(),
                value: b"v".to_vec(),
                ttl: Some(Duration::ZERO),
                reply: put_tx,
            },
        },
        &put_rx,
    )?;
    put?;

    let (commit_tx, commit_rx) = mpsc::sync_channel(1);
    let committed: Result<(crate::tree::Hash, _), ShardError> = enqueue_wake_and_pump(
        &mut scheduler,
        pid,
        &commands,
        ShardCommand {
            id: 2,
            kind: ShardCommandKind::Commit { reply: commit_tx },
        },
        &commit_rx,
    )?;
    committed?;
    assert!(
        !is_dirty(&cell.snapshot()),
        "the shard is CLEAN after committing the put (setup precondition)"
    );

    // The sweep: DeleteIfExpired buffers a tombstone for the expired key — no public
    // write — and must re-dirty the shard through the cell.
    let (sweep_tx, sweep_rx) = mpsc::sync_channel(1);
    let removed: Result<bool, ShardError> = enqueue_wake_and_pump(
        &mut scheduler,
        pid,
        &commands,
        ShardCommand {
            id: 3,
            kind: ShardCommandKind::DeleteIfExpired {
                key: b"ephemeral".to_vec(),
                reply: sweep_tx,
            },
        },
        &sweep_rx,
    )?;
    assert!(
        removed?,
        "the expired key is swept (buffers a removal tombstone)"
    );
    assert!(
        is_dirty(&cell.snapshot()),
        "a TTL sweep that buffers a removal marks the shard DIRTY via the cell (§2.1)"
    );

    // And the next commit persists the removal, returning the shard to clean.
    let (commit2_tx, commit2_rx) = mpsc::sync_channel(1);
    let recommitted: Result<(crate::tree::Hash, _), ShardError> = enqueue_wake_and_pump(
        &mut scheduler,
        pid,
        &commands,
        ShardCommand {
            id: 4,
            kind: ShardCommandKind::Commit { reply: commit2_tx },
        },
        &commit2_rx,
    )?;
    recommitted?;
    assert!(
        !is_dirty(&cell.snapshot()),
        "committing the swept removal returns the shard to clean"
    );
    Ok(())
}