aion-rs 0.31.0

//! Run-awareness of the query mailbox seam's `resolve_workflow` (#105).
//!
//! After a continue-as-new two handles share one workflow id — on the nif path
//! the predecessor's handle is never removed, and on the API path
//! `lifecycle::continue_as_new` records the terminal, starts the successor, and
//! only then removes it. These tests pin that the seam names the CURRENT run
//! rather than whichever handle a `HashMap` scan happened to yield.
//!
//! 🔴 THE DEFECT THESE GUARD IS NONDETERMINISTIC. The replaced implementation
//! scanned `Registry::list`, i.e. `HashMap::values()` under `RandomState`, so it
//! returned the right handle about half the time. The tests below are
//! deterministic on the fix; a MUTATION CHECK against the old scan is not, and
//! must be repeated — one green mutant run is luck, not evidence of vacuity.
//! Recorded red count when the fix was reverted: see `gate-logs/105-*`.

use std::sync::{Arc, Weak};
use std::time::Duration;

use aion_core::{ContentType, Payload, RunId, WorkflowId, WorkflowStatus};
use aion_package::ContentHash;
use aion_store::InMemoryStore;

use super::*;
use crate::durability::Recorder;
use crate::engine_seam::{
    EngineHandle, EngineSeamError, WorkflowMailboxMessage, WorkflowProcessHandle, WorkflowResidency,
};
use crate::query::QueryError;
use crate::registry::{
    CompletionNotifier, HandleResidency, Registry, WorkflowHandle, WorkflowHandleParts,
};
use crate::runtime::nif_query::{is_query_registered, register_query_impl};
use crate::runtime::{RuntimeConfig, RuntimeHandle, UndeliveredWake};

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

/// A predecessor's pid and a successor's pid, chosen distinct so an assertion
/// on the resolved pid names WHICH run answered rather than merely that one did.
const PREDECESSOR_PID: u64 = 4_001;
const SUCCESSOR_PID: u64 = 4_002;

fn handle_for(
    store: &Arc<InMemoryStore>,
    workflow_id: &WorkflowId,
    run_id: &RunId,
    pid: u64,
    cached_status: WorkflowStatus,
) -> WorkflowHandle {
    WorkflowHandle::new(WorkflowHandleParts {
        workflow_id: workflow_id.clone(),
        run_id: run_id.clone(),
        pid,
        workflow_type: "checkout".to_owned(),
        namespace: String::from("default"),
        loaded_version: ContentHash::from_bytes([9; 32]),
        cached_status,
        residency: HandleResidency::Resident,
        recorder: Recorder::resume_at(workflow_id.clone(), Arc::clone(store) as _, 0),
        completion: CompletionNotifier::new(),
    })
}

fn seam(registry: &Arc<Registry>) -> QueryMailboxEngine {
    // `resolve_workflow` reads only the registry, so the other two seats are
    // deliberately empty rather than faked: a stub here could be consulted by a
    // future edit without the test noticing, and an empty `Weak` cannot be.
    QueryMailboxEngine::new(Arc::clone(registry), Weak::new(), Weak::new())
}

fn live_seam() -> Result<(Arc<RuntimeHandle>, QueryMailboxEngine), crate::EngineError> {
    let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(
        Some(1),
        crate::runtime::config::TEST_STOP_DRAIN_TIMEOUT,
    ))?);
    let mailbox = QueryMailboxEngine::new(
        Arc::new(Registry::default()),
        Arc::downgrade(runtime.nif_state()),
        Arc::downgrade(&runtime),
    );
    Ok((runtime, mailbox))
}

