From 337370b4cfaf2659027dce9fcf5d2d1b22a905cb Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Wed, 11 Nov 2026 16:59:57 +0530 Subject: [PATCH] fix(engine): report the same run error whatever order parallel failures finish in When several parallel branches fail under `stop` in one superstep, each handler offered its error to the run's terminal-error slot first-writer- wins, so the reported error followed completion order: one graph and input could report different errors under different latencies, and disagree with the runtime, which already names a failed step by the lowest active-set index. The slot now keeps a TerminalError { step, branch, error } and terminal_error::offer keeps the lowest (superstep, active-set index): the earliest superstep wins, then the lowest index, which for a fan-out is the edge declaration order and the order parallel updates are folded in. A single failure is reported unchanged. --- README.md | 5 +- crates/tinyflows/src/engine.rs | 3 + crates/tinyflows/src/engine/build.rs | 2 +- .../tinyflows/src/engine/build/activation.rs | 5 +- crates/tinyflows/src/engine/build/handlers.rs | 2 +- crates/tinyflows/src/engine/build/outcome.rs | 15 ++- .../src/engine/resumable/continuation.rs | 5 +- crates/tinyflows/src/engine/run_state.rs | 5 +- crates/tinyflows/src/engine/terminal_error.rs | 53 ++++++++ .../src/engine/terminal_error_tests.rs | 44 +++++++ .../tests/parallel_failure_order_tests.rs | 120 ++++++++++++++++++ 11 files changed, 244 insertions(+), 15 deletions(-) create mode 100644 crates/tinyflows/src/engine/terminal_error.rs create mode 100644 crates/tinyflows/src/engine/terminal_error_tests.rs create mode 100644 crates/tinyflows/tests/parallel_failure_order_tests.rs diff --git a/README.md b/README.md index eb0d752b..7f863859 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,10 @@ Rust 2024 · MSRV 1.85 · `#![forbid(unsafe_code)]` · GPL-3.0-or-later. - Per-node error handling: `on_error` policy (`stop` / `continue` / `route`), bounded `retry`, and an `error` output port for routing failures to a recovery - sub-graph. + sub-graph. When several parallel branches fail under `stop` in one super-step, + the run reports the error of the branch earliest in that step's active set + (for a fan-out, its edge order), never the first to finish; an earlier + super-step's failure always wins. - Node postconditions: `config.postcondition = { "require": "non_empty" }` (or `field_present` / `non_empty_list` with a dotted `field`) is checked against every emitted item, so an empty or misshapen output is a failed attempt that diff --git a/crates/tinyflows/src/engine.rs b/crates/tinyflows/src/engine.rs index 6719ee17..20d45359 100644 --- a/crates/tinyflows/src/engine.rs +++ b/crates/tinyflows/src/engine.rs @@ -295,6 +295,9 @@ use build::build_graph; mod run_config; use run_config::RunConfig; +mod terminal_error; +use terminal_error::TerminalSlot; + mod run_state; pub use run_state::resume; use run_state::{build_and_run, default_thread_id, merge_approvals}; diff --git a/crates/tinyflows/src/engine/build.rs b/crates/tinyflows/src/engine/build.rs index 383b8f10..364ac7db 100644 --- a/crates/tinyflows/src/engine/build.rs +++ b/crates/tinyflows/src/engine/build.rs @@ -30,7 +30,7 @@ pub(super) fn build_graph( capabilities: &Capabilities, observer: &Arc, steps: &Arc>>, - terminal_error: &Arc>>, + terminal_error: &TerminalSlot, config: &RunConfig, ) -> Result<(CompiledGraph, String)> { let checkpointer = config.checkpointer.clone(); diff --git a/crates/tinyflows/src/engine/build/activation.rs b/crates/tinyflows/src/engine/build/activation.rs index 77c317f9..f49a05e1 100644 --- a/crates/tinyflows/src/engine/build/activation.rs +++ b/crates/tinyflows/src/engine/build/activation.rs @@ -10,7 +10,7 @@ pub(super) struct HandlerData { pub(super) agents: Arc>, pub(super) observer: Arc, pub(super) steps: Arc>>, - pub(super) terminal_error: Arc>>, + pub(super) terminal_error: TerminalSlot, pub(super) token: CancellationToken, pub(super) routing: HandlerRouting, pub(super) plain_targets: Vec, @@ -61,6 +61,8 @@ impl HandlerData { let lane = lane_context(ctx.send_arg.as_ref()); let lane_send_arg = lane.is_some().then(|| ctx.send_arg.clone()).flatten(); let activation_step = ctx.step; + // A sequential step has no fork; it stops at its first failure. + let branch = ctx.fork.as_ref().map_or(0, |fork| fork.branch_index); // A checkpointed resume (see `ResumableRun::resume`) delivers a resume // value to the interrupted node via `NodeContext::resume`. A resume // approves *this* gate only when it is a bare `true` (backward-compat, @@ -444,6 +446,7 @@ impl HandlerData { &lane, &resume_value, activation_step, + branch, &steps, &terminal_error, &plain_targets, diff --git a/crates/tinyflows/src/engine/build/handlers.rs b/crates/tinyflows/src/engine/build/handlers.rs index a499ca96..d2dfd2cb 100644 --- a/crates/tinyflows/src/engine/build/handlers.rs +++ b/crates/tinyflows/src/engine/build/handlers.rs @@ -7,7 +7,7 @@ pub(super) fn register_handlers( capabilities: &Capabilities, observer: &Arc, steps: &Arc>>, - terminal_error: &Arc>>, + terminal_error: &TerminalSlot, token: &CancellationToken, node_timeout: Option, loop_edges: &std::collections::HashSet<(String, String)>, diff --git a/crates/tinyflows/src/engine/build/outcome.rs b/crates/tinyflows/src/engine/build/outcome.rs index 6505a545..d4c8abfd 100644 --- a/crates/tinyflows/src/engine/build/outcome.rs +++ b/crates/tinyflows/src/engine/build/outcome.rs @@ -22,8 +22,11 @@ pub(super) async fn finish_execution( lane: &Option, resume_value: &Option, activation_step: usize, + // This activation's index in its superstep's active set; picks the + // reported error when several branches fail together. + branch: usize, steps: &Arc>>, - terminal_error: &Arc>>, + terminal_error: &TerminalSlot, plain_targets: &[String], emit: F, ) -> crate::graph::Result> @@ -269,9 +272,9 @@ where // into `EngineError::Capability` and a host could not // tell a loop that ran away from an HTTP call that // 500'd. `build_and_run` prefers whatever lands here. - // First writer wins: the failure that ends the run is - // the one worth reporting, and a parallel superstep - // may produce several. + // A parallel superstep may produce several; the rule in + // `terminal_error::offer` keeps the lowest-ordered one, not + // the first to finish. _ => { let message = err.to_string(); // A lane failure belongs to its gather, whose @@ -286,9 +289,7 @@ where let mut slot = terminal_error .lock() .expect("terminal error mutex poisoned"); - if slot.is_none() { - *slot = Some(err); - } + crate::engine::terminal_error::offer(&mut slot, activation_step, branch, err); drop(slot); // The message still travels the string channel // unchanged, so every path that only sees the diff --git a/crates/tinyflows/src/engine/resumable/continuation.rs b/crates/tinyflows/src/engine/resumable/continuation.rs index 5ed96efd..f686f3ae 100644 --- a/crates/tinyflows/src/engine/resumable/continuation.rs +++ b/crates/tinyflows/src/engine/resumable/continuation.rs @@ -55,7 +55,7 @@ async fn resume_with_checkpointer_inner( // `resume` loads the state persisted under `thread_id`. Node handlers fire // `observer.on_step_finish` for every node that runs after the interrupt // boundary, so a host observer sees the resumed steps live. - let terminal_error: Arc>> = Arc::new(Mutex::new(None)); + let terminal_error: TerminalSlot = Arc::new(Mutex::new(None)); let mut config = RunConfig::new(workflow)?.with_checkpointer(checkpointer, thread_id); if let Some(token) = &token { config = config.with_token(token.clone()); @@ -79,7 +79,8 @@ async fn resume_with_checkpointer_inner( let structured = terminal_error .lock() .expect("terminal error mutex poisoned") - .take(); + .take() + .map(|held| held.error); return Err(structured.unwrap_or_else(|| EngineError::Capability(error.to_string()))); } }; diff --git a/crates/tinyflows/src/engine/run_state.rs b/crates/tinyflows/src/engine/run_state.rs index 375f52f7..8b21f8e8 100644 --- a/crates/tinyflows/src/engine/run_state.rs +++ b/crates/tinyflows/src/engine/run_state.rs @@ -105,7 +105,7 @@ pub(super) async fn build_and_run( // Where a run-ending node failure leaves its structured error on the way // out (see the `stop`-policy branch in `build_graph`), so the caller gets // the real `EngineError` instead of its `Display` string. - let terminal_error: Arc>> = Arc::new(Mutex::new(None)); + let terminal_error: TerminalSlot = Arc::new(Mutex::new(None)); // Build and compile the graph under the host-supplied checkpointer; // `trigger_id` is the graph's entry node. @@ -205,7 +205,8 @@ pub(super) async fn build_and_run( let structured = terminal_error .lock() .expect("terminal error mutex poisoned") - .take(); + .take() + .map(|held| held.error); return Err(structured.unwrap_or_else(|| EngineError::Capability(e.to_string()))); } }; diff --git a/crates/tinyflows/src/engine/terminal_error.rs b/crates/tinyflows/src/engine/terminal_error.rs new file mode 100644 index 00000000..84b1ec44 --- /dev/null +++ b/crates/tinyflows/src/engine/terminal_error.rs @@ -0,0 +1,53 @@ +//! The structured error a failed run reports, chosen independently of timing. +//! +//! A `stop`-policy node failure ends the run, but the runtime can only carry a +//! string out of a node handler, so the handler also leaves its real +//! [`EngineError`] in a shared slot for the entry point to return. A parallel +//! superstep can fail on several branches at once, and they reach the slot in +//! completion order — so the slot must not simply keep whichever came first. + +use std::sync::{Arc, Mutex}; + +use crate::error::EngineError; + +/// A run-ending node failure, with the position it was raised at. +pub(crate) struct TerminalError { + step: usize, + branch: usize, + pub(crate) error: EngineError, +} + +/// The slot every node handler of one run shares. +pub(crate) type TerminalSlot = Arc>>; + +/// Offers `error`, raised at superstep `step` by the activation at index +/// `branch` of that superstep's active set, as the run's reported error. +/// +/// The rule: the earliest superstep wins, and within one superstep the lowest +/// active-set index wins. That index is the order the runtime folds parallel +/// branches in, and the order it already uses to name a failed step, so the +/// structured error and the runtime's own step error agree. Completion order +/// never matters: an earlier superstep's error is never replaced by a later +/// one, whatever its index. A sequential step has no fork and reports index 0; +/// it stops at its first failure, so it offers at most one error per step. +pub(crate) fn offer( + slot: &mut Option, + step: usize, + branch: usize, + error: EngineError, +) { + if slot + .as_ref() + .is_none_or(|held| (step, branch) < (held.step, held.branch)) + { + *slot = Some(TerminalError { + step, + branch, + error, + }); + } +} + +#[cfg(test)] +#[path = "terminal_error_tests.rs"] +mod tests; diff --git a/crates/tinyflows/src/engine/terminal_error_tests.rs b/crates/tinyflows/src/engine/terminal_error_tests.rs new file mode 100644 index 00000000..8de4d646 --- /dev/null +++ b/crates/tinyflows/src/engine/terminal_error_tests.rs @@ -0,0 +1,44 @@ +//! Unit tests for choosing a run's reported error among several failures. + +use super::*; + +fn failure(name: &str) -> EngineError { + EngineError::Capability(name.to_string()) +} + +fn reported(slot: &Option) -> Option { + slot.as_ref().map(|held| held.error.to_string()) +} + +fn expected(name: &str) -> Option { + Some(failure(name).to_string()) +} + +#[test] +fn the_lowest_branch_wins_whichever_arrives_first() { + let mut lowest_first = None; + offer(&mut lowest_first, 2, 0, failure("branch 0")); + offer(&mut lowest_first, 2, 1, failure("branch 1")); + + let mut lowest_last = None; + offer(&mut lowest_last, 2, 1, failure("branch 1")); + offer(&mut lowest_last, 2, 0, failure("branch 0")); + + assert_eq!(reported(&lowest_first), expected("branch 0")); + assert_eq!(reported(&lowest_last), expected("branch 0")); +} + +#[test] +fn an_earlier_superstep_is_not_replaced_by_a_lower_branch_later() { + let mut slot = None; + offer(&mut slot, 1, 3, failure("step 1")); + offer(&mut slot, 2, 0, failure("step 2")); + assert_eq!(reported(&slot), expected("step 1")); +} + +#[test] +fn a_single_failure_is_reported_as_is() { + let mut slot = None; + offer(&mut slot, 4, 2, failure("only")); + assert_eq!(reported(&slot), expected("only")); +} diff --git a/crates/tinyflows/tests/parallel_failure_order_tests.rs b/crates/tinyflows/tests/parallel_failure_order_tests.rs new file mode 100644 index 00000000..2c34fe89 --- /dev/null +++ b/crates/tinyflows/tests/parallel_failure_order_tests.rs @@ -0,0 +1,120 @@ +#![cfg(feature = "mock")] +//! Which error a run reports when several parallel branches fail together. +//! +//! The rule: within one superstep the failing branch with the lowest +//! active-set index names the run's error, whichever finished first. For a +//! static fan-out that index follows the fan-out node's edge order. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::{Value, json}; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::compiler::compile; +use tinyflows::engine::{CancellationToken, run_intercepted}; +use tinyflows::interception::{StepAction, StepFrame, StepInterceptor, StepPhase}; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; +use tinyflows::observability::{NoopObserver, RunObserver}; + +fn node(id: &str, kind: NodeKind, config: Value) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: vec![], + position: None, + } +} + +fn edge(from: &str, to: &str) -> Edge { + Edge { + from_node: from.to_string(), + from_port: "main".to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build runtime") +} + +/// Fails `n1` and `n2` outright, after a per-node yield count. +struct FailBoth(Vec); + +#[async_trait] +impl StepInterceptor for FailBoth { + async fn intercept(&self, frame: StepFrame<'_>) -> StepAction { + let index = match frame.node.id.as_str() { + "n1" => 0, + "n2" => 1, + _ => return StepAction::Continue { state_patch: None }, + }; + if frame.phase != StepPhase::Before { + return StepAction::Continue { state_patch: None }; + } + for _ in 0..self.0[index] { + tokio::task::yield_now().await; + } + StepAction::Fail { + message: format!("unhandled failure at {}", frame.node.id), + } + } +} + +/// Two parallel nodes fail without a handler, under opposite latencies. `n1` +/// is declared first, so it holds the lower active-set index and its error is +/// the one reported — also when `n2` finishes first. +#[test] +fn parallel_failures_report_the_lowest_ordered_branch_under_any_latency() { + let graph = WorkflowGraph { + name: "parallel_failures".to_string(), + nodes: vec![ + node("trigger", NodeKind::Trigger, Value::Null), + node("fan", NodeKind::OutputParser, Value::Null), + node("n1", NodeKind::OutputParser, Value::Null), + node("n2", NodeKind::OutputParser, Value::Null), + node("join", NodeKind::Merge, Value::Null), + ], + edges: vec![ + edge("trigger", "fan"), + edge("fan", "n1"), + edge("fan", "n2"), + edge("n1", "join"), + edge("n2", "join"), + ], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + + let error_under = |delays: Vec| { + runtime().block_on(async { + let observer = Arc::new(NoopObserver) as Arc; + run_intercepted( + &compiled, + json!({}), + &mock_capabilities(), + &observer, + CancellationToken::new(), + Arc::new(FailBoth(delays)), + ) + .await + .map(|_| ()) + .expect_err("both branches fail") + .to_string() + }) + }; + // `[10, 0]`: `n2` finishes first. `[0, 10]`: `n1` does. + for delays in [vec![10, 0], vec![0, 10]] { + let error = error_under(delays.clone()); + assert!( + error.contains("unhandled failure at n1"), + "delays {delays:?} reported {error:?}" + ); + } +}