liminal-server 0.23.1

use std::error::Error;
use std::sync::atomic::Ordering;
use std::thread;
use std::time::Duration;

use super::{ShutdownHandle, wait_after_force_close};
use crate::server::connection::ConnectionSupervisor;

#[test]
fn shutdown_handle_initiates_once() {
    let handle = ShutdownHandle::new();

    assert!(!handle.is_initiated());
    assert!(handle.initiate());
    assert!(handle.is_initiated());
    assert!(!handle.initiate());
}

/// What a helper answers when the test signal lock is poisoned.
const SIGNAL_LOCK_POISONED: &str =
    "the test signal lock was poisoned: another test thread panicked holding it";

/// Blocks until a waiter has parked in `wait`, or fails by name if it
/// returned without parking. Both test-only counts rise under
/// `signal_lock` and are signalled on the test condvar, and this wakes
/// holding that lock, so neither can be missed and nothing here waits on
/// something that might never come.
fn await_parked(handle: &ShutdownHandle) -> Result<(), Box<dyn Error>> {
    let state = &handle.inner;
    let mut guard = state.signal_lock.lock().map_err(|_| SIGNAL_LOCK_POISONED)?;
    loop {
        if state.returned.load(Ordering::SeqCst) > 0 {
            return Err("the waiter returned from wait() without parking".into());
        }
        if state.parked.load(Ordering::SeqCst) > 0 {
            drop(guard);
            return Ok(());
        }
        guard = state.signal.wait(guard).map_err(|_| SIGNAL_LOCK_POISONED)?;
    }
}

/// After `initiate`, blocks until the parked waiter has returned, or fails
/// by name when it cannot: `notify` never reached its condvar, so no wake
/// was sent; or the waiter woke and parked again (`parked` rose past
/// `parked_before`, read before `initiate`), so it did not see the
/// shutdown and nothing will wake it again. Every rise of `parked` and
/// `returned` is signalled on the test condvar under `signal_lock`, which
/// is refused by name if poisoned, never recovered. `wait_lock` is not
/// taken here, so a test that poisons it still reads the waiter's return.
fn await_returned(handle: &ShutdownHandle, parked_before: usize) -> Result<(), Box<dyn Error>> {
    let state = &handle.inner;
    if state.notified.load(Ordering::SeqCst) == 0 {
        return Err("initiate() never notified the parked waiter".into());
    }
    let mut guard = state.signal_lock.lock().map_err(|_| SIGNAL_LOCK_POISONED)?;
    while state.returned.load(Ordering::SeqCst) == 0 {
        if state.parked.load(Ordering::SeqCst) > parked_before {
            return Err("the waiter woke and parked again: it did not see the shutdown".into());
        }
        guard = state.signal.wait(guard).map_err(|_| SIGNAL_LOCK_POISONED)?;
    }
    drop(guard);
    Ok(())
}

#[test]
fn shutdown_handle_wait_unblocks_on_initiate() -> Result<(), Box<dyn std::error::Error>> {
    let handle = ShutdownHandle::new();
    let waiter = handle.clone();
    let worker = thread::spawn(move || {
        waiter.wait();
        waiter.is_initiated()
    });

    await_parked(&handle)?;
    let parked = handle.inner.parked.load(Ordering::SeqCst);
    assert!(handle.initiate());
    await_returned(&handle, parked)?;
    let observed = worker.join().map_err(|_| "wait worker panicked")?;

    assert!(observed);
    Ok(())
}

/// The production lock under test.
fn wait_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
    &state.wait_lock
}

/// The test helpers' own lock.
fn signal_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
    &state.signal_lock
}

/// Poisons the handle's lock that `lock` names by panicking a thread
/// while it holds the guard.
fn poison(handle: &ShutdownHandle, lock: fn(&super::ShutdownState) -> &std::sync::Mutex<()>) {
    let poisoner = handle.clone();
    let outcome = thread::spawn(move || {
        let _guard = lock(&poisoner.inner).lock();
        std::panic::resume_unwind(Box::new("poison a shutdown lock"));
    })
    .join();
    assert!(outcome.is_err(), "the poisoner thread must have panicked");
}

/// Poisons the handle's `wait_lock`.
fn poison_wait_lock(handle: &ShutdownHandle) {
    poison(handle, wait_lock_of);
}

