liminal-server 0.22.0

//! 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(())
}