aion-rs 0.31.0

//! Singleton process-exit drainer contract tests.

use std::collections::HashSet;
use std::sync::{Arc, mpsc};
use std::time::{Duration, Instant};

use beamr::scheduler::EXIT_EVENT_CAPACITY;

use super::{ProcessEnding, ProcessExitRegistry};
use crate::EngineError;
use crate::runtime::{RuntimeConfig, RuntimeHandle, SignalDeliveryConfig, UndeliveredWake};

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

#[test]
fn runtime_claims_the_only_exit_event_subscription_with_typed_duplicate_failure() -> TestResult {
    let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(
        Some(1),
        crate::runtime::config::TEST_STOP_DRAIN_TIMEOUT,
    ))?);
    assert!(runtime.scheduler.subscribe_exit_events().is_none());

    let duplicate = ProcessExitRegistry::new(
        Arc::clone(&runtime.scheduler),
        runtime.stop_drain_timeout(),
        runtime.signal_delivery().max_enqueue_attempts as usize,
    )
    .err()
    .ok_or("a second process-exit registry claimed the scheduler subscription")?;

    assert!(matches!(
        duplicate,
        EngineError::ProcessExitSubscriptionUnavailable
    ));
    runtime.shutdown()?;
    Ok(())
}

#[test]
fn lagged_stream_resynchronizes_every_registered_outcome() -> TestResult {
    let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(
        Some(1),
        crate::runtime::config::TEST_STOP_DRAIN_TIMEOUT,
    ))?);
    runtime.process_exits.pause_for_test();

    let first_pid = runtime.spawn_test_process()?;
    runtime.cancel_pid(first_pid)?;
    runtime
        .process_exits
        .wait_for_pause_for_test(Duration::from_secs(10))?;

    let mut pids = Vec::with_capacity(EXIT_EVENT_CAPACITY + 2);
    pids.push(first_pid);
    for _ in 0..=EXIT_EVENT_CAPACITY {
        let pid = runtime.spawn_test_process()?;
        runtime.cancel_pid(pid)?;
        pids.push(pid);
    }

    let (sender, receiver) = mpsc::channel();
    for pid in &pids {
        let callback_sender = sender.clone();
        let callback_pid = *pid;
        runtime.monitor_process_for_test(*pid, move |outcome| {
            let _ = callback_sender.send((callback_pid, outcome.is_ok()));
        })?;
    }
    drop(sender);
    runtime.process_exits.release_for_test();

    let deadline = Instant::now() + Duration::from_secs(20);
    let mut observed = HashSet::with_capacity(pids.len());
    while observed.len() < pids.len() {
        let remaining = deadline.saturating_duration_since(Instant::now());
        let (pid, outcome_available) = receiver.recv_timeout(remaining)?;
        if !outcome_available {
            return Err(format!("process {pid} did not receive its cached exit outcome").into());
        }
        observed.insert(pid);
    }

    wait_until(deadline, || {
        runtime.process_exits.lag_recoveries_for_test() > 0
            && pids
                .iter()
                .all(|pid| runtime.process_cleanup_complete_for_test(*pid))
    })?;
    assert_eq!(observed.len(), pids.len());
    assert!(runtime.process_exits.lag_recoveries_for_test() > 0);
    runtime.shutdown()?;
    Ok(())
}

fn wait_until(deadline: Instant, predicate: impl Fn() -> bool) -> TestResult {
    while Instant::now() < deadline {
        if predicate() {
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(1));
    }
    Err("process-exit condition did not become true before its deadline".into())
}

