haematite 0.7.0

//! COMMIT-COLLAPSE §4 ⟨r6⟩ promise-compaction bound tests.
//!
//! O(dirty) global commit stops sending a `Commit` to a clean shard, which
//! removes the incidental promise compaction the empty commit used to perform. The
//! actor replaces it with a DELIBERATE compaction: after each forced promise append
//! it compacts once promise churn crosses [`PROMISE_COMPACTION_THRESHOLD`], so a
//! churning clean shard's WAL stays bounded at `marker + threshold + 1` frames
//! rather than growing one force-synced frame per election message forever.
//!
//! These drive the compaction through the REAL [`ShardActor::record_promise`] path
//! against an on-disk WAL, then reopen from disk to prove the latest promise (and,
//! for a marker-bearing shard, the committed root) survives the replacement — the
//! `[latest promise]` marker-less shape and the `[marker][latest promise]`
//! marker-bearing shape both recover correctly.

use super::{Ballot, ShardActor};
use crate::store::MemoryStore;
use crate::sync::topology::SyncNodeId;
use crate::wal::{DurableWal, FsyncPolicy, PROMISE_COMPACTION_THRESHOLD, WalError, WalRecovery};
use std::error::Error;
use std::path::{Path, PathBuf};

type BoxError = Box<dyn Error>;

struct TempWal {
    _dir: tempfile::TempDir,
    path: PathBuf,
}

fn temp_wal() -> Result<TempWal, WalError> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("shard.wal");
    Ok(TempWal { _dir: dir, path })
}

fn ballot(counter: u64) -> Ballot {
    Ballot::new(counter, SyncNodeId::from("owner"))
}

/// Reopen the actor from the SAME on-disk WAL (crash-recovery shape). A fresh
/// `MemoryStore` is passed; a marker-bearing WAL verifies its root against the
/// store, so the caller seeds the store with the committing actor's nodes when a
/// committed root is expected.
fn reopen(path: &Path, store: &MemoryStore) -> Result<ShardActor, WalError> {
    let recovered = WalRecovery::recover_path(path, store)?;
    let wal = DurableWal::new(path, FsyncPolicy::CommitOnly)?;
    ShardActor::from_recovered(wal, recovered, store, crate::tree::TreePolicy::V1_DEFAULT)
}

/// Count length-prefixed frames (`[u32 LE len][len bytes]`) in a raw WAL file — the
/// on-disk frame census the bound is stated in (promise frames are skipped by the
/// higher-level `read_file` summary, so this counts them directly).
fn frame_count(path: &Path) -> Result<usize, BoxError> {
    let bytes = std::fs::read(path)?;
    let mut offset = 0_usize;
    let mut frames = 0_usize;
    while offset < bytes.len() {
        let end = offset.checked_add(4).ok_or("frame length overflow")?;
        let len_bytes = bytes.get(offset..end).ok_or("truncated frame length")?;
        let len =
            u32::from_le_bytes([len_bytes[0], len_bytes[1], len_bytes[2], len_bytes[3]]) as usize;
        offset = end.checked_add(len).ok_or("frame body overflow")?;
        frames = frames.saturating_add(1);
    }
    Ok(frames)
}

/// Marker-LESS shard (never committed): threshold + 10 promise appends with no data
/// write and no commit keep the WAL bounded (compaction fired), and the latest
/// promise survives recovery with root `None` — pinning the `[latest promise]`
/// replacement shape.
#[test]
fn promise_churn_on_markerless_shard_stays_bounded_and_recovers() -> Result<(), BoxError> {
    let temp = temp_wal()?;
    let appends = PROMISE_COMPACTION_THRESHOLD + 10;
    {
        let wal = DurableWal::new(&temp.path, FsyncPolicy::CommitOnly)?;
        let mut actor = ShardActor::new(wal);
        for counter in 1..=appends as u64 {
            actor.record_promise(ballot(counter))?;
        }
        drop(actor);
    }

    let frames = frame_count(&temp.path)?;
    assert!(
        frames <= PROMISE_COMPACTION_THRESHOLD + 1,
        "WAL bounded at marker+threshold+1: {frames} frames after {appends} appends"
    );
    assert!(
        frames < appends,
        "compaction must have fired ({frames} frames < {appends} appends)"
    );

    let store = MemoryStore::new();
    let recovered = reopen(&temp.path, &store)?;
    assert_eq!(
        recovered.committed_root(),
        None,
        "a marker-less compacted shard recovers with no committed root"
    );
    assert_eq!(
        recovered.promised(),
        &ballot(appends as u64),
        "the latest promise survives compaction across recovery"
    );
    Ok(())
}

/// Marker-BEARING shard: after a committed root, threshold + 10 promise appends stay
/// bounded and recovery restores BOTH the committed root and the latest promise —
/// pinning the `[marker][latest promise]` replacement shape.
#[test]
fn promise_churn_on_markerbearing_shard_preserves_root_and_promise() -> Result<(), BoxError> {
    let temp = temp_wal()?;
    let appends = PROMISE_COMPACTION_THRESHOLD + 10;
    let mut store = MemoryStore::new();
    let committed_root;
    {
        let wal = DurableWal::new(&temp.path, FsyncPolicy::CommitOnly)?;
        let mut actor = ShardActor::new(wal);
        actor.put_with_ttl(b"k".to_vec(), b"v".to_vec(), None, &store)?;
        committed_root = actor.commit(&mut store)?;
        for counter in 1..=appends as u64 {
            actor.record_promise(ballot(counter))?;
        }
        drop(actor);
    }

    let frames = frame_count(&temp.path)?;
    assert!(
        frames <= PROMISE_COMPACTION_THRESHOLD + 2,
        "WAL bounded (marker + threshold + 1): {frames} frames after {appends} appends"
    );

    let recovered = reopen(&temp.path, &store)?;
    assert_eq!(
        recovered.committed_root(),
        Some(committed_root),
        "the committed root survives marker-bearing promise compaction"
    );
    assert_eq!(
        recovered.promised(),
        &ballot(appends as u64),
        "the latest promise survives marker-bearing compaction across recovery"
    );
    Ok(())
}