aion/runtime/

handle.rs

//! `RuntimeHandle` spawn, register, cancel, and shutdown support.

use std::sync::{Arc, Mutex};

use aion_core::{ActivityError, Payload};
use beamr::atom::AtomTable;
use beamr::module::ModuleRegistry;
use beamr::native::BifRegistryImpl;
use beamr::process::ExitReason;
use beamr::scheduler::{Scheduler, SchedulerConfig};
use beamr::term::Term;

use crate::error::EngineError;

use super::config::{RuntimeConfig, SignalDeliveryConfig};
#[cfg(test)]
use super::nif::Mfa;
use super::nif::NifRegistration;
use super::payload::payload_to_term;

use self::registration::{nif_registration_error, register_all_bifs};

/// Local BEAM process identifier exposed by the runtime boundary.
pub type Pid = u64;

type RetainedHeap = Box<[u64]>;
type RetainedHeaps = Vec<RetainedHeap>;
type RetainedSpawnHeaps = Arc<dashmap::DashMap<Pid, Mutex<RetainedHeaps>>>;

/// Runtime-owned workflow or activity input terms.
///
/// The wrapper keeps the beamr term representation inside the runtime module
/// while later lifecycle and payload code decide how durable payloads become VM
/// terms.
#[derive(Debug, Default, Eq, PartialEq)]
pub struct RuntimeInput {
    terms: Vec<Term>,
    heaps: RetainedHeaps,
}

impl RuntimeInput {
    /// Convert one durable payload into the single BEAM argument used by
    /// in-VM activity dispatch.
    ///
    /// The runtime boundary owns this representation. JSON payloads are passed
    /// as BEAM binary terms and any boxed host heap backing those terms is
    /// retained until the spawned process is observed exiting or cancelled.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when a JSON number does not fit in an
    /// immediate small integer.
    pub fn from_payload(payload: &Payload) -> Result<Self, EngineError> {
        let (term, heaps) = payload_to_term(payload)?.into_parts();
        Ok(Self {
            terms: vec![term],
            heaps,
        })
    }

    /// Number of terms supplied to the BEAM entrypoint.
    #[must_use]
    pub fn arity(&self) -> u8 {
        u8::try_from(self.terms.len()).unwrap_or(u8::MAX)
    }

    fn into_spawn_parts(self) -> (Vec<Term>, RetainedHeaps) {
        (self.terms, self.heaps)
    }
}

/// Handle to the embedded beamr scheduler and code-server state.
pub struct RuntimeHandle {
    pub(super) scheduler: Arc<Scheduler>,
    pub(super) atom_table: Arc<AtomTable>,
    pub(super) module_registry: Arc<ModuleRegistry>,
    pub(super) native_registry: Arc<BifRegistryImpl>,
    nif_state: Arc<super::nif_state::EngineNifState>,
    activity_results: Arc<dashmap::DashMap<(Pid, Pid), Payload>>,
    activity_errors: Arc<dashmap::DashMap<(Pid, Pid), ActivityError>>,
    /// Per-workflow synchronization for retained activity delivery and death draining.
    ///
    /// Each workflow has an independent gate, so an exited process that remains
    /// in beamr's process table cannot block unrelated workflows. A
    /// dead gate remains until process-table removal is observed, preventing
    /// delivery from inserting behind that workflow's death sweep.
    activity_delivery_gates: dashmap::DashMap<Pid, Arc<activity_delivery::ActivityDeliveryGate>>,
    /// One-based delivery attempt that produced a retained two-phase activity
    /// outcome, keyed like [`Self::activity_results`] / [`Self::activity_errors`]
    /// (#197). Retained in the same gate transaction as the final outcome and
    /// taken atomically with that outcome, so recorded terminals carry the
    /// genuine attempt. Absence means the first delivery (paths that never
    /// retry — outbox re-delivery and in-VM execution — retain nothing).
    activity_delivery_attempts: Arc<dashmap::DashMap<(Pid, Pid), u32>>,
    #[cfg(test)]
    activity_delivery_test_seams: activity_delivery::ActivityDeliveryTestSeams,
    /// Live in-VM activity children per workflow pid.
    ///
    /// A BEAM link tears a child down when its workflow dies ABNORMALLY, but
    /// a `Normal` exit never propagates through links (classic BEAM
    /// semantics), so a workflow that completes while an in-VM runner is
    /// still executing — e.g. after a `with_timeout` expiry abandoned the
    /// await — would orphan the child and its completion waiter forever. The
    /// workflow process monitor kills children still registered here when the
    /// workflow exits (for any reason), and [`Self::shutdown`] kills every
    /// remaining child so no waiter outlives the scheduler.
    in_vm_children: Arc<dashmap::DashMap<Pid, std::collections::HashSet<Pid>>>,
    registered_nif_modules: Arc<dashmap::DashSet<String>>,
    spawn_heaps: RetainedSpawnHeaps,
    signal_delivery: SignalDeliveryConfig,
    /// Flag gating the durable-outbox fan-out dispatch path; read by
    /// `nif_collect.rs` to route fresh fan-out members and completions.
    outbox_enabled: bool,
    /// Bounded follow-up wakes for delivered mailbox markers, healing
    /// beamr 0.4.9's lost-wakeup window (see [`super::wake_confirm`]).
    pub(super) wake_confirmer: super::wake_confirm::WakeConfirmer,
    /// Per-pid outcomes established before top-level process pids are published.
    pub(super) process_exits: Arc<super::process_exit::ProcessExitRegistry>,
    /// Runtime-provisioned owner for every process abort and shared cleanup.
    pub(super) cleanup_executor: super::cleanup_executor::CleanupExecutor,
    /// Identity-bearing abort jobs retained for deduplicated retries.
    pub(super) abort_jobs: dashmap::DashMap<Pid, Arc<super::monitor::UnmonitoredProcessAbortJob>>,
}