/// A poisoned test signal lock is refused by name, never recovered: the
/// helpers answer an error that names it.
#[test]
fn a_poisoned_test_signal_lock_is_refused_by_name() {
    let handle = ShutdownHandle::new();
    poison(&handle, signal_lock_of);
    assert!(handle.initiate());

    let refused =
        await_returned(&handle, 0).map_or_else(|error| error.to_string(), |()| String::new());
    assert_eq!(refused, SIGNAL_LOCK_POISONED);
    let refused = await_parked(&handle).map_or_else(|error| error.to_string(), |()| String::new());
    assert_eq!(refused, SIGNAL_LOCK_POISONED);
}

/// A poisoned wait lock must not silently masquerade as a completed
/// shutdown: `wait()` still returns (behaviour unchanged) but logs at
/// error level.
#[test]
fn wait_on_poisoned_lock_returns_and_logs_error() {
    let handle = ShutdownHandle::new();
    poison_wait_lock(&handle);

    let log = crate::test_log::CapturedLog::default();
    log.capture(|| handle.wait());

    let output = log.contents();
    assert!(
        output.contains("ERROR"),
        "a poisoned wait lock must be logged at error level, got: {output}"
    );
}

/// A waiter already parked in `wait()` must still be woken by `notify()`
/// when the lock has been poisoned after it parked: `initiate()` must never
/// silently skip the wake.
#[test]
fn notify_on_poisoned_lock_still_wakes_parked_waiter() -> Result<(), Box<dyn std::error::Error>> {
    let handle = ShutdownHandle::new();
    let waiter = handle.clone();
    let worker = thread::spawn(move || waiter.wait());
    // The waiter parks on the condvar, releasing the lock, so the poisoner
    // below can take it.
    await_parked(&handle)?;
    poison_wait_lock(&handle);

    let parked = handle.inner.parked.load(Ordering::SeqCst);
    assert!(handle.initiate());

    // A parked waiter must wake after notify under a poisoned lock.
    await_returned(&handle, parked)?;
    worker.join().map_err(|_| "wait worker panicked")?;
    Ok(())
}

/// The close-settle wait returns immediately when nothing is tracked, so a
/// shutdown with no connections spends no time here at all.
#[test]
fn the_close_settle_returns_immediately_when_no_connections_are_active()
-> Result<(), Box<dyn std::error::Error>> {
    let supervisor = ConnectionSupervisor::new()?;

    let started = std::time::Instant::now();
    wait_after_force_close(&supervisor);
    let elapsed = started.elapsed();

    assert!(
        elapsed < Duration::from_millis(50),
        "the close settle took {elapsed:?} with no connections tracked"
    );
    supervisor.shutdown();
    Ok(())
}

/// Both production modules are inspected so moving a close leg cannot hide a drain.
#[test]
fn shutdown_source_has_no_drain_and_no_reap_count_sleep_loop() {
    let implementation = shutdown_implementation();
    for forbidden in [
        "DRAIN_PROGRESS_INTERVAL",
        "FORCE_CLOSE_SETTLE_TIMEOUT",
        "FORCE_CLOSE_POLL_INTERVAL",
        "reap_crashed_connections",
        "fn drain_connections",
    ] {
        assert!(
            !implementation.contains(forbidden),
            "retired poll/reap/drain token `{forbidden}` must not appear in the shutdown implementation"
        );
    }
}

/// The configured drain timeout reaches the shutdown path and is used for
/// exactly one thing: being named in the line that says it is ignored. If a
/// future edit re-arms it as a deadline, this goes red.
#[test]
fn the_configured_drain_timeout_is_never_turned_into_a_deadline() {
    let implementation = shutdown_implementation();
    assert!(
        !implementation.contains("+ drain_timeout"),
        "drain_timeout must never be added to an Instant to form a shutdown deadline"
    );
    assert!(
        implementation.contains("ignored_drain_timeout = ?drain_timeout"),
        "drain_timeout must still be named in the shutdown log line that reports it ignored"
    );
}

fn shutdown_implementation() -> String {
    [
        include_str!("shutdown.rs"),
        include_str!("shutdown_sequence.rs"),
    ]
    .map(|source| source.split("mod tests").next().unwrap_or(source))
    .join("\n")
}