haematite-cli 0.21.1

//! STOR-003 HR4: the release thread, driven through the service as a
//! client drives it. Every wait is on the thread's own signal.

use std::collections::HashSet;
use std::error::Error;
use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
use std::ops::Range;
use std::path::{Path, PathBuf};
use std::sync::mpsc;

use haematite::SyncCounts;
use haematite::branch::{KeptLimit, RefStore};
use haematite::db::vacuum::CANONICAL_REFS_DIR;
use haematite::db::{DatabaseConfig, ShardReclaimAnswer};
use haematite::store::{Disk, NodeCacheBudget};
use haematite::tree::{Hash, Node};
use serde_json::Value;

use super::{Config, KeepConfig, Releaser, Settings};
use crate::serve::protocol::{
    BranchDiscard, Change, CollectionWrite, Fork, Mutation, Operation, Selection, StoreRead,
    Version,
};
use crate::serve::service::Service;
use crate::serve::storage::{Crash, Keep, Storage};

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

/// A store of one shard keeping one version a record and no past roots, so
/// only the history rows keep an old version.
fn create(directory: &Path) -> Result<Storage, Box<dyn Error>> {
    Ok(Storage::create_kept(
        DatabaseConfig {
            data_dir: directory.join("database"),
            shard_count: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        },
        NodeCacheBudget::Unlimited,
        KeptLimit::new(1)?,
        fixture_long_wait_interval(),
    )?)
}

fn settings(batch: usize, keep: Keep) -> Result<Settings, Box<dyn Error>> {
    Ok(Settings {
        max_rows: NonZeroUsize::new(64).ok_or("zero rows")?,
        batch: NonZeroUsize::new(batch).ok_or("zero batch")?,
        keep,
    })
}

/// Keep the head alone, whatever its age.
const fn head_only() -> Keep {
    Keep {
        versions: NonZeroU64::new(1),
        days: Some(0),
    }
}

/// Keep every version: both settings unset.
const KEEP_ALL: Keep = Keep {
    versions: None,
    days: None,
};

/// A service over a new store, its thread started and the release it asks
/// at start ended; the head of `main`.
fn served(directory: &Path, release: Settings) -> Result<(Service, Version), Box<dyn Error>> {
    let storage = create(directory)?;
    let head = head(&storage)?;
    Ok((started(storage, release)?, head))
}

fn started(storage: Storage, release: Settings) -> Result<Service, Box<dyn Error>> {
    let mut service = Service::new(storage, NonZeroU32::new(256).ok_or("zero page")?)?;
    service.start_release(release)?;
    releaser(&service)?.wait_idle();
    Ok(service)
}

fn head(storage: &Storage) -> Result<Version, Box<dyn Error>> {
    Ok(storage
        .capture(&Selection {
            branch: "main".to_owned(),
            version: None,
        })?
        .version)
}

fn releaser(service: &Service) -> Result<&Releaser, Box<dyn Error>> {
    Ok(service
        .releaser()
        .ok_or("the release thread was not started")?)
}

/// One write to `main` at `expected`, as a client sends it; its version.
fn write(service: &Service, tag: usize, expected: Version) -> Result<Version, Box<dyn Error>> {
    let receipt = service.dispatch(Operation::Mutate(Mutation {
        request_id: format!("write-{tag}"),
        branch: "main".to_owned(),
        expected,
        change: Change::Collection(CollectionWrite {
            collection: format!("c-{tag}"),
            schema: None,
        }),
    }))?;
    Ok(serde_json::from_value(receipt["version"].clone())?)
}

/// A write for each tag, each waited out with any release it made due.
fn writes(service: &Service, tags: Range<usize>, from: Version) -> Result<Version, Box<dyn Error>> {
    let releaser = releaser(service)?;
    let mut at = from;
    for tag in tags {
        at = write(service, tag, at)?;
        releaser.wait_idle();
    }
    Ok(at)
}

fn stats(service: &Service, version: Option<Version>) -> Result<Value, Box<dyn Error>> {
    Ok(service.dispatch(Operation::StoreStats(StoreRead {
        selection: Selection {
            branch: "main".to_owned(),
            version,
        },
    }))?)
}