impl RuntimeHandle {
    /// Construct and start an embedded runtime from builder-supplied config.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when beamr cannot start its scheduler.
    /// Returns [`EngineError::Gate3BifReplacementMissing`] if beamr's complete
    /// Gate-3 table no longer contains a required tracked fun-spawn BIF.
    pub fn new(config: RuntimeConfig) -> Result<Self, EngineError> {
        let atom_table = Arc::new(AtomTable::with_common_atoms());
        let module_registry = Arc::new(ModuleRegistry::new());
        // One NIF state per runtime instance, recovered by every native call
        // through beamr's NIF private data — never process-wide globals.
        let nif_state = Arc::new(super::nif_state::EngineNifState::default());
        let scheduler_config = SchedulerConfig {
            thread_count: config.thread_count,
            nif_private_data: Some(Arc::clone(&nif_state) as _),
            ..Default::default()
        };
        let native_registry = Arc::new(BifRegistryImpl::new());
        register_all_bifs(&native_registry, &atom_table, &nif_state)?;
        let scheduler = Arc::new(
            Scheduler::with_code_server(
                scheduler_config,
                Arc::clone(&module_registry),
                Arc::clone(&atom_table),
                Arc::clone(&native_registry),
            )
            .map_err(runtime_error_from_display)?,
        );
        let wake_confirmer = super::wake_confirm::WakeConfirmer::new(config.signal_delivery)?;
        let shutdown_timeout = config.signal_delivery.cleanup_shutdown_timeout();
        let cleanup_executor = super::cleanup_executor::CleanupExecutor::new(
            config.signal_delivery.max_enqueue_attempts as usize,
            shutdown_timeout,
        )?;
        // Claim beamr's singleton stream before this scheduler can publish a pid
        // through any RuntimeHandle spawn API.
        let process_exits = super::process_exit::ProcessExitRegistry::new(
            Arc::clone(&scheduler),
            shutdown_timeout,
            config.signal_delivery.max_enqueue_attempts as usize,
        )?;
        nif_state.set_process_exit_registry(&process_exits)?;

        Ok(Self {
            scheduler,
            atom_table,
            module_registry,
            native_registry,
            nif_state,
            activity_results: Arc::new(dashmap::DashMap::new()),
            activity_errors: Arc::new(dashmap::DashMap::new()),
            activity_delivery_gates: dashmap::DashMap::new(),
            activity_delivery_attempts: Arc::new(dashmap::DashMap::new()),
            #[cfg(test)]
            activity_delivery_test_seams: activity_delivery::ActivityDeliveryTestSeams::default(),
            in_vm_children: Arc::new(dashmap::DashMap::new()),
            registered_nif_modules: Arc::new(dashmap::DashSet::new()),
            spawn_heaps: Arc::new(dashmap::DashMap::new()),
            signal_delivery: config.signal_delivery,
            outbox_enabled: config.outbox_enabled,
            wake_confirmer,
            process_exits,
            cleanup_executor,
            abort_jobs: dashmap::DashMap::new(),
        })
    }

