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                // D5: completions delivered after the workflow stopped
63                // awaiting them (race losers, post-exit deliveries) are
64                // never taken; drop them with the process.
65                runtime.drain_activity_completions(pid);
66                callback(outcome);
67            })
68            .map_err(|error| EngineError::Runtime {
69                reason: format!("failed to spawn workflow monitor for process {pid}: {error}"),
70            })?;
71        Ok(ProcessMonitorHandle::installed())
72    }
73
74    /// Test-only monitor installation status probe.
75    ///
76    /// # Errors
77    ///
78    /// Returns [`EngineError`] if the runtime rejects the monitor installation.
79    #[cfg(test)]
80    pub fn monitor_process_for_test<F>(
81        self: &Arc<Self>,
82        pid: Pid,
83        callback: F,
84    ) -> Result<ProcessMonitorHandle, EngineError>
85    where
86        F: FnOnce(Result<WorkflowProcessOutcome, EngineError>) + Send + 'static,
87    {
88        self.monitor_process(pid, callback)
89    }
90}
91
92#[cfg(test)]
93mod tests {
94    use std::sync::Arc;
95    use std::sync::mpsc;
96    use std::time::Duration;
97
98    use crate::runtime::{RuntimeConfig, RuntimeHandle};
99
100    type TestResult = Result<(), Box<dyn std::error::Error>>;
101
102    #[test]
103    fn monitor_installs_for_process_that_already_exited() -> TestResult {
104        let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(Some(1)))?);
105        let pid = runtime.spawn_test_process()?;
106        runtime.cancel_pid(pid)?;
107        assert!(
108            !runtime.is_live(pid),
109            "terminated test process should leave the live table"
110        );
111
112        // A workflow can finish on a scheduler thread before its completion
113        // monitor installs; the monitor must still observe the exit outcome
114        // through the scheduler's tombstone instead of rejecting the pid.
115        let (sender, receiver) = mpsc::channel();
116        let handle = runtime.monitor_process_for_test(pid, move |outcome| {
117            let _ = sender.send(outcome.is_ok());
118        })?;
119
120        assert!(handle.is_installed());
121        let callback_fired = receiver.recv_timeout(Duration::from_secs(10))?;
122        // The outcome conversion result is exercised elsewhere; this test
123        // pins the contract that the callback fires for an exited process.
124        let _ = callback_fired;
125        runtime.shutdown()?;
126        Ok(())
127    }
128
129    #[test]
130    fn monitor_rejects_pid_never_spawned_by_this_runtime() -> TestResult {
131        let runtime = Arc::new(RuntimeHandle::new(RuntimeConfig::new(Some(1)))?);
132
133        let error = runtime
134            .monitor_process_for_test(9_999, |_| {})
135            .err()
136            .ok_or("monitor accepted a pid this runtime never spawned")?;
137
138        assert!(error.to_string().contains("never spawned"));
139        runtime.shutdown()?;
140        Ok(())
141    }
142}
aion/runtime/monitor.rs

aion/runtime/
monitor.rs