/// The file of one shard's named root node.
fn root_file(directory: &Path, root: Hash) -> PathBuf {
    let root = root.to_string();
    let (prefix, name) = root.split_at(2);
    directory
        .join("database")
        .join("shard-0")
        .join("store")
        .join(prefix)
        .join(name)
}

/// A released version cannot be read. Its root remains when a live anchor
/// reaches it; an unreachable released root leaves disk after reclaim.
#[test]
fn a_hundred_writes_at_batch_fifty_make_exactly_two_releases() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, mut at) = served(directory.path(), settings(50, head_only())?)?;
    let creation = at;
    let refs_dir = directory.path().join("database").join(CANONICAL_REFS_DIR);
    let main =
        RefStore::read_record(&refs_dir, "@haem/branch/main")?.ok_or("main has no reference")?;
    let shard = main.shards.first().ok_or("main has no shard")?;
    assert_eq!(shard.shard_id, 0);
    assert_eq!(shard.fork_anchor, shard.head);
    let anchor = shard.fork_anchor;
    let mut early = vec![(creation, anchor)];
    assert!(root_file(directory.path(), anchor).try_exists()?);
    // The release asked at start, with every live branch, is not one the
    // writes made due: the two are counted from after it.
    let releaser = releaser(&service)?;
    let start = releaser.releases();
    for tag in 0..50 {
        at = write(&service, tag, at)?;
        let main = RefStore::read_record(&refs_dir, "@haem/branch/main")?
            .ok_or("main has no reference after its write")?;
        let shard = main
            .shards
            .first()
            .ok_or("main has no shard after its write")?;
        assert_eq!(shard.shard_id, 0);
        assert_eq!(shard.fork_anchor, anchor);
        assert!(root_file(directory.path(), shard.head).try_exists()?);
        early.push((at, shard.head));
        releaser.wait_idle();
    }
    assert_eq!(releaser.releases() - start, 1);
    let first = releaser
        .last()
        .and_then(|last| last.answer.released)
        .ok_or("the first release took nothing")?;
    assert_eq!(first.first, creation);
    let Err(failure) = service.dispatch(Operation::StoreStats(StoreRead {
        selection: Selection {
            branch: "main".to_owned(),
            version: Some(creation),
        },
    })) else {
        return Err("the creation version was still readable after release".into());
    };
    assert_eq!(failure.code, "version_released");
    writes(&service, 50..100, at)?;
    assert_eq!(releaser.releases() - start, 2);
    let last = releaser.last().ok_or("no release was kept")?;
    assert!(last.answer.released.is_some());
    assert!(!last.answer.more);
    assert!(last.answer.pass_asked, "{:?}", last.answer.pass_refused);
    let (passes_at_release, _) = reclaim(&service)?;
    service.activate_reclaim()?;
    service.quiesce_reclaim()?;
    let (passes, ended_ms) = reclaim(&service)?;
    let refs = RefStore::open(&refs_dir)?;
    let main = refs
        .get("@haem/branch/main")
        .ok_or("main has no reference")?;
    assert_eq!(
        main.shards.first().ok_or("main has no shard")?.fork_anchor,
        anchor
    );
    assert!(
        root_file(directory.path(), anchor).try_exists()?,
        "main fork anchor on shard 0: {anchor}"
    );
    let mut roots = refs.protected_roots();
    roots.extend(refs.kept_roots());
    let reached = reachable(directory.path(), roots)?;
    let order = |version: Version| (version.generation, version.sequence);
    let (version, root) = early
        .into_iter()
        .find(|(version, root)| {
            order(*version) >= order(first.first)
                && order(*version) <= order(first.last)
                && !reached.contains(root)
        })
        .ok_or("no released root is outside the live anchors, heads and kept trees")?;
    let file = root_file(directory.path(), root);
    assert!(
        !file.try_exists()?,
        "released {}.{} root {root}; main fork anchor {anchor}; passes finished \
         {passes_at_release} at release, {passes} after quiesce; last ended \
         {ended_ms:?} ms; {}; reclaim {:?}",
        version.generation,
        version.sequence,
        file.display(),
        service.reclaim_status()?,
    );
    Ok(())
}