/// The successor answers, not whichever handle the map scan yielded.
#[test]
fn a_continued_workflow_resolves_to_its_current_run() -> TestResult {
    let registry = Arc::new(Registry::default());
    let store = Arc::new(InMemoryStore::default());
    let workflow_id = WorkflowId::new_v4();
    let predecessor_run = RunId::new_v4();
    let successor_run = RunId::new_v4();

    // Insertion order is the live continue-as-new order: the predecessor was
    // registered long before the successor was started.
    registry.insert(
        (workflow_id.clone(), predecessor_run.clone()),
        handle_for(
            &store,
            &workflow_id,
            &predecessor_run,
            PREDECESSOR_PID,
            WorkflowStatus::ContinuedAsNew,
        ),
    )?;
    registry.insert(
        (workflow_id.clone(), successor_run.clone()),
        handle_for(
            &store,
            &workflow_id,
            &successor_run,
            SUCCESSOR_PID,
            WorkflowStatus::Running,
        ),
    )?;

    // Asserting the SUCCESSOR'S PID, not merely `Resident`: the predecessor is
    // also resident here, so a residency-only assertion would pass on the
    // defect whenever the scan happened to pick the predecessor.
    assert_eq!(
        seam(&registry).resolve_workflow(&workflow_id)?,
        WorkflowResidency::Resident(WorkflowProcessHandle::new(SUCCESSOR_PID)),
        "the seam must name the current run, not whichever handle a HashMap scan yielded"
    );
    Ok(())
}

/// A workflow whose current run has been removed is `Unknown`, even while a
/// predecessor's handle is still registered.
///
/// This is the ONE deliberate behaviour change in #105 and it is pinned rather
/// than left to be discovered. The replaced scan found the stale predecessor,
/// read its terminal `cached_status`, and answered `Terminal` — which
/// `QueryService` maps to `QueryError::NotRunning`. Consulting the index
/// answers `Unknown`, which maps to `QueryError::Unknown`. Both REFUSE the
/// query; `Unknown` is the accurate one when no run is current.
///
/// Reachable through the public API exactly as written: the nif continue-as-new
/// path never removes the predecessor's handle, so once the successor is
/// removed, `forget_live_index_entry` drops the index entry it owned and leaves
/// a live handle behind it.
#[test]
fn a_stale_predecessor_handle_without_an_index_entry_is_unknown() -> TestResult {
    let registry = Arc::new(Registry::default());
    let store = Arc::new(InMemoryStore::default());
    let workflow_id = WorkflowId::new_v4();
    let predecessor_run = RunId::new_v4();
    let successor_run = RunId::new_v4();

    registry.insert(
        (workflow_id.clone(), predecessor_run.clone()),
        handle_for(
            &store,
            &workflow_id,
            &predecessor_run,
            PREDECESSOR_PID,
            WorkflowStatus::ContinuedAsNew,
        ),
    )?;
    registry.insert(
        (workflow_id.clone(), successor_run.clone()),
        handle_for(
            &store,
            &workflow_id,
            &successor_run,
            SUCCESSOR_PID,
            WorkflowStatus::Running,
        ),
    )?;
    registry.remove(&workflow_id, &successor_run)?;

    // The predecessor's handle is still registered — the premise of the test,
    // asserted rather than assumed, because a `remove` that also dropped it
    // would make this case vacuous while staying green.
    assert!(
        registry.get(&workflow_id, &predecessor_run)?.is_some(),
        "premise: the predecessor's handle outlives the successor's removal"
    );
    assert_eq!(
        seam(&registry).resolve_workflow(&workflow_id)?,
        WorkflowResidency::Unknown,
        "no current run means Unknown, not the stale predecessor's Terminal"
    );
    Ok(())
}