/// The registry's verdict that a process ended is the ground the query and
/// signal wake-failure classifiers stand on, so it may rest on the registry's
/// own record and nothing else.
///
/// Registration is forgotten when a record is retired, and the registration
/// watermark cannot tell a reaped process from a pid that was never
/// registered. Reading that watermark as a terminal answered "this workflow
/// ended" for every unregistered pid at or below it — pid zero, which the
/// scheduler never allocates, among them — which is how an undelivered
/// message came to be reported as a completed workflow.
#[test]
fn a_never_registered_pid_below_the_watermark_has_no_recorded_ending() -> TestResult {
    let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(
        Some(1),
        crate::runtime::config::TEST_STOP_DRAIN_TIMEOUT,
    ))?);
    let registered = runtime.spawn_test_process()?;
    let above = registered
        .checked_add(1)
        .ok_or("fixture control: the registration watermark is at the pid ceiling")?;

    // Pid zero is the sentinel the registry's own test seams use for "no
    // pid": the scheduler never allocates it, so it is certainly a pid this
    // registry never registered, and it is at or below every watermark.
    let never_allocated = 0;
    let mut unregistered = vec![never_allocated];
    unregistered.extend(
        (1..registered)
            .rev()
            .filter(|pid| !runtime.process_exits.contains(*pid))
            .take(4),
    );

    for pid in unregistered {
        assert!(
            runtime.process_exits.below_registration_watermark(pid),
            "fixture control: pid {pid} must sit at or below the registration watermark \
             with no record held"
        );
        assert_eq!(
            runtime.process_exits.ending_knowledge(pid)?,
            ProcessEnding::Forgotten,
            "an unregistered pid below the watermark is forgotten, not terminal"
        );
        assert!(
            !runtime.process_exits.has_terminal(pid)?,
            "pid {pid} was never registered here, so the registry holds no ending for it"
        );
        assert!(
            !runtime.process_ending_recorded(pid)?,
            "the shared completion classifier must never read the watermark as an ending"
        );
    }

    // The controls on either side of the watermark: a registered process that
    // has not exited is known and pending, and a pid above the watermark is
    // provably outside anything this registry has registered.
    assert_eq!(
        runtime.process_exits.ending_knowledge(registered)?,
        ProcessEnding::Pending
    );
    assert!(!runtime.process_exits.has_terminal(registered)?);
    assert_eq!(
        runtime.process_exits.ending_knowledge(above)?,
        ProcessEnding::NeverRegistered
    );
    assert!(!runtime.process_exits.has_terminal(above)?);

    runtime.shutdown()?;
    Ok(())
}

/// The counterpart of the pin above: once the drainer publishes a terminal on
/// a held record, the registry reports the ending it actually holds.
#[test]
fn a_published_terminal_on_a_held_record_is_a_recorded_ending() -> TestResult {
    let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(
        Some(1),
        crate::runtime::config::TEST_STOP_DRAIN_TIMEOUT,
    ))?);
    let pid = runtime.spawn_test_process()?;
    runtime.cancel_pid(pid)?;
    let record = runtime
        .process_exits
        .find(pid)
        .ok_or("fixture control: a spawned process must hold an exit record")?;
    drop(record.wait()?);

    assert_eq!(
        runtime.process_exits.ending_knowledge(pid)?,
        ProcessEnding::Recorded
    );
    assert!(runtime.process_exits.has_terminal(pid)?);
    assert!(
        runtime.process_ending_recorded(pid)?,
        "a published terminal is a recorded ending without any cleanup tombstone"
    );
    runtime.shutdown()?;
    Ok(())
}

/// How long these pins wait for the drainer to reach or leave its test pause.
const EXIT_IN_FLIGHT_FIXTURE_TIMEOUT: Duration = Duration::from_secs(10);

/// Build a runtime whose wake-readiness window is `ready_timeout`.
///
/// That window is the bound the undelivered-wake classifier parks an exit in
/// flight under, so these pins set it explicitly rather than inheriting a
/// default they would then be measuring against.
fn runtime_with_ready_timeout(
    ready_timeout: Duration,
) -> Result<Arc<RuntimeHandle>, Box<dyn std::error::Error>> {
    let delivery = SignalDeliveryConfig::new(
        ready_timeout,
        1,
        Duration::from_millis(1),
        Duration::from_millis(1),
    );
    Ok(Arc::new(RuntimeHandle::new(
        RuntimeConfig::new(Some(1), crate::runtime::config::TEST_STOP_DRAIN_TIMEOUT)
            .with_signal_delivery(delivery),
    )?))
}

/// Cancel a process behind a held drainer and confirm the exit-in-flight
/// window: the registry holds the record with nothing published on it, the
/// pid has left the scheduler's table, and no cleanup tombstone exists.
fn exit_in_flight_pid(runtime: &Arc<RuntimeHandle>) -> Result<u64, Box<dyn std::error::Error>> {
    let pid = runtime.spawn_test_process()?;
    runtime.cancel_pid(pid)?;
    wait_until(Instant::now() + EXIT_IN_FLIGHT_FIXTURE_TIMEOUT, || {
        !runtime.is_live(pid)
    })?;
    assert!(
        !runtime.process_cleanup_started(pid),
        "fixture control: Aion cleanup cannot have started without a monitor"
    );
    assert_eq!(
        runtime.process_exits.ending_knowledge(pid)?,
        ProcessEnding::Pending,
        "fixture control: the held drainer must leave the record unpublished"
    );
    Ok(pid)
}