/// The nodes reached by the fixture's surviving branch roots. This chooses
/// the deletion witness from references, independently of file absence.
fn reachable(directory: &Path, roots: HashSet<Hash>) -> Result<HashSet<Hash>, Box<dyn Error>> {
    let store = Disk::new(
        directory.join("database").join("shard-0").join("store"),
        NodeCacheBudget::Unlimited,
    )?;
    let mut reached = HashSet::new();
    let mut pending: Vec<Hash> = roots.into_iter().collect();
    while let Some(root) = pending.pop() {
        if !reached.insert(root) {
            continue;
        }
        let node = store
            .get(&root)?
            .ok_or_else(|| format!("live root lost node {root}"))?;
        if let Node::Internal(internal) = node.as_ref() {
            pending.extend(internal.children().iter().map(|(_, child)| *child));
        }
    }
    Ok(reached)
}

/// Shard 0's passes finished, and when its last pass ended in ms.
fn reclaim(service: &Service) -> Result<(u64, Option<u64>), Box<dyn Error>> {
    match service.reclaim_status()?.shards.first() {
        Some(ShardReclaimAnswer::Answered { status, .. }) => {
            Ok((status.passes_finished, status.last_ended_at_ms))
        }
        other => Err(format!("shard 0's reclaim did not answer: {other:?}").into()),
    }
}

/// Three advances pass the batch threshold with keep unset, without a
/// release or a release-thread storage lock. Check whether it is due before
/// waiting so the thread's second guard cannot hide a scheduling fault.
#[test]
fn three_thresholds_with_keep_unset_make_no_release_or_thread_lock() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, at) = served(directory.path(), settings(1, KEEP_ALL)?)?;
    let releaser = releaser(&service)?;
    assert!(!releaser.due(1));
    writes(&service, 0..3, at)?;
    assert_eq!(releaser.releases(), 0);
    assert_eq!(releaser.locks(), 0);
    assert_eq!(releaser.durable(), SyncCounts::default());
    let status = stats(&service, None)?;
    assert_eq!(
        status["release"]["not_released"],
        "retention.keep.versions and retention.keep.days are not set"
    );
    Ok(())
}

/// HR4-fix (1): a release that fails is not asked again on the next write.
/// Batch 5: the fifth write's release fails; four more writes make no
/// attempt, the fifth makes one; the failure is kept beside the success.
#[test]
fn a_failed_release_is_asked_again_only_after_another_batch() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, at) = served(directory.path(), settings(5, head_only())?)?;
    let releaser = releaser(&service)?;
    let locks = releaser.locks();
    let at = writes(&service, 0..4, at)?;
    releaser.fail_next(Crash::AfterDecision);
    let at = writes(&service, 4..5, at)?;
    assert_eq!(releaser.locks() - locks, 1);
    let failed = releaser.last_failure().ok_or("the failure was not kept")?;
    assert_eq!(failed.branch, "main");
    let at = writes(&service, 5..9, at)?;
    assert_eq!(releaser.locks() - locks, 1);
    writes(&service, 9..10, at)?;
    assert_eq!(releaser.locks() - locks, 2);
    let last = releaser.last().ok_or("the second release did not end")?;
    assert!(last.at >= failed.at);
    let status = stats(&service, None)?;
    assert_eq!(status["release"]["last_failure"]["branch"], "main");
    Ok(())
}

/// HR4-fix (2): the count of advances is in memory, so a service restarted
/// before the batch still releases: every live branch is asked at start.
#[test]
fn a_restart_below_the_batch_still_releases() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, at) = served(directory.path(), settings(50, head_only())?)?;
    writes(&service, 0..10, at)?;
    drop(service);
    let storage = Storage::open_kept(
        &directory.path().join("database"),
        NodeCacheBudget::Unlimited,
        KeptLimit::new(1)?,
        fixture_long_wait_interval(),
    )?;
    let service = started(storage, settings(50, head_only())?)?;
    let releaser = releaser(&service)?;
    assert_eq!(releaser.releases(), 1);
    let last = releaser.last().ok_or("no release at start")?;
    assert!(last.answer.released.is_some());
    Ok(())
}