/// The live-run index is last-writer-wins by CALL ORDER, so registration order
/// is load-bearing for this seam.
///
/// ⚠️ THIS PINS A PRECONDITION, NOT A BLESSING. `Registry::insert` upserts the
/// index unconditionally, and its comment reads "the newest run for a workflow
/// id wins" — which is only the same thing while runs are inserted in age
/// order. There is no way to recover recency from the ids themselves: `RunId`
/// is a v4 UUID and carries no ordering. Call order is the only signal
/// available, so this is inherent rather than a defect with a fix.
///
/// Live continue-as-new cannot produce the reversed order — the predecessor is
/// registered long before the successor exists. This test exists so that if any
/// future path (recovery, resurrection, an out-of-order sweep) ever registers a
/// superseded run last, the dependency is already written down at the seam that
/// relies on it instead of being rediscovered from a wrong answer in
/// production.
#[test]
fn the_live_run_index_follows_registration_order() -> TestResult {
    let registry = Arc::new(Registry::default());
    let store = Arc::new(InMemoryStore::default());
    let workflow_id = WorkflowId::new_v4();
    let predecessor_run = RunId::new_v4();
    let successor_run = RunId::new_v4();

    // Reversed against the live order: the successor is registered FIRST.
    registry.insert(
        (workflow_id.clone(), successor_run.clone()),
        handle_for(
            &store,
            &workflow_id,
            &successor_run,
            SUCCESSOR_PID,
            WorkflowStatus::Running,
        ),
    )?;
    registry.insert(
        (workflow_id.clone(), predecessor_run.clone()),
        handle_for(
            &store,
            &workflow_id,
            &predecessor_run,
            PREDECESSOR_PID,
            WorkflowStatus::ContinuedAsNew,
        ),
    )?;

    assert_eq!(
        registry.live_run_pid(&workflow_id)?,
        Some((predecessor_run, PREDECESSOR_PID)),
        "the index names the LAST run registered, whatever its age"
    );
    assert_eq!(
        seam(&registry).resolve_workflow(&workflow_id)?,
        WorkflowResidency::Terminal,
        "so a superseded run registered last would be the run this seam answers for"
    );
    Ok(())
}

/// Exit cleanup removes the registration before beamr necessarily retires the
/// pid. That forced ordering is completion, not an unknown query name.
#[test]
fn cleanup_started_while_pid_is_live_drops_the_query_reply() -> TestResult {
    let (runtime, mailbox) = live_seam()?;
    let pid = runtime.spawn_test_process()?;
    let state = runtime.nif_state();
    register_query_impl(state, "state", "{}", Some(pid))?;
    assert!(
        is_query_registered(state, pid, "state")?,
        "fixture control: the query handler must be registered before cleanup"
    );

    // Force the exact monitor ordering without timing: cleanup has removed the
    // handler and stamped the tombstone, while beamr still retains the live pid.
    state.cleanup_process(pid);
    assert!(
        runtime.is_live(pid),
        "fixture control: cleanup must precede scheduler pid retirement"
    );
    assert!(
        runtime.process_cleanup_started(pid),
        "fixture control: cleanup must stamp the exit tombstone"
    );
    assert!(
        !is_query_registered(state, pid, "state")?,
        "fixture control: cleanup must remove the registered handler"
    );

    let (reply_to, reply_from) = tokio::sync::oneshot::channel();
    mailbox.deliver_workflow_message(
        WorkflowProcessHandle::new(pid),
        WorkflowMailboxMessage::Query {
            name: "state".to_owned(),
            payload: Payload::new(ContentType::Json, b"{}".to_vec()),
            reply_to,
        },
    )?;
    assert_eq!(
        reply_from.blocking_recv()?,
        Err(QueryError::ReplyDropped),
        "completion cleanup must yield ReplyDropped, never UnknownQuery"
    );
    runtime.shutdown()?;
    Ok(())
}

/// A live process whose handler name never existed still reports author error.
#[test]
fn live_pid_without_cleanup_reports_unknown_query() -> TestResult {
    let (runtime, mailbox) = live_seam()?;
    let pid = runtime.spawn_test_process()?;
    assert!(runtime.is_live(pid));
    assert!(!runtime.process_cleanup_started(pid));

    let (reply_to, reply_from) = tokio::sync::oneshot::channel();
    mailbox.deliver_workflow_message(
        WorkflowProcessHandle::new(pid),
        WorkflowMailboxMessage::Query {
            name: "missing".to_owned(),
            payload: Payload::new(ContentType::Json, b"{}".to_vec()),
            reply_to,
        },
    )?;
    assert_eq!(
        reply_from.blocking_recv()?,
        Err(QueryError::UnknownQuery("missing".to_owned()))
    );
    runtime.shutdown()?;
    Ok(())
}

