aion/runtime/
outcome.rs

1//! Workflow process outcome conversion at the beamr boundary.
2
3use aion_core::{Payload, WorkflowError};
4use beamr::atom::{Atom, AtomTable};
5use beamr::process::ExitReason;
6use beamr::scheduler::Scheduler;
7use beamr::term::Term;
8use beamr::term::boxed::Tuple;
9
10use crate::{EngineError, Pid};
11
12use super::payload::term_to_payload;
13
14/// Runtime-converted terminal workflow process outcome.
15pub enum WorkflowProcessOutcome {
16    /// The workflow process returned normally with a payload result.
17    Completed(Payload),
18    /// The workflow process exited abnormally with a workflow error.
19    Failed(WorkflowError),
20}
21
22pub(super) fn workflow_outcome(
23    scheduler: &Scheduler,
24    atoms: &AtomTable,
25    pid: Pid,
26) -> Result<Result<Payload, WorkflowError>, EngineError> {
27    match workflow_process_outcome(scheduler, atoms, pid)? {
28        WorkflowProcessOutcome::Completed(payload) => Ok(Ok(payload)),
29        WorkflowProcessOutcome::Failed(error) => Ok(Err(error)),
30    }
31}
32
33pub(super) fn workflow_process_outcome(
34    scheduler: &Scheduler,
35    atoms: &AtomTable,
36    pid: Pid,
37) -> Result<WorkflowProcessOutcome, EngineError> {
38    let (reason, owned_result) = scheduler.run_until_exit(pid);
39    if reason != ExitReason::Normal {
40        // The VM execution error, when present, is the authoritative exit
41        // cause and must be checked first: beamr leaves `current_exception`
42        // set after a *caught* raise until the next try_end/catch_end, so a
43        // later unrelated VM error exits with that stale exception attached
44        // and reporting the exception first would blame workflow code for an
45        // engine-level failure. The residue is appended as context.
46        let exception = scheduler.take_exit_exception(pid);
47        if let Some(error) = scheduler.take_exit_error(pid) {
48            let formatted = error.format_with_atoms(atoms);
49            let residue = exception.map_or_else(String::new, |exception| {
50                format!(
51                    " (residual exception: {})",
52                    exception.format_with_atoms(atoms)
53                )
54            });
55            return Ok(WorkflowProcessOutcome::Failed(WorkflowError {
56                message: format!(
57                    "workflow process {pid} exited: {reason:?}: VM execution error: {formatted}{residue}"
58                ),
59                details: None,
60            }));
61        }
62        if let Some(exception) = exception {
63            let formatted = exception.format_with_atoms(atoms);
64            let view = exception.view();
65            let details = term_to_payload(view.reason, atoms).ok();
66            return Ok(WorkflowProcessOutcome::Failed(WorkflowError {
67                message: format!("workflow process {pid} exited: {formatted}"),
68                details,
69            }));
70        }
71    }
72    convert_process_outcome(atoms, pid, reason, owned_result.root())
73}
74
75pub(super) fn convert_process_outcome(
76    atoms: &AtomTable,
77    pid: Pid,
78    reason: ExitReason,
79    result: Term,
80) -> Result<WorkflowProcessOutcome, EngineError> {
81    if reason == ExitReason::Normal {
82        unwrap_gleam_result(result, atoms, pid)
83    } else {
84        let formatted = beamr::term::format::format_term(result, atoms);
85        let details = term_to_payload(result, atoms).ok();
86        Ok(WorkflowProcessOutcome::Failed(WorkflowError {
87            message: format!("workflow process {pid} exited: {reason:?}: {formatted}"),
88            details,
89        }))
90    }
91}
92
93fn unwrap_gleam_result(
94    result: Term,
95    atoms: &AtomTable,
96    pid: Pid,
97) -> Result<WorkflowProcessOutcome, EngineError> {
98    if let Some(tuple) = Tuple::new(result) {
99        if tuple.arity() == 2 {
100            if let (Some(tag), Some(value)) = (tuple.get(0), tuple.get(1)) {
101                if let Some(atom) = tag.as_atom() {
102                    if atom == Atom::OK {
103                        return Ok(WorkflowProcessOutcome::Completed(term_to_payload(
104                            value, atoms,
105                        )?));
106                    }
107                    if atom == Atom::ERROR {
108                        let details = term_to_payload(value, atoms).ok();
109                        return Ok(WorkflowProcessOutcome::Failed(WorkflowError {
110                            message: format!("workflow {pid} returned error"),
111                            details,
112                        }));
113                    }
114                }
115            }
116        }
117    }
118    Ok(WorkflowProcessOutcome::Completed(term_to_payload(
119        result, atoms,
120    )?))
121}
aion/runtime/outcome.rs

aion/runtime/
outcome.rs