    /// This runtime instance's engine-scoped NIF state.
    pub(crate) fn nif_state(&self) -> &Arc<super::nif_state::EngineNifState> {
        &self.nif_state
    }

    /// Builder-supplied delivery/readiness policy for spawn-window waits.
    pub(crate) fn signal_delivery(&self) -> SignalDeliveryConfig {
        self.signal_delivery
    }

    /// Whether the durable-outbox fan-out dispatch path is enabled.
    ///
    /// Read by `nif_collect.rs` to route fresh fan-out members through the
    /// durable outbox and record completions via the dedup primitive.
    pub(crate) fn outbox_enabled(&self) -> bool {
        self.outbox_enabled
    }

    /// Install collected NIF entries into beamr's native registry.
    ///
    /// Consumes the registration collection so no caller can append more entries
    /// after this installation step. Callers must invoke this before loading and
    /// spawning workflow modules whose imports depend on these NIFs.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::NifRegistration`] when beamr rejects an entry,
    /// including duplicate module/function/arity registrations.
    pub fn install_nifs(&self, registration: NifRegistration) -> Result<(), EngineError> {
        for entry in registration.into_entries() {
            let mfa = entry.mfa;
            let module = self.atom_table.intern(&mfa.module);
            let function = self.atom_table.intern(&mfa.function);
            let capability = beamr::native::Capability::ExternalIo;
            let result = if entry.is_dirty {
                self.native_registry.register_dirty(
                    module,
                    function,
                    mfa.arity,
                    entry.function,
                    beamr::scheduler::dirty::DirtySchedulerKind::Cpu,
                    capability,
                )
            } else {
                self.native_registry.register(
                    module,
                    function,
                    mfa.arity,
                    entry.function,
                    capability,
                )
            };
            result.map_err(|error| nif_registration_error(&mfa, error))?;
            self.registered_nif_modules.insert(mfa.module);
        }

        Ok(())
    }

    /// Return module names that have registered NIFs and should not be
    /// content-hash renamed during package loading.
    #[must_use]
    pub fn registered_nif_modules(&self) -> Vec<String> {
        let mut module_names: Vec<_> = self
            .registered_nif_modules
            .iter()
            .map(|module_name| module_name.key().clone())
            .collect();
        module_names.sort();
        module_names
    }

    /// Spawn a top-level workflow process at a deployed module/function entrypoint.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when the module/function/arity cannot be
    /// resolved or beamr rejects the spawn request.
    pub fn spawn_workflow(
        &self,
        deployed_module: &str,
        function: &str,
        input: RuntimeInput,
    ) -> Result<Pid, EngineError> {
        self.spawn_process(deployed_module, function, input)
    }

    /// Spawn a top-level workflow process with trap-exit enabled before it runs.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when the module/function/arity cannot be
    /// resolved or beamr rejects the spawn request.
    pub fn spawn_workflow_trapping(
        &self,
        deployed_module: &str,
        function: &str,
        input: RuntimeInput,
    ) -> Result<Pid, EngineError> {
        self.release_dead_spawn_heaps();
        let module = self.atom_table.intern(deployed_module);
        let function = self.atom_table.intern(function);
        let (terms, heaps) = input.into_spawn_parts();
        let pid = self.spawn_with_exit_ownership(|| {
            self.scheduler
                .spawn_trap_exit(module, function, terms)
                .map_err(runtime_error_from_display)
        })?;
        self.retain_spawn_heaps(pid, heaps);
        Ok(pid)
    }

    /// Spawn an activity child process linked to its workflow parent.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when the parent process is not live, the
    /// module/function/arity cannot be resolved, or beamr rejects the linked
    /// spawn request.
    pub fn spawn_activity(
        &self,
        parent_pid: Pid,
        deployed_module: &str,
        function: &str,
        input: RuntimeInput,
    ) -> Result<Pid, EngineError> {
        self.release_dead_spawn_heaps();
        self.ensure_live_pid(parent_pid)?;
        self.wait_for_process_ready(parent_pid)?;
        let module = self.atom_table.intern(deployed_module);
        let function_atom = self.atom_table.intern(function);
        let (terms, heaps) = input.into_spawn_parts();
        let pid = self.spawn_with_exit_ownership(|| {
            self.scheduler
                .spawn_link(parent_pid, module, function_atom, terms)
                .map_err(runtime_error_from_display)
        })?;
        self.retain_spawn_heaps(pid, heaps);
        Ok(pid)
    }

