liminal-server 0.28.6

//! DUR-021 R1 and R3 at the store: held at open, released once and again
//! without harm, the watch thread named, no quiesce on these paths.

use std::time::Instant;

use haematite::db::{ReclaimPhase, ShardReclaimAnswer};
use haematite::{Database, DatabaseConfig, ReclaimStart, RuntimeOptions};

use super::reclaim_placement::{WATCH_THREAD, arm_watch, release};

fn held_store(dir: &std::path::Path) -> Result<Database, Box<dyn std::error::Error>> {
    Database::create(
        DatabaseConfig {
            data_dir: dir.to_path_buf(),
            shard_count: 2,
            distributed: None,
            executor_threads: None,
            node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
            retention: super::services::DURABLE_RETENTION,
        },
        RuntimeOptions {
            long_wait_interval: std::time::Duration::from_secs(5),
            reclaim_start: ReclaimStart::Held,
        },
    )
    .map_err(Into::into)
}

#[test]
fn a_held_store_runs_no_pass_until_released_and_a_second_release_is_harmless()
-> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let database = held_store(dir.path())?;
    for shard in &database.reclaim_status().shards {
        match shard {
            ShardReclaimAnswer::Answered { status, .. } => {
                assert_eq!(status.phase, ReclaimPhase::Idle, "held: no step runs");
                assert_eq!(status.passes_finished, 0);
            }
            ShardReclaimAnswer::NotAnswering { shard, reason } => {
                return Err(format!("shard {shard} not answering: {reason}").into());
            }
        }
    }
    database.release_reclaim()?;
    database.release_reclaim()?;
    assert_eq!(database.reclaim_status().shards.len(), 2);
    Ok(())
}

#[test]
fn the_watch_thread_is_named_and_ends_when_the_store_closes()
-> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let database = held_store(dir.path())?;
    arm_watch(&database);
    let store = haematite::EventStore::new(database);
    release(&store, Instant::now());
    let named = std::thread::current().name().map(str::to_owned);
    assert_ne!(
        named.as_deref(),
        Some(WATCH_THREAD),
        "the watch is not this thread"
    );
    drop(store);
    // The watch holds the receiver only; with the store closed its senders
    // are gone and the thread's recv ends. Nothing here waits on a timer.
    Ok(())
}

/// The seats' server, 12 October: the release waited 1,625 ms for shard 4
/// to finish one reclaim step before the server said started and started its
/// services. A release that cannot answer does not hold the start: the call
/// returns while the release is still waiting, and the release still runs,
/// on its own named thread.
#[test]
fn a_release_held_by_a_slow_reclaim_step_does_not_hold_the_start()
-> Result<(), Box<dyn std::error::Error>> {
    let (step_ends, slow_step) = std::sync::mpsc::channel::<()>();
    let (released, release_ran) = std::sync::mpsc::channel();
    let began = std::time::Instant::now();
    super::reclaim_placement::off_the_start_path(move || {
        // A shard mid-step: the release waits until the step ends.
        let _ = slow_step.recv_timeout(std::time::Duration::from_secs(2));
        let _ = released.send(std::thread::current().name().map(str::to_owned));
    });
    let returned_after = began.elapsed();
    assert!(
        release_ran.try_recv().is_err(),
        "the release finished before the step ended"
    );
    assert!(
        returned_after < std::time::Duration::from_millis(100),
        "the start waited {returned_after:?} for the release"
    );
    step_ends.send(())?;
    let thread = release_ran.recv_timeout(std::time::Duration::from_secs(10))?;
    assert_eq!(
        thread.as_deref(),
        Some(super::reclaim_placement::RELEASE_THREAD)
    );
    Ok(())
}

/// DUR-021 R5: haematite 0.33.0's `reclaim_status` boots every shard that has not started, so
/// the release asks none for its status and logs its time alone.
#[test]
fn a_release_logs_its_time_and_asks_no_shard_for_its_status()
-> Result<(), Box<dyn std::error::Error>> {
    let dir = tempfile::tempdir()?;
    let spine = haematite::EventStore::new(held_store(dir.path())?);
    let log = crate::test_log::CapturedLog::default();
    log.capture(|| release(&spine, Instant::now()));
    let text = log.contents();
    assert!(
        text.contains("spine reclaim released") && text.contains("boot_elapsed_ms"),
        "{text}"
    );
    assert!(!text.contains("phases"), "no shard was asked: {text}");
    assert!(
        !text.contains("spine reclaim resumed") && !text.contains("starts from nothing"),
        "no resume line until reclaim_status_published: {text}"
    );
    Ok(())
}