aion/runtime/
monitor.rs

1//! Runtime-owned process monitoring helpers.
2
3use std::sync::Arc;
4use std::sync::atomic::{AtomicBool, Ordering};
5
6use crate::{EngineError, Pid, RuntimeHandle};
7
8use super::outcome::{self, WorkflowProcessOutcome};
9
10/// Handle returned after installing a workflow process monitor.
11#[derive(Clone)]
12pub struct ProcessMonitorHandle {
13    installed: Arc<AtomicBool>,
14}
15
16impl ProcessMonitorHandle {
17    fn installed() -> Self {
18        Self {
19            installed: Arc::new(AtomicBool::new(true)),
20        }
21    }
22
23    /// Returns whether the runtime accepted monitor installation.
24    #[must_use]
25    pub fn is_installed(&self) -> bool {
26        self.installed.load(Ordering::Acquire)
27    }
28}
29
30impl RuntimeHandle {
31    /// Install a runtime-owned monitor that invokes `callback` when `pid` exits.
32    ///
33    /// The callback runs on a dedicated monitor thread outside workflow dirty NIF
34    /// execution. The runtime boundary owns the process wait and BEAM term
35    /// conversion so lifecycle code never imports beamr types.
36    ///
37    /// # Errors
38    ///
39    /// Returns [`EngineError::Runtime`] when `pid` is neither live nor a
40    /// process this runtime previously spawned. A monitored process that
41    /// already exited is accepted: its outcome is still observable through
42    /// the scheduler's exit tombstone, so the callback fires immediately
43    /// instead of the spawn path spuriously failing for fast-completing
44    /// workflows.
45    pub fn monitor_process<F>(
46        self: &Arc<Self>,
47        pid: Pid,
48        callback: F,
49    ) -> Result<ProcessMonitorHandle, EngineError>
50    where
51        F: FnOnce(Result<WorkflowProcessOutcome, EngineError>) + Send + 'static,
52    {
53        self.ensure_monitorable_pid(pid)?;
54        let runtime = Arc::clone(self);
55        std::thread::Builder::new()
56            .name(format!("aion-workflow-monitor-{pid}"))
57            .spawn(move || {
58                let outcome =
59                    outcome::workflow_process_outcome(&runtime.scheduler, &runtime.atom_table, pid);
60                runtime.release_spawn_heaps(pid);
61                runtime.nif_state().cleanup_process(pid);
62                // A Normal workflow exit does not propagate through BEAM
63                // links, so any in-VM activity child still running (e.g.
64                // abandoned by a with_timeout expiry) must be torn down
65                // here — the exit side of the "side effects die with the
66                // run" contract, and what unblocks the child's exit watcher.
67                runtime.kill_in_vm_children(pid);
68                // D5: completions delivered after the workflow stopped
69                // awaiting them (race losers, post-exit deliveries) are
70                // never taken; drop them with the process.
71                runtime.drain_activity_completions(pid);
72                callback(outcome);
73            })
74            .map_err(|error| EngineError::Runtime {
75                reason: format!("failed to spawn workflow monitor for process {pid}: {error}"),
76            })?;
77        Ok(ProcessMonitorHandle::installed())
78    }
79
80    /// Test-only monitor installation status probe.
81    ///
82    /// # Errors
83    ///
84    /// Returns [`EngineError`] if the runtime rejects the monitor installation.
85    #[cfg(test)]
86    pub fn monitor_process_for_test<F>(
87        self: &Arc<Self>,
88        pid: Pid,
89        callback: F,
90    ) -> Result<ProcessMonitorHandle, EngineError>
91    where
92        F: FnOnce(Result<WorkflowProcessOutcome, EngineError>) + Send + 'static,
93    {
94        self.monitor_process(pid, callback)
95    }
96}
97
98#[cfg(test)]
99mod tests {
100    use std::sync::Arc;
101    use std::sync::mpsc;
102    use std::time::Duration;
103
104    use crate::runtime::{RuntimeConfig, RuntimeHandle};
105
106    type TestResult = Result<(), Box<dyn std::error::Error>>;
107
108    #[test]
109    fn monitor_installs_for_process_that_already_exited() -> TestResult {
110        let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(Some(1)))?);
111        let pid = runtime.spawn_test_process()?;
112        runtime.cancel_pid(pid)?;
113        assert!(
114            !runtime.is_live(pid),
115            "terminated test process should leave the live table"
116        );
117
118        // A workflow can finish on a scheduler thread before its completion
119        // monitor installs; the monitor must still observe the exit outcome
120        // through the scheduler's tombstone instead of rejecting the pid.
121        let (sender, receiver) = mpsc::channel();
122        let handle = runtime.monitor_process_for_test(pid, move |outcome| {
123            let _ = sender.send(outcome.is_ok());
124        })?;
125
126        assert!(handle.is_installed());
127        let callback_fired = receiver.recv_timeout(Duration::from_secs(10))?;
128        // The outcome conversion result is exercised elsewhere; this test
129        // pins the contract that the callback fires for an exited process.
130        let _ = callback_fired;
131        runtime.shutdown()?;
132        Ok(())
133    }
134
135    #[test]
136    fn monitor_rejects_pid_never_spawned_by_this_runtime() -> TestResult {
137        let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(Some(1)))?);
138
139        let error = runtime
140            .monitor_process_for_test(9_999, |_| {})
141            .err()
142            .ok_or("monitor accepted a pid this runtime never spawned")?;
143
144        assert!(error.to_string().contains("never spawned"));
145        runtime.shutdown()?;
146        Ok(())
147    }
148}
aion/runtime/monitor.rs

aion/runtime/
monitor.rs