/// beamr drops the process-table row on its own schedule and the drainer
/// publishes the terminal afterwards, so a single read taken between the two
/// sees neither. The classifier parks on the record's own publication for the
/// readiness window instead of answering from that gap: reading the gap as
/// "did not end" reports an accepted signal as an undelivered one, and the
/// signal is already durable, so a caller that re-sends on that failure
/// appends it twice.
#[test]
fn an_exit_in_flight_is_waited_out_and_resolves_to_the_recorded_ending() -> TestResult {
    // A window far longer than the release below needs, so the pin measures
    // the wait rather than the machine it runs on.
    let runtime = runtime_with_ready_timeout(EXIT_IN_FLIGHT_FIXTURE_TIMEOUT)?;
    runtime.pause_exit_drainer_for_test(EXIT_IN_FLIGHT_FIXTURE_TIMEOUT)?;
    let pid = exit_in_flight_pid(&runtime)?;

    let releaser = {
        let runtime = Arc::clone(&runtime);
        std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(5));
            runtime.release_exit_drainer_for_test();
        })
    };
    let verdict = runtime.classify_undelivered_wake(pid);
    releaser
        .join()
        .map_err(|_| "the drainer-release thread panicked")?;

    assert_eq!(
        verdict?,
        UndeliveredWake::ProcessEnded,
        "an exit that publishes inside the readiness window is a recorded ending"
    );
    runtime.shutdown()?;
    Ok(())
}

/// The bound on that wait is real: an exit that publishes nothing inside the
/// readiness window is reported as still in flight, never silently answered
/// either way, and the window expiring does not make the ending appear.
#[test]
fn an_exit_publishing_nothing_inside_the_window_is_reported_still_in_flight() -> TestResult {
    let runtime = runtime_with_ready_timeout(Duration::from_millis(20))?;
    runtime.pause_exit_drainer_for_test(EXIT_IN_FLIGHT_FIXTURE_TIMEOUT)?;
    let pid = exit_in_flight_pid(&runtime)?;

    let verdict = runtime.classify_undelivered_wake(pid);
    runtime.release_exit_drainer_for_test();
    assert_eq!(
        verdict?,
        UndeliveredWake::ExitInFlight,
        "a terminal that never publishes inside the window leaves the exit in flight"
    );

    // Released, the same pid settles as the recorded ending it always was.
    wait_until(Instant::now() + EXIT_IN_FLIGHT_FIXTURE_TIMEOUT, || {
        matches!(
            runtime.classify_undelivered_wake(pid),
            Ok(UndeliveredWake::ProcessEnded)
        )
    })?;
    runtime.shutdown()?;
    Ok(())
}

/// The classifier decides everything but the cleanup tombstone from ONE read
/// of the registry, and that read keeps the evidence it finds.
///
/// This is the state that read would see when the drainer publishes between
/// the tombstone check and it: a record still held, a terminal published on
/// it, no tombstone, and the pid already gone from the scheduler's table. The
/// only thing that can answer it correctly is the read's own `Recorded` arm —
/// liveness says "gone" and the record is not `Pending` — so a classifier that
/// asked merely whether the registry was "not Pending" would report an ending
/// the registry holds as a refusal, and the caller would re-send a signal that
/// was already accepted.
#[test]
fn a_terminal_published_before_the_registry_read_is_an_ending_not_a_refusal() -> TestResult {
    let runtime = runtime_with_ready_timeout(Duration::from_millis(20))?;
    let pid = runtime.spawn_test_process()?;
    runtime.cancel_pid(pid)?;
    wait_until(Instant::now() + EXIT_IN_FLIGHT_FIXTURE_TIMEOUT, || {
        matches!(runtime.process_exits.has_terminal(pid), Ok(true))
    })?;
    wait_until(Instant::now() + EXIT_IN_FLIGHT_FIXTURE_TIMEOUT, || {
        !runtime.is_live(pid)
    })?;
    assert!(
        !runtime.process_cleanup_started(pid),
        "fixture control: no monitor is installed, so no tombstone can exist"
    );
    assert_eq!(
        runtime.process_exits.ending_knowledge(pid)?,
        ProcessEnding::Recorded,
        "fixture control: the registry must still hold the published terminal"
    );

    assert_eq!(
        runtime.classify_undelivered_wake(pid)?,
        UndeliveredWake::ProcessEnded,
        "a terminal the registry holds is an ending on the read that finds it"
    );
    runtime.shutdown()?;
    Ok(())
}