/// How long a test waits on the exit drainer: to reach its pause point, and
/// to publish a terminal once it is running.
const DRAINER_TIMEOUT: Duration = Duration::from_secs(10);

/// Block until the exit registry holds the terminal of a process that ended.
///
/// The wait is the record's own publication signal. Nothing in this fixture
/// retires the record — retirement follows a monitor callback, an abort job
/// or an activity outcome read, and none is installed here — so the terminal
/// is still held when the wait returns.
fn wait_for_recorded_terminal(runtime: &RuntimeHandle, pid: u64) -> TestResult {
    let deadline = std::time::Instant::now() + DRAINER_TIMEOUT;
    while std::time::Instant::now() < deadline {
        if runtime.process_exits.has_terminal(pid)? {
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(5));
    }
    Err(format!("the exit registry never recorded the terminal of pid {pid}").into())
}

fn query_message(
    name: &str,
) -> (
    WorkflowMailboxMessage,
    tokio::sync::oneshot::Receiver<Result<Payload, QueryError>>,
) {
    let (reply_to, reply_from) = tokio::sync::oneshot::channel();
    (
        WorkflowMailboxMessage::Query {
            name: name.to_owned(),
            payload: Payload::new(ContentType::Json, b"{}".to_vec()),
            reply_to,
        },
        reply_from,
    )
}

/// The exit registry may record the terminal before the asynchronous Aion
/// cleanup callback runs. That record is an ending on both drop sites.
#[test]
fn recorded_terminal_before_cleanup_drops_the_query_reply() -> TestResult {
    let (runtime, mailbox) = live_seam()?;
    let pid = runtime.spawn_test_process()?;
    register_query_impl(runtime.nif_state(), "state", "{}", Some(pid))?;
    runtime.cancel_pid(pid)?;
    wait_for_recorded_terminal(&runtime, pid)?;
    assert!(
        !runtime.process_cleanup_started(pid),
        "fixture control: Aion cleanup must not have started without a monitor"
    );
    assert_eq!(
        runtime.classify_undelivered_wake(pid)?,
        UndeliveredWake::ProcessEnded
    );

    let (message, reply_from) = query_message("missing");
    mailbox.deliver_workflow_message(WorkflowProcessHandle::new(pid), message)?;
    assert_eq!(
        reply_from.blocking_recv()?,
        Err(QueryError::ReplyDropped),
        "an unregistered name on a recorded ending must be ReplyDropped"
    );

    let (message, reply_from) = query_message("state");
    mailbox.deliver_workflow_message(WorkflowProcessHandle::new(pid), message)?;
    assert!(
        reply_from.blocking_recv().is_err(),
        "a wake refusal on a recorded ending must drop the parked reply sender, \
         which the query service reads as ReplyDropped"
    );
    runtime.shutdown()?;
    Ok(())
}

