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>,
    /// Engine-owned executor for background tasks that append to durable
    /// history: child-terminal watchers, spawn recovery, and process-exit
    /// completion retries.
    ///
    /// Owned here rather than by any bridge because a bridge is optional and
    /// these tasks are not. Its shutdown aborts AND awaits, which is what stops
    /// a task outliving the epoch and becoming a second writer against a
    /// successor engine over the same store.
    engine_tasks: Arc<super::engine_tasks::EngineTaskRuntime>,
    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,
    completion_retry: super::config::CompletionRetryConfig,
    /// 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,
            // `None` here is beamr's own default threshold, not a value aion
            // chose. See `RuntimeConfig::jit_threshold` for what a large value
            // does and — importantly — what it does not do.
            jit_threshold: config.jit_threshold,
            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,
            engine_tasks: Arc::new(super::engine_tasks::EngineTaskRuntime::new()?),
            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,
            completion_retry: config.completion_retry,
            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
    }

    /// Builder-supplied backoff ladder for durable completion retries.
    ///
    /// Deliberately not [`Self::signal_delivery`]: that policy bounds a
    /// mailbox-enqueue wait measured in scheduler ticks, and a durable retry
    /// against a failing store is a different question with a different answer.
    pub(crate) fn completion_retry(&self) -> super::config::CompletionRetryConfig {
        self.completion_retry
    }

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

    /// The engine-owned executor for durable background tasks.
    ///
    /// Every caller shares this one instance: the child bridge, spawn recovery
    /// and the process-exit completion retry. A second executor with the same
    /// epoch-close discipline would be the same rule in two places.
    pub(crate) fn engine_tasks(&self) -> Arc<super::engine_tasks::EngineTaskRuntime> {
        Arc::clone(&self.engine_tasks)
    }

    /// Arm the injected process-exit drain failure, so a test can make
    /// [`Self::shutdown`] fail deterministically.
    ///
    /// Exists because none of the drain failures can be produced on demand,
    /// which is how the "every teardown step still runs" property stayed
    /// unpinned.
    ///
    /// 🔴 THE SET IS OPEN, AND TWO EARLIER REVISIONS OF THIS COMMENT CLAIMED
    /// OTHERWISE. The first said they are all timeout-shaped. The second
    /// replaced that with an eleven-variant enumeration — three timeout-shaped,
    /// five poison-shaped, three panic-shaped — and asserted "what they share is
    /// a precondition: each needs a stalled, poisoned or dead worker thread".
    /// **Both are false, and the second is the more dangerous because it looks
    /// exhaustive.**
    ///
    /// `close_and_join_all` ends with
    /// `handle.join().map_err(|_| ProcessExitDrainerPanicked)??` — and the
    /// SECOND `?` propagates the drainer thread's own `Result<(), EngineError>`
    /// verbatim (`runtime/process_exit.rs`). Whatever that thread can return
    /// reaches this function's caller. That includes at least
    /// `ProcessExitEventStreamDisconnected` (`runtime/process_exit_drainer.rs`
    /// — beamr disconnected its publisher) and
    /// `ProcessExitOutcomeMissingAfterEvent` (a beamr contract breach surfaced
    /// through `registry.process_event`). Neither is timeout-, poison- or
    /// panic-shaped, and neither needs a worker thread of ours to be stalled,
    /// poisoned or dead.
    ///
    /// So: **do not enumerate this set, and do not reason from a shared shape or
    /// a shared precondition.** The property that actually holds, and the only
    /// one this seam needs, is that none of them can be arranged by a test
    /// through this type's public surface. `process_exits` stays private; this
    /// is the one named seam, and it is `#[cfg(test)]` so it cannot reach a
    /// shipped binary.
    ///
    /// The injected variant is `ProcessExitRegistryPoisoned` because that is
    /// what the injection point (`begin_shutdown` → `lock_lifecycle`) can
    /// actually raise. A fault wearing a label its own injection site cannot
    /// issue is a fixture modelling a machine that does not exist.
    ///
    /// Crate-visible rather than module-visible because the property it exists
    /// to measure lives at [`crate::Engine::shutdown`], one module over — a seam
    /// only reachable from its own module cannot test the caller that wraps it.
    #[cfg(test)]
    pub(crate) fn force_process_exit_drain_failure(&self) {
        self.process_exits.force_shutdown_failure();
    }

    /// Shut down the embedded scheduler and wait for worker threads to stop.
    ///
    /// # Errors
    ///
    /// Returns the **first** typed failure raised by the process-exit drain
    /// (`begin_shutdown`, `close_and_join_all`) or the cleanup-executor drain.
    /// Those failures are an OPEN set — see
    /// [`Self::force_process_exit_drain_failure`] for why they cannot be
    /// enumerated — and the shape does not matter here, because none of them
    /// short-circuits this function: the error is carried to the end and
    /// returned only after the
    /// engine-task epoch has been closed and the scheduler stopped. See the
    /// comment on `first_error` in the body for why that ordering is a
    /// correctness requirement rather than a tidiness preference.
    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);
        }
        // 🔴 Every fallible drain below RECORDS its failure instead of
        // returning it, so that the epoch close at the end of this function is
        // reached on every path.
        //
        // The epoch close aborts and awaits tasks that append terminal events.
        // A drain timeout is not an unrelated inconvenience — it is precisely
        // the condition under which those tasks are still armed, because one
        // degraded store both stalls the drain and is what the completion
        // retries are waiting on. If a `?` here returned early, the retries
        // would keep running; the operator, seeing a shutdown error, restarts;
        // and the successor engine recovers the same histories while this
        // process is still appending to them — two writers for one workflow,
        // which is the invariant-3 violation this whole mechanism exists to
        // prevent.
        //
        // Nothing downstream would catch it on THIS path either. `Drop for
        // Engine` exists, and what it does is enumerated in exactly one place —
        // its own doc comment, which is the authority; this comment deliberately
        // does not repeat the list, because it has already been stale once (it
        // named two of the three things that drop does, and was written when
        // there were two). What matters here is the shared PROPERTY of every
        // item on that list: none of them await, and none of them run at all
        // while the caller still holds the engine to read this function's `Err`.
        //
        // `EngineTaskRuntime::drop` is no help under the
        // same conditions: an attempt in flight holds its own strong handle, so
        // that backstop is pinned shut for exactly the span of the append it
        // would need to stop. Closing the epoch here is the only mechanism that
        // is guaranteed to run, which is why it does not sit behind a `?`.
        let mut first_error: Option<EngineError> = None;

        match self.process_exits.begin_shutdown() {
            Ok(pids) => {
                for pid in pids {
                    if self.is_live(pid) {
                        self.scheduler.terminate_process(pid, ExitReason::Kill);
                    }
                }
            }
            Err(error) => first_error = Some(error),
        }
        // 🔴 EVERY FAILURE IS REPORTED, EVEN THOUGH ONLY ONE CAN BE RETURNED.
        // Three steps below can each fail independently and the signature can
        // carry one `EngineError`, so a later failure that is not the first
        // would otherwise vanish with no trace at all — a swallowed `Result` in
        // the teardown path of a durable engine, which this codebase forbids
        // outright. `keep_shutdown_error` returns the first and emits every
        // subsequent one at `error` level with the step that produced it.
        // 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.
        keep_shutdown_error(
            &mut first_error,
            "cleanup_executor.shutdown",
            self.cleanup_executor.shutdown(),
        );
        // Pending wake follow-ups are moot once process observation is closed.
        self.wake_confirmer.shutdown();
        keep_shutdown_error(
            &mut first_error,
            "process_exits.close_and_join_all",
            self.process_exits.close_and_join_all(),
        );
        // Close the engine-task epoch here, after every process-exit callback
        // has been drained (so an exit observed during teardown still gets its
        // one attempt) and before the scheduler goes away.
        //
        // This must be on THIS path, not only on the child bridge's. The
        // executor is owned here precisely because completion retries append
        // terminal events on a core lifecycle path, and a core path's epoch
        // close must not depend on whether an optional bridge was installed:
        // a retry still running after this returns could append against a
        // store a successor engine is also recovering (invariant 3).
        //
        // A second `shutdown` call is a no-op: it re-gates and re-sweeps the
        // three already-empty maps `gate_and_abort` covers (`watches`,
        // `spawn_retries`, `completion_retries`), finds the runtime slot empty,
        // and returns without awaiting anything. That makes the bridge's own
        // later call harmless
        // because the two are strictly sequential (`Engine::shutdown` runs this
        // one, then the bridge's). It does NOT mean "returning implies
        // quiescence" for a hypothetical concurrent second caller — only the
        // first call awaits.
        self.engine_tasks.shutdown();
        self.scheduler.shutdown();
        self.spawn_heaps.clear();

        match first_error {
            Some(error) => Err(error),
            None => 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()
    }
}

/// Keep the FIRST teardown failure for the caller, and report every subsequent
/// one rather than dropping it.
///
/// Teardown accumulates instead of failing fast — every step must run, because
/// a skipped one leaves a durable writer armed. The cost of that choice is that
/// more than one step can fail while only one `EngineError` can be returned.
/// Reporting the others here is what keeps "accumulate and continue" from
/// becoming "swallow and continue": the return value carries the first, the log
/// carries the rest, and no failure is lost.
fn keep_shutdown_error(
    first_error: &mut Option<EngineError>,
    step: &'static str,
    result: Result<(), EngineError>,
) {
    let Err(error) = result else {
        return;
    };
    if first_error.is_none() {
        *first_error = Some(error);
        return;
    }
    tracing::error!(
        step,
        error = %error,
        "a further runtime-shutdown step failed after an earlier one; only the first failure \
         can be returned, so this one is reported here"
    );
}

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;