    /// Spawn an in-VM activity child process linked to its workflow parent,
    /// running a zero-arity closure (the SDK-composed runner thunk).
    ///
    /// beamr deep-copies the closure's environment into the child's own heap
    /// before the child becomes runnable (`Scheduler::spawn_link_closure`), so
    /// no spawn heap is retained here and the caller's heap may move (GC) the
    /// moment this returns. The child does not trap exits: workflow
    /// cancellation propagates through the link, and an abnormal child exit is
    /// observed by the in-VM completion watcher via [`Self::in_vm_child_outcome`].
    ///
    /// No parent readiness wait is performed: the only production caller is
    /// the dispatch NIF executing ON the parent process, which is therefore
    /// already materialized (a readiness poll on an `Executing` slot would be
    /// pointless at best).
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when the parent is not live, the term
    /// is not a zero-arity closure, or its module cannot be resolved.
    pub fn spawn_activity_closure(
        &self,
        parent_pid: Pid,
        closure_term: Term,
    ) -> Result<Pid, EngineError> {
        self.release_dead_spawn_heaps();
        self.ensure_live_pid(parent_pid)?;
        let pid = self.spawn_with_exit_ownership(|| {
            self.scheduler
                .spawn_link_closure(parent_pid, closure_term)
                .map_err(runtime_error_from_display)
        })?;
        self.in_vm_children
            .entry(parent_pid)
            .or_default()
            .insert(pid);
        // Close the external-kill registration race: if the workflow died
        // between the liveness check above and this registration, the
        // monitor's `kill_in_vm_children` sweep may already have run (and
        // beamr's link may never have been established — a caller that died
        // mid-spawn yields an UNLINKED child), which would leave a hanging
        // runner alive until engine shutdown. Re-checking AFTER registration
        // makes both orderings safe: a parent death after this point observes
        // the registration and is swept by the monitor; a death before it is
        // torn down here (both kill paths are idempotent — the sweep guards
        // with `is_live`).
        if !self.is_live(parent_pid) {
            self.kill_in_vm_children(parent_pid);
            return Err(EngineError::Runtime {
                reason: format!(
                    "in-vm activity child spawn: parent workflow process {parent_pid} exited during spawn"
                ),
            });
        }
        Ok(pid)
    }

    /// Drop a finished in-VM child from its workflow's teardown set (called
    /// by the completion waiter once the child's cached outcome is decoded).
    pub(crate) fn deregister_in_vm_child(&self, parent_pid: Pid, child_pid: Pid) {
        if let Some(mut children) = self.in_vm_children.get_mut(&parent_pid) {
            children.remove(&child_pid);
        }
        self.in_vm_children
            .remove_if(&parent_pid, |_, children| children.is_empty());
    }

    /// Kill every in-VM activity child still registered for `workflow_pid`.
    ///
    /// Invoked by the workflow process monitor on workflow exit: a `Normal`
    /// exit does not propagate through BEAM links, so a completed workflow
    /// would otherwise orphan a still-running runner and its completion waiter.
    /// Killing publishes the child's durable outcome, which wakes that waiter;
    /// delivery to the dead workflow is refused and nothing is retained.
    pub(crate) fn kill_in_vm_children(&self, workflow_pid: Pid) {
        let Some((_, children)) = self.in_vm_children.remove(&workflow_pid) else {
            return;
        };
        for child_pid in children {
            if self.is_live(child_pid) {
                tracing::debug!(
                    workflow_pid,
                    child_pid,
                    "killing orphaned in-vm activity child on workflow exit"
                );
                self.scheduler
                    .terminate_process(child_pid, ExitReason::Kill);
            }
            self.release_spawn_heaps(child_pid);
        }
    }

    /// Return whether the registered native activity entry is dirty for arity 1.
    #[must_use]
    pub fn is_dirty(&self, module: &str, function: &str) -> bool {
        self.is_dirty_with_arity(module, function, 1)
    }