/// A pid that is absent from the process table, with no cleanup tombstone and
/// no terminal in the exit registry, carries no record of an ending. Neither
/// drop site may answer it as a workflow that ended before answering, and the
/// unregistered-name site may not answer it as the author's error either.
///
/// The drainer is held for the whole window here, so the exit never publishes
/// and every site reports the exit-in-flight fault by name. Released, the same
/// pid settles as the completion it always was.
#[test]
fn absent_pid_without_a_recorded_ending_is_never_reply_dropped() -> TestResult {
    let (runtime, mailbox) = live_seam()?;
    runtime.process_exits.pause_for_test();
    runtime
        .process_exits
        .wait_for_pause_for_test(DRAINER_TIMEOUT)?;
    let pid = runtime.spawn_test_process()?;
    register_query_impl(runtime.nif_state(), "state", "{}", Some(pid))?;
    runtime.cancel_pid(pid)?;

    // Everything observed while the drainer is held is collected first and
    // asserted after the release, so a failed assertion never strands it.
    let live = runtime.is_live(pid);
    let cleanup_started = runtime.process_cleanup_started(pid);
    let terminal_recorded = runtime.process_exits.has_terminal(pid);
    let classified = runtime.classify_undelivered_wake(pid);
    let (message, unregistered_reply) = query_message("missing");
    let unregistered_delivery =
        mailbox.deliver_workflow_message(WorkflowProcessHandle::new(pid), message);
    let unregistered_reply = unregistered_reply.blocking_recv();
    let (message, registered_reply) = query_message("state");
    let registered_delivery =
        mailbox.deliver_workflow_message(WorkflowProcessHandle::new(pid), message);
    let registered_reply = registered_reply.blocking_recv();
    runtime.process_exits.release_for_test();

    assert!(
        !live,
        "fixture control: the pid must be absent from the table"
    );
    assert!(
        !cleanup_started,
        "fixture control: no cleanup tombstone may exist without a monitor"
    );
    assert!(
        !terminal_recorded?,
        "fixture control: the held drainer must not have recorded a terminal"
    );
    assert_eq!(
        classified?,
        UndeliveredWake::ExitInFlight,
        "an exit that publishes nothing inside the readiness window is in flight, \
         never a completion and never nothing at all"
    );

    unregistered_delivery?;
    // An exit in flight is neither an ending nor an author error:
    // `UnknownQuery` rides `unknown_query` to a 400 / InvalidArgument, a
    // non-retryable verdict about a query name over a process that is no
    // longer there. The engine fault rides `backend`, which the caller can
    // re-issue once the exit lands.
    let unregistered_reason = match unregistered_reply? {
        Err(QueryError::Engine(EngineSeamError::Delivery { reason })) => reason,
        other => {
            return Err(format!(
                "an unregistered name over an exit in flight must be the engine fault, got: \
                 {other:?}"
            )
            .into());
        }
    };
    assert!(
        unregistered_reason.contains("exit still in flight")
            && unregistered_reason.contains("within the readiness window"),
        "the engine fault must name both facts it rests on, got: {unregistered_reason}"
    );

    let refusal = match registered_delivery {
        Ok(()) => {
            return Err("a wake refusal with no recorded ending was swallowed as delivered".into());
        }
        Err(refusal) => refusal.to_string(),
    };
    assert!(
        refusal.contains("query wake marker delivery failed")
            && refusal.contains("exit still in flight")
            && refusal.contains("within the readiness window"),
        "the delivery refusal must be surfaced with its cause and both facts, got: {refusal}"
    );
    assert!(
        registered_reply.is_err(),
        "the rolled-back reply sender must be dropped, never left parked"
    );

    // Once the registry records the ending, the same query is a completion.
    wait_for_recorded_terminal(&runtime, pid)?;
    let (message, reply_from) = query_message("missing");
    mailbox.deliver_workflow_message(WorkflowProcessHandle::new(pid), message)?;
    assert_eq!(reply_from.blocking_recv()?, Err(QueryError::ReplyDropped));
    runtime.shutdown()?;
    Ok(())
}

/// The wake-marker failure arm uses the same completion discriminator as the
/// absent-registration arm. The low-level beamr enqueue has no deterministic
/// live-pid refusal seam, so this pins the shared classifier at that boundary.
#[test]
fn cleanup_started_while_pid_is_live_classifies_wake_failure_as_completion() -> TestResult {
    let (runtime, _mailbox) = live_seam()?;
    let pid = runtime.spawn_test_process()?;
    runtime.nif_state().cleanup_process(pid);
    assert!(
        runtime.is_live(pid),
        "fixture control: cleanup must precede scheduler pid retirement"
    );
    assert!(runtime.process_cleanup_started(pid));
    assert_eq!(
        runtime.classify_undelivered_wake(pid)?,
        UndeliveredWake::ProcessEnded,
        "wake-marker failure after cleanup starts must drop the reply, never report an engine fault"
    );
    runtime.shutdown()?;
    Ok(())
}