/// Condition 4: a write arriving while a release holds the storage waits
/// for it, and then succeeds; it is never told `busy`.
#[test]
fn a_write_during_a_release_waits_and_then_succeeds() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, at) = served(directory.path(), settings(50, head_only())?)?;
    let releaser = releaser(&service)?;
    let start = releaser.releases();
    let at = writes(&service, 0..49, at)?;
    releaser.hold_next();
    let at = write(&service, 49, at)?;
    releaser.wait_held();
    let (sent, written) = mpsc::channel();
    std::thread::scope(|scope| -> TestResult {
        let writer =
            scope.spawn(|| sent.send(write(&service, 50, at).map_err(|error| error.to_string())));
        // The write cannot finish while the release holds the storage.
        assert!(written.try_recv().is_err());
        releaser.let_go();
        let next = written.recv()??;
        assert_ne!(next, at);
        writer.join().map_err(|_| "the write panicked")??;
        Ok(())
    })?;
    releaser.wait_idle();
    assert_eq!(releaser.releases() - start, 1);
    Ok(())
}

/// Condition 1: a thread that ended is a named state in `store.stats`,
/// never silence; one never started says so.
#[test]
fn a_killed_thread_is_named_died_in_the_status() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, _) = served(directory.path(), settings(50, head_only())?)?;
    assert_eq!(stats(&service, None)?["release"]["state"], "waiting");
    let releaser = releaser(&service)?;
    releaser.kill();
    releaser.wait_died();
    let status = stats(&service, None)?;
    assert_eq!(status["release"]["state"], "died");
    let reason = status["release"]["died"].as_str().ok_or("no reason")?;
    assert!(reason.starts_with("release_thread_killed"), "{reason}");

    let unstarted = tempfile::tempdir()?;
    let storage = create(unstarted.path())?;
    let service = Service::new(storage, NonZeroU32::MIN)?;
    assert_eq!(stats(&service, None)?["release"]["state"], "not_started");
    Ok(())
}

/// Condition 2: what waits is a set of names, so three wakes of a branch
/// while it is being released hold one entry.
#[test]
fn a_branch_waits_once_however_often_it_is_woken() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, at) = served(directory.path(), settings(50, head_only())?)?;
    let releaser = releaser(&service)?;
    let start = releaser.releases();
    let at = writes(&service, 0..49, at)?;
    releaser.hold_next();
    write(&service, 49, at)?;
    releaser.wait_held();
    for _ in 0..3 {
        releaser.wake("main");
    }
    assert_eq!(releaser.status()["waiting"], 1);
    releaser.let_go();
    releaser.wait_idle();
    assert_eq!(releaser.releases() - start, 2);
    Ok(())
}