    /// Return whether the registered native entry is dirty for the supplied arity.
    #[must_use]
    pub fn is_dirty_with_arity(&self, module: &str, function: &str, arity: u8) -> bool {
        let module = self.atom_table.intern(module);
        let function = self.atom_table.intern(function);
        self.native_registry
            .lookup(module, function, arity)
            .is_some_and(|entry| entry.dirty_kind.is_some())
    }

    /// Cancel a live process by PID.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when `pid` is not live.
    pub fn cancel_pid(&self, pid: Pid) -> Result<(), EngineError> {
        self.ensure_live_pid(pid)?;
        self.scheduler.terminate_process(pid, ExitReason::Kill);
        self.release_spawn_heaps(pid);
        Ok(())
    }

    /// Set a live process' trap-exit flag, returning the previous value.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when `pid` is not live.
    pub fn set_trap_exit(&self, pid: Pid, value: bool) -> Result<bool, EngineError> {
        self.scheduler
            .set_trap_exit(pid, value)
            .map_err(runtime_error_from_display)
    }

    /// Return true when `pid` is currently live.
    #[must_use]
    pub fn is_live(&self, pid: Pid) -> bool {
        self.scheduler.process_table().get(pid).is_some()
    }

    /// Return a live process' trap-exit flag.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when `pid` is not live.
    pub fn trap_exit(&self, pid: Pid) -> Result<bool, EngineError> {
        self.scheduler
            .trap_exit(pid)
            .ok_or_else(|| runtime_error(format!("process {pid} is not live")))
    }

    /// Return true when two live processes have a bidirectional link.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when either process is not live.
    pub fn is_linked(&self, left: Pid, right: Pid) -> Result<bool, EngineError> {
        self.ensure_live_pid(left)?;
        self.ensure_live_pid(right)?;
        Ok(self.scheduler.is_linked(left, right))
    }

    /// Shut down the embedded scheduler and wait for worker threads to stop.
    ///
    /// # Errors
    ///
    /// Currently infallible; reserved for typed runtime shutdown failures.
    pub fn shutdown(&self) -> Result<(), EngineError> {
        // Kill every still-live in-VM activity child before stopping the
        // scheduler so the singleton drainer can capture each durable outcome.
        let workflow_pids: Vec<Pid> = self
            .in_vm_children
            .iter()
            .map(|entry| *entry.key())
            .collect();
        for workflow_pid in workflow_pids {
            self.kill_in_vm_children(workflow_pid);
        }
        for pid in self.process_exits.begin_shutdown()? {
            if self.is_live(pid) {
                self.scheduler.terminate_process(pid, ExitReason::Kill);
            }
        }
        // Abort jobs may be blocked waiting for exactly the exits published
        // above, so drain the bounded executor only after every registered pid
        // has been force-unblocked.
        self.cleanup_executor.shutdown()?;
        // Pending wake follow-ups are moot once process observation is closed.
        self.wake_confirmer.shutdown();
        self.process_exits.close_and_join_all()?;
        self.scheduler.shutdown();
        self.spawn_heaps.clear();
        Ok(())
    }

    fn spawn_process(
        &self,
        deployed_module: &str,
        function: &str,
        input: RuntimeInput,
    ) -> Result<Pid, EngineError> {
        self.release_dead_spawn_heaps();
        let module = self.atom_table.intern(deployed_module);
        let function = self.atom_table.intern(function);
        let (terms, heaps) = input.into_spawn_parts();
        let pid = self.spawn_with_exit_ownership(|| {
            self.scheduler
                .spawn(module, function, terms)
                .map_err(runtime_error_from_display)
        })?;
        self.retain_spawn_heaps(pid, heaps);
        Ok(pid)
    }

    fn retain_spawn_heaps(&self, pid: Pid, heaps: RetainedHeaps) {
        if heaps.is_empty() {
            return;
        }
        self.spawn_heaps.insert(pid, Mutex::new(heaps));
    }

    pub(super) fn release_spawn_heaps(&self, pid: Pid) {
        self.spawn_heaps.remove(&pid);
    }

    fn release_dead_spawn_heaps(&self) {
        let dead_pids: Vec<Pid> = self
            .spawn_heaps
            .iter()
            .filter_map(|entry| {
                let pid = *entry.key();
                self.scheduler
                    .process_table()
                    .get(pid)
                    .is_none()
                    .then_some(pid)
            })
            .collect();
        for pid in dead_pids {
            self.release_spawn_heaps(pid);
        }
    }