/// Condition 6: every limit refused by its name at start, zero and past
/// its most; `keep.days` of 0 is a setting (keep nothing for its age).
#[test]
fn every_setting_is_refused_by_name() -> TestResult {
    let refused = |config: Config, name: &str| {
        let answer = config.settings().err().unwrap_or_default();
        assert!(answer.contains(name), "{answer}");
    };
    refused(
        Config {
            release_max_rows: 0,
            ..Config::default()
        },
        "retention.release_max_rows",
    );
    refused(
        Config {
            release_max_rows: 65_537,
            ..Config::default()
        },
        "retention.release_max_rows",
    );
    refused(
        Config {
            release_batch: 0,
            ..Config::default()
        },
        "retention.release_batch",
    );
    let keep = |versions, days| Config {
        keep: KeepConfig { versions, days },
        ..Config::default()
    };
    refused(keep(Some(0), None), "retention.keep.versions");
    refused(keep(None, Some(36_501)), "retention.keep.days");
    let days_zero = keep(Some(1), Some(0)).settings()?;
    assert_eq!(days_zero.keep, head_only());
    let defaults = Config::default().settings()?;
    assert_eq!(defaults.max_rows.get(), 1_024);
    assert_eq!(defaults.batch.get(), 256);
    assert_eq!(defaults.keep, KEEP_ALL);
    let unknown = serde_json::from_str::<Config>(r#"{"keep":{"version":1}}"#);
    assert!(unknown.is_err());
    Ok(())
}

/// `main` forked as `work` on `storage`: `work`'s version.
fn fork_work(storage: &mut Storage) -> Result<Version, Box<dyn Error>> {
    let at = head(storage)?;
    storage.mutate(&Mutation {
        request_id: "fork-work".to_owned(),
        branch: "main".to_owned(),
        expected: at,
        change: Change::Fork(Fork {
            name: "work".to_owned(),
            from: at,
        }),
    })?;
    Ok(storage
        .capture(&Selection {
            branch: "work".to_owned(),
            version: None,
        })?
        .version)
}

fn discard_work(at: Version) -> Mutation {
    Mutation {
        request_id: "discard-work".to_owned(),
        branch: "work".to_owned(),
        expected: at,
        change: Change::Discard(BranchDiscard {
            name: "work".to_owned(),
        }),
    }
}

/// The driver asks at start for every name a discard left rows of, read from
/// one page of their markers
/// (docs/design/HAEM-DISCARDED-ROWS-RELEASE-2026-10-06.md).
#[test]
fn the_driver_releases_a_discarded_generation_at_start() -> TestResult {
    let directory = tempfile::tempdir()?;
    let mut storage = create(directory.path())?;
    let at = fork_work(&mut storage)?;
    storage.mutate(&discard_work(at))?;
    let service = started(storage, settings(50, head_only())?)?;
    let last = releaser(&service)?.last().ok_or("no release at start")?;
    assert_eq!(last.branch, "work", "the discarded name is asked at start");
    assert!(last.answer.released.is_some(), "{:?}", last.answer);
    Ok(())
}

/// A discard wakes the driver for its name: its rows go without waiting for
/// a restart or a batch of writes.
#[test]
fn a_discard_wakes_the_driver_for_its_name() -> TestResult {
    let directory = tempfile::tempdir()?;
    let mut storage = create(directory.path())?;
    let at = fork_work(&mut storage)?;
    let service = started(storage, settings(50, head_only())?)?;
    service.dispatch(Operation::Mutate(discard_work(at)))?;
    let releaser = releaser(&service)?;
    releaser.wait_idle();
    let last = releaser.last().ok_or("no release after the discard")?;
    assert_eq!(last.branch, "work", "the discard woke the driver for work");
    assert!(last.answer.released.is_some(), "{:?}", last.answer);
    Ok(())
}

/// A day, and a second, in nanoseconds.
const PAST_A_DAY: u64 = (86_400 + 1) * 1_000_000_000;

/// How long a test waits for a release its moved clock makes due; the thread
/// reads the moved clock every 10 ms (releaser.rs `TEST_WAIT`).
const DUE_WAIT: std::time::Duration = std::time::Duration::from_secs(10);

/// Keep one version and one day.
const fn a_day() -> Keep {
    Keep {
        versions: NonZeroU64::new(1),
        days: Some(1),
    }
}

/// Keep one day: a discarded generation too young to go is asked again when
/// its row falls due, with no restart and no write; the clock is moved, never
/// slept on (docs/design/HAEM-DISCARDED-ROWS-RELEASE-2026-10-06.md).
#[test]
fn a_held_generation_is_released_when_its_day_has_passed() -> TestResult {
    let directory = tempfile::tempdir()?;
    let mut storage = create(directory.path())?;
    let at = fork_work(&mut storage)?;
    let service = started(storage, settings(50, a_day())?)?;
    service.dispatch(Operation::Mutate(discard_work(at)))?;
    let releaser = releaser(&service)?;
    releaser.wait_idle();
    let held = releaser.last().ok_or("no release after the discard")?;
    assert_eq!(held.branch, "work");
    assert!(
        held.answer.released.is_none(),
        "too young to go: {:?}",
        held.answer
    );
    assert!(
        releaser.next_due().is_some(),
        "the held row's day is waited for"
    );
    let before = releaser.releases();
    releaser.advance_clock(PAST_A_DAY);
    assert!(
        releaser.wait_releases(before + 1, DUE_WAIT),
        "no release once the day had passed"
    );
    let last = releaser.last().ok_or("no release")?;
    assert_eq!(last.branch, "work");
    assert!(last.answer.released.is_some(), "{:?}", last.answer);
    assert_eq!(releaser.next_due(), None, "nothing is held any more");
    Ok(())
}

/// With nothing held by days the thread waits with no deadline: it is not
/// woken in a loop, so its wakes do not grow while it is idle.
#[test]
fn an_idle_releaser_with_nothing_due_does_not_wake() -> TestResult {
    let directory = tempfile::tempdir()?;
    let (service, _at) = served(directory.path(), settings(50, head_only())?)?;
    let releaser = releaser(&service)?;
    releaser.wait_idle();
    assert_eq!(releaser.next_due(), None, "nothing is held by days");
    let before = releaser.wakes();
    for _ in 0..10_000 {
        std::thread::yield_now();
    }
    assert_eq!(releaser.wakes(), before, "the idle thread woke");
    Ok(())
}

/// The version `name` is at now.
fn version_of(storage: &Storage, name: &str) -> Result<Version, Box<dyn Error>> {
    Ok(storage
        .capture(&Selection {
            branch: name.to_owned(),
            version: None,
        })?
        .version)
}

/// `name` forked from `from` at its head.
fn fork_from(storage: &mut Storage, from: &str, name: &str) -> TestResult {
    let at = version_of(storage, from)?;
    storage.mutate(&Mutation {
        request_id: format!("fork-{name}"),
        branch: from.to_owned(),
        expected: at,
        change: Change::Fork(Fork {
            name: name.to_owned(),
            from: at,
        }),
    })?;
    Ok(())
}

/// `name` discarded at its head.
fn discard(storage: &mut Storage, name: &str) -> TestResult {
    storage.mutate(&Mutation {
        request_id: format!("discard-{name}"),
        branch: name.to_owned(),
        expected: version_of(storage, name)?,
        change: Change::Discard(BranchDiscard {
            name: name.to_owned(),
        }),
    })?;
    Ok(())
}

/// Outside review 15: with two markers a page, the first two discarded
/// generations held for good by a fork from each, the walk still reaches
/// the third and releases it within three pages, at start, with no write.
#[test]
fn the_marker_walk_reaches_a_generation_behind_two_held_ones() -> TestResult {
    let directory = tempfile::tempdir()?;
    let mut storage = create(directory.path())?;
    // Markers are ordered by the name's hash: take the names in that order.
    let mut names = ["gone-a", "gone-b", "gone-c"];
    names.sort_by_key(|name| Hash::of(name.as_bytes()).as_bytes().to_vec());
    let [first, second, third] = names;
    for name in names {
        fork_from(&mut storage, "main", name)?;
    }
    for held in [first, second] {
        fork_from(&mut storage, held, &format!("{held}-fork"))?;
    }
    for name in names {
        discard(&mut storage, name)?;
    }
    let mut release = settings(50, head_only())?;
    release.max_rows = NonZeroUsize::new(2).ok_or("zero rows")?;
    let service = started(storage, release)?;
    let releaser = releaser(&service)?;

    let taken = releaser.released();
    assert!(
        taken.iter().any(|name| name == third),
        "{third} released: {taken:?}"
    );
    assert!(
        !taken.iter().any(|name| name == first || name == second),
        "the held generations stay: {taken:?}"
    );
    let pages = releaser.pages();
    assert!(pages <= 3, "within three pages: {pages}");
    Ok(())
}

const fn fixture_long_wait_interval() -> std::time::Duration {
    std::time::Duration::from_secs(1)
}