    pub(super) fn ensure_live_pid(&self, pid: Pid) -> Result<(), EngineError> {
        if self.scheduler.process_table().get(pid).is_some() {
            Ok(())
        } else {
            Err(runtime_error(format!("process {pid} is not live")))
        }
    }

    #[cfg(test)]
    pub(crate) fn live_processes_for_test(&self) -> usize {
        self.scheduler.process_table().len()
    }

    /// Spawn an inert test process without module code.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when beamr rejects the test spawn.
    #[cfg(test)]
    pub fn spawn_test_process(&self) -> Result<Pid, EngineError> {
        self.spawn_with_exit_ownership(|| Ok(self.scheduler.spawn_test_process(false)))
    }

    /// Spawn an inert test process with explicit trap-exit state.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when beamr rejects the test spawn.
    #[cfg(test)]
    pub fn spawn_test_process_with_trap_exit(&self, trap_exit: bool) -> Result<Pid, EngineError> {
        self.spawn_with_exit_ownership(|| Ok(self.scheduler.spawn_test_process(trap_exit)))
    }

    /// Spawn an inert linked test child without enabling trap-exit on the child.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when the parent is not live or beamr
    /// rejects the linked spawn.
    #[cfg(test)]
    pub fn spawn_linked_test_process(&self, parent_pid: Pid) -> Result<Pid, EngineError> {
        self.ensure_live_pid(parent_pid)?;
        self.spawn_with_exit_ownership(|| {
            self.scheduler
                .spawn_linked_test_process(parent_pid)
                .map_err(runtime_error_from_display)
        })
    }

    /// Return true when a live process has a trapped EXIT message from `source_pid`.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when `target_pid` is not live.
    #[cfg(test)]
    pub fn has_trapped_exit_message(
        &self,
        target_pid: Pid,
        source_pid: Pid,
    ) -> Result<bool, EngineError> {
        self.ensure_live_pid(target_pid)?;
        Ok(self
            .scheduler
            .has_trapped_exit_message(target_pid, source_pid)
            .unwrap_or(false))
    }

    /// Poll until a trapped EXIT message from `source_pid` arrives at `target_pid`.
    ///
    /// beamr delivers exit signals asynchronously after process termination.
    /// Tests that assert on trapped exit messages must wait for delivery.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] if the message does not arrive within 50ms.
    #[cfg(test)]
    pub fn wait_for_trapped_exit(
        &self,
        target_pid: Pid,
        source_pid: Pid,
    ) -> Result<(), EngineError> {
        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(50);
        while std::time::Instant::now() < deadline {
            if self
                .scheduler
                .has_trapped_exit_message(target_pid, source_pid)
                .unwrap_or(false)
            {
                return Ok(());
            }
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
        Err(runtime_error(format!(
            "trapped exit from {source_pid} to {target_pid} did not arrive"
        )))
    }

    /// Terminate a test process with a trappable abnormal reason.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Runtime`] when `pid` is not live.
    #[cfg(test)]
    pub fn terminate_test_process_with_error(&self, pid: Pid) -> Result<(), EngineError> {
        self.ensure_live_pid(pid)?;
        self.scheduler.terminate_process(pid, ExitReason::Error);
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn lookup_native_for_test(
        &self,
        module: &str,
        function: &str,
        arity: u8,
    ) -> Option<beamr::native::NativeEntry> {
        let module = self.atom_table.intern(module);
        let function = self.atom_table.intern(function);
        self.native_registry.lookup(module, function, arity)
    }

    #[cfg(test)]
    pub(crate) fn retained_spawn_heap_count_for_test(&self) -> usize {
        self.release_dead_spawn_heaps();
        self.spawn_heaps.len()
    }
}

fn runtime_error(reason: String) -> EngineError {
    EngineError::Runtime { reason }
}

fn runtime_error_from_display(reason: impl std::fmt::Display) -> EngineError {
    runtime_error(reason.to_string())
}

mod activity_delivery;
mod delivery;
mod process_ownership;
mod readiness;
mod registration;
mod spawn_bifs;

pub(crate) use delivery::InVmChildOutcome;

#[cfg(test)]
#[path = "handle/test_support.rs"]
mod test_support;

#[cfg(test)]
#[path = "handle/tests.rs"]
mod tests;