From 0addd37db8937a2ac59d67ada4563c0c08aae42d Mon Sep 17 00:00:00 2001 From: Raphael Date: Mon, 10 Aug 2026 17:16:13 +0200 Subject: [PATCH 1/3] feat: implemented sub flow execution --- Cargo.lock | 2 + crates/taurus-core/src/runtime/engine.rs | 394 +++++++++++++- .../src/runtime/engine/executor.rs | 204 ++++++-- .../src/runtime/engine/sub_flow_registry.rs | 195 +++++++ crates/taurus-core/src/runtime/remote/mod.rs | 11 + .../providers/remote/nats_remote_runtime.rs | 303 ++++++++++- crates/taurus/Cargo.toml | 2 + crates/taurus/src/app/worker.rs | 493 +++++++++++++++++- 8 files changed, 1543 insertions(+), 61 deletions(-) create mode 100644 crates/taurus-core/src/runtime/engine/sub_flow_registry.rs diff --git a/Cargo.lock b/Cargo.lock index 9f7e2e6..74c7151 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2834,6 +2834,7 @@ dependencies = [ "async-nats", "base64 0.23.1", "code0-flow", + "env_logger", "futures-lite", "log", "opentelemetry", @@ -2847,6 +2848,7 @@ dependencies = [ "tonic", "tonic-health", "tucana", + "uuid", ] [[package]] diff --git a/crates/taurus-core/src/runtime/engine.rs b/crates/taurus-core/src/runtime/engine.rs index 056ca2c..e575d4d 100644 --- a/crates/taurus-core/src/runtime/engine.rs +++ b/crates/taurus-core/src/runtime/engine.rs @@ -6,16 +6,22 @@ mod compiler; mod executor; pub(crate) mod model; +mod sub_flow_registry; + +use std::sync::Arc; use futures_lite::future::block_on; -use tucana::shared::{ExecutionFlow, NodeExecutionResult, NodeFunction, Value}; +use tucana::shared::value::Kind; +use tucana::shared::{ExecutionFlow, ListValue, NodeExecutionResult, NodeFunction, Value}; use crate::handler::registry::FunctionStore; +use crate::runtime::execution::trace::TraceRun; use crate::runtime::execution::value_store::ValueStore; use crate::runtime::remote::RemoteRuntime; use crate::types::exit_reason::ExitReason; use crate::types::signal::Signal; use compiler::compile_flow; +use sub_flow_registry::SubFlowRegistry; /// Unique identifier for one top-level flow execution. pub type ExecutionId = uuid::Uuid; @@ -23,6 +29,11 @@ pub type ExecutionId = uuid::Uuid; /// Runtime engine entrypoint used by runtime binaries and CLI tools. pub struct ExecutionEngine { handlers: FunctionStore, + /// Registry of sub-flow node ranges minted while a remote node call is + /// outstanding -- shared with every `EngineExecutor` (to mint) and with + /// the `sub_flow_execution.*` NATS subscriber (via `execute_sub_flow`, + /// to look up and run). See `sub_flow_registry` for the full rationale. + sub_flow_registry: SubFlowRegistry, } /// Full result of one engine execution, including per-node results for reporting. @@ -44,6 +55,7 @@ impl ExecutionEngine { pub fn new() -> Self { Self { handlers: FunctionStore::default(), + sub_flow_registry: SubFlowRegistry::new(), } } @@ -140,8 +152,11 @@ impl ExecutionEngine { ) -> EngineExecutionReport { let mut value_store = ValueStore::new(flow_input.unwrap_or_default(), with_trace); + // Wrapped in `Arc` here, at the point the flow is compiled, so that + // minting a sub-flow registry entry is a cheap refcount bump instead + // of a deep clone of the node graph (see `sub_flow_registry`). let compiled = match compile_flow(project_id, start_node_id, node_functions) { - Ok(plan) => plan, + Ok(plan) => Arc::new(plan), Err(err) => { let runtime_error = err.as_runtime_error(); let signal = Signal::Failure(runtime_error); @@ -152,16 +167,100 @@ impl ExecutionEngine { }; } }; + let start_idx = compiled.start_idx; - let (signal, trace_run) = executor::execute_compiled( + let (signal, trace_run) = executor::execute_compiled_from( execution_id, &compiled, + start_idx, + &self.handlers, + &mut value_store, + remote, + with_trace, + self.sub_flow_registry.clone(), + ) + .await; + Self::finish_report(signal, trace_run, &mut value_store, with_trace) + } + + /// Run a previously minted sub-flow node range (see `SubFlowRegistry`). + /// + /// `parameters` are the action-supplied positional values from + /// `ActionSubFlowExecutionRequest.parameters` -- bound the same way a + /// normal top-level flow execution binds `ExecutionFlow.input_value`, + /// wrapped as a single `ListValue` so `Target::FlowInput` references + /// inside the sub-flow's node range resolve positionally against them. + /// + /// Returns `None` if `execution_identifier` doesn't match any pending + /// sub-flow -- already completed (parent call resolved and the entry + /// was removed), never minted, or minted by a process instance that has + /// since restarted (the registry is in-memory only). + pub async fn execute_sub_flow( + &self, + execution_identifier: &str, + parameters: Vec, + remote: Option<&dyn RemoteRuntime>, + with_trace: bool, + ) -> Option { + let pending = self.sub_flow_registry.get(execution_identifier)?; + // Bump the parent call's idle-timeout activity marker: this lookup + // is itself proof the parent call is still being actively driven. + pending.activity.notify_one(); + + let flow_input = Value { + kind: Some(Kind::ListValue(ListValue { values: parameters })), + }; + let mut value_store = ValueStore::new(flow_input, with_trace); + + // Deliberately *not* `pending.parent_execution_id`: if a node inside + // this sub-flow's own node range is itself dispatched remotely, it + // needs a fresh, unique `execution_identifier` for its own + // `ActionExecutionRequest`. Aquila's `PendingReplyStore` + // (`nats_bridge.rs`) is a flat `HashMap` with last-write-wins semantics on collision — and + // the parent's own remote call is *guaranteed* to still be + // outstanding under `parent_execution_id` for as long as this + // sub-flow run can happen at all (that's the entire premise of + // sub-flow execution). Reusing it here would silently clobber the + // parent's pending-reply entry the moment this run makes its own + // remote call, cross-wiring both calls' eventual replies. The + // action doesn't need this id to equal the parent's to correlate + // the sub-flow run back to its session -- it already has that via + // the sub-flow's own minted id (`execution_identifier` above), + // which travelled to it in `ActionSubFlowExecutionRequest`. + let run_execution_id = uuid::Uuid::new_v4().to_string(); + log::debug!( + "Running sub flow execution_identifier={} for parent_execution_id={} as run_execution_id={}", + execution_identifier, + pending.parent_execution_id, + run_execution_id + ); + + let (signal, trace_run) = executor::execute_compiled_from( + &run_execution_id, + &pending.flow, + pending.start_idx, &self.handlers, &mut value_store, remote, with_trace, + self.sub_flow_registry.clone(), ) .await; + Some(Self::finish_report( + signal, + trace_run, + &mut value_store, + with_trace, + )) + } + + fn finish_report( + signal: Signal, + trace_run: Option, + value_store: &mut ValueStore, + with_trace: bool, + ) -> EngineExecutionReport { if with_trace && let Some(trace_run) = trace_run { println!( "{}", @@ -188,7 +287,7 @@ mod tests { use async_trait::async_trait; use std::sync::{Arc, Mutex}; use std::time::Duration; - use tucana::aquila::{ActionExecutionRequest, action_node_value}; + use tucana::aquila::{ActionExecutionRequest, ActionNodeSubFlowValue, action_node_value}; use tucana::shared::{ InputType, ListValue, NodeExecutionResult, NodeParameter, NodeValue, ReferenceValue, Struct, SubFlow, SubFlowFunction, SubFlowSetting, Value, node_execution_result, @@ -1000,7 +1099,10 @@ mod tests { echo_first_arg_handler, 1, )]); - let engine = ExecutionEngine { handlers }; + let engine = ExecutionEngine { + handlers, + sub_flow_registry: SubFlowRegistry::new(), + }; let add_node = node( 1, @@ -1239,6 +1341,283 @@ mod tests { ); } + /// Records the outgoing request and, while the (mocked) remote call is + /// still "in flight", probes the sub-flow registry directly through a + /// handle cloned from the engine before the run started -- proving the + /// entry exists *during* the call, not just inferring it from the + /// request shape. + struct SubFlowMintProbeRuntime { + registry: SubFlowRegistry, + result: NodeExecutionResult, + requests: Arc>>, + found_pending_during_call: Arc>>, + } + + #[async_trait] + impl RemoteRuntime for SubFlowMintProbeRuntime { + async fn execute_remote( + &self, + execution: RemoteExecution, + ) -> Result + { + if let Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier, + })) = &execution.request.parameters[0].value + { + *self + .found_pending_during_call + .lock() + .expect("probe recorder should not be poisoned") = + Some(self.registry.get(execution_identifier).is_some()); + } + self.requests + .lock() + .expect("request recorder should not be poisoned") + .push(execution.request.clone()); + Ok(self.result.clone()) + } + } + + #[test] + fn remote_node_with_sub_flow_parameter_mints_uuid_instead_of_executing_eagerly() { + let engine = ExecutionEngine::new(); + let requests = Arc::new(Mutex::new(Vec::new())); + let found_pending_during_call = Arc::new(Mutex::new(None)); + let remote = SubFlowMintProbeRuntime { + registry: engine.sub_flow_registry.clone(), + result: NodeExecutionResult { + started_at: 1, + finished_at: 2, + parameter_results: Vec::new(), + id: Some(node_execution_result::Id::NodeId(1)), + result: Some(node_execution_result::Result::Success(int_value(1))), + }, + requests: Arc::clone(&requests), + found_pending_during_call: Arc::clone(&found_pending_during_call), + }; + + // Node 1 is dispatched remotely and takes node 2's range as a + // sub-flow-valued parameter instead of a literal. + let mut remote_node = node( + 1, + "remote::open_stream", + vec![thunk_param(100, "on_message", 2)], + None, + ); + remote_node.definition_source = Some("action.example".to_string()); + + // Node 2 is never actually run by this test -- only the compile + + // mint + request-shape behavior is under test here. + let sub_flow_target = node( + 2, + "std::control::value", + vec![literal_param(200, "value", int_value(9))], + None, + ); + + let flow = ExecutionFlow { + flow_id: 10, + project_id: 42, + starting_node_id: 1, + node_functions: vec![remote_node, sub_flow_target], + input_value: None, + }; + + let report = engine.execute_flow_report("test", flow, Some(&remote), false); + assert_eq!(report.exit_reason, ExitReason::Success); + + assert_eq!( + *found_pending_during_call + .lock() + .expect("probe recorder should not be poisoned"), + Some(true), + "registry should hold a matching entry while the remote call is outstanding" + ); + + let requests = requests + .lock() + .expect("request recorder should not be poisoned"); + assert_eq!(requests.len(), 1); + let parameters = &requests[0].parameters; + assert_eq!(parameters.len(), 1); + + let execution_identifier = match ¶meters[0].value { + Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier, + })) => { + assert!( + uuid::Uuid::parse_str(execution_identifier).is_ok(), + "expected a minted UUID, got {:?}", + execution_identifier + ); + execution_identifier.clone() + } + other => panic!( + "expected a minted sub_flow parameter, got {:?} -- the parameter must not be \ + eagerly resolved to a literal for a remote node", + other + ), + }; + + // The parent node's own remote call has resolved (successfully), so + // the registry entry it minted must already have been cleaned up. + assert!( + engine.sub_flow_registry.get(&execution_identifier).is_none(), + "registry entry should be removed once the parent call resolves" + ); + } + + /// Records every outgoing `execution_identifier`. Blocks on `release` + /// only for the parent's own call (identified by its `SubFlow`-valued + /// parameter) so the test can drive `execute_sub_flow` while that call + /// is still outstanding -- exactly the condition under which a nested + /// remote call inside the sub-flow's own node range would collide with + /// the parent's still-registered entry in aquila's `PendingReplyStore` + /// if it reused the parent's `execution_identifier`. + struct NestedRemoteCapturingRuntime { + result: NodeExecutionResult, + minted_sub_flow_id: Arc>>, + release: Arc, + execution_ids: Arc>>, + } + + #[async_trait] + impl RemoteRuntime for NestedRemoteCapturingRuntime { + async fn execute_remote( + &self, + execution: RemoteExecution, + ) -> Result + { + self.execution_ids + .lock() + .expect("execution id recorder should not be poisoned") + .push(execution.request.execution_identifier.clone()); + + if let Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier, + })) = execution + .request + .parameters + .first() + .and_then(|param| param.value.as_ref()) + { + *self + .minted_sub_flow_id + .lock() + .expect("mint recorder should not be poisoned") = + Some(execution_identifier.clone()); + // Stay outstanding, mirroring the parent's remote call + // staying open while sub-flow traffic happens. + self.release.notified().await; + } + + Ok(self.result.clone()) + } + } + + #[test] + fn sub_flow_execution_uses_a_fresh_execution_identifier_for_its_own_remote_calls() { + let engine = ExecutionEngine::new(); + let minted_sub_flow_id = Arc::new(Mutex::new(None)); + let release = Arc::new(tokio::sync::Notify::new()); + let execution_ids = Arc::new(Mutex::new(Vec::new())); + let remote = NestedRemoteCapturingRuntime { + result: NodeExecutionResult { + started_at: 1, + finished_at: 2, + parameter_results: Vec::new(), + id: Some(node_execution_result::Id::NodeId(1)), + result: Some(node_execution_result::Result::Success(int_value(1))), + }, + minted_sub_flow_id: Arc::clone(&minted_sub_flow_id), + release: Arc::clone(&release), + execution_ids: Arc::clone(&execution_ids), + }; + + // Node 1: dispatched remotely (the "parent" call, analogous to + // `open_stream`), takes node 2's range as a sub-flow parameter. + let mut remote_node = node( + 1, + "remote::open_stream", + vec![thunk_param(100, "on_message", 2)], + None, + ); + remote_node.definition_source = Some("action.svc".to_string()); + + // Node 2: the sub-flow's own body -- itself dispatched remotely, + // exactly the scenario in question: a remote call made *from + // within* a sub-flow's node range while the parent call is open. + let mut sub_flow_target = node( + 2, + "remote::callback", + vec![literal_param(200, "value", int_value(5))], + None, + ); + sub_flow_target.definition_source = Some("action.svc".to_string()); + + let flow = ExecutionFlow { + flow_id: 1, + project_id: 1, + starting_node_id: 1, + node_functions: vec![remote_node, sub_flow_target], + input_value: None, + }; + + std::thread::scope(|scope| { + scope.spawn(|| { + let report = engine.execute_flow_report("parent-id", flow, Some(&remote), false); + assert_eq!(report.exit_reason, ExitReason::Success); + }); + + // Wait for the parent's remote call to mint and capture the + // sub-flow id, proving the parent call is genuinely still + // outstanding at this point. + let sub_flow_execution_id = loop { + if let Some(id) = minted_sub_flow_id + .lock() + .expect("mint recorder should not be poisoned") + .clone() + { + break id; + } + std::thread::sleep(Duration::from_millis(1)); + }; + + // Drive the sub-flow's node range while node 1's own call + // (execution_identifier = "parent-id") is still blocked on + // `release` -- node 2 being remote means this issues a second, + // concurrently-outstanding `ActionExecutionRequest`. + let sub_report = futures_lite::future::block_on(engine.execute_sub_flow( + &sub_flow_execution_id, + vec![int_value(7)], + Some(&remote), + false, + )) + .expect("registry should still have the entry while the parent call is outstanding"); + assert_eq!(sub_report.exit_reason, ExitReason::Success); + + release.notify_one(); + }); + + let ids = execution_ids + .lock() + .expect("execution id recorder should not be poisoned") + .clone(); + assert_eq!(ids.len(), 2, "expected exactly one call per remote node"); + assert_eq!(ids[0], "parent-id"); + assert_ne!( + ids[1], "parent-id", + "the sub-flow's own remote call must not reuse the parent's execution_identifier -- \ + doing so would collide with the parent's still-outstanding entry in aquila's \ + PendingReplyStore" + ); + assert!( + uuid::Uuid::parse_str(&ids[1]).is_ok(), + "expected a freshly minted run id, got {:?}", + ids[1] + ); + } + #[test] fn remote_execution_rejects_empty_action_definition_source() { let engine = ExecutionEngine::new(); @@ -1270,7 +1649,10 @@ mod tests { fn node_execution_result_tracks_actual_node_duration() { let mut handlers = FunctionStore::new(); handlers.populate(&[FunctionRegistration::eager("test::sleep", sleep_handler, 0)]); - let engine = ExecutionEngine { handlers }; + let engine = ExecutionEngine { + handlers, + sub_flow_registry: SubFlowRegistry::new(), + }; let sleep_node = node(1, "test::sleep", vec![], None); let report = engine.execute_graph_report("test", 1, vec![sleep_node], None, None, false); diff --git a/crates/taurus-core/src/runtime/engine/executor.rs b/crates/taurus-core/src/runtime/engine/executor.rs index f4e16d6..7c152cf 100644 --- a/crates/taurus-core/src/runtime/engine/executor.rs +++ b/crates/taurus-core/src/runtime/engine/executor.rs @@ -1,9 +1,12 @@ //! Runtime engine execution loop for compiled flow plans. -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use futures_lite::future::block_on; -use tucana::aquila::{ActionExecutionRequest, ActionNodeValue, action_node_value}; +use tokio::sync::Notify; +use tucana::aquila::{ + ActionExecutionRequest, ActionNodeSubFlowValue, ActionNodeValue, action_node_value, +}; use tucana::shared::node_execution_result::Result as TucanaNodeResult; use tucana::shared::reference_value::Target; use tucana::shared::value::Kind; @@ -16,6 +19,7 @@ use crate::handler::registry::{FunctionStore, HandlerFunctionEntry}; use crate::runtime::engine::model::{ CompiledArg, CompiledFlow, CompiledNode, CompiledThunk, NodeExecutionTarget, }; +use crate::runtime::engine::sub_flow_registry::SubFlowRegistry; use crate::runtime::execution::trace::{ ArgKind, ArgTrace, EdgeKind, Outcome, ReferenceKind, TraceRun, }; @@ -26,27 +30,33 @@ use crate::time::now_unix_micros; use crate::types::errors::runtime_error::RuntimeError; use crate::types::signal::Signal; -pub async fn execute_compiled( +/// Executes a compiled flow plan starting at `start_idx` -- used both by a +/// normal top-level run (`start_idx == flow.start_idx`) and by the +/// `sub_flow_execution.*` subscriber running a previously minted sub-flow +/// node range standalone (`ExecutionEngine::execute_sub_flow`). +#[allow(clippy::too_many_arguments)] +pub async fn execute_compiled_from( execution_id: &str, - flow: &CompiledFlow, + flow: &Arc, + start_idx: usize, handlers: &FunctionStore, value_store: &mut ValueStore, remote: Option<&dyn RemoteRuntime>, with_trace: bool, + sub_flow_registry: SubFlowRegistry, ) -> (Signal, Option) { // Keep trace allocation fully optional so the hot path stays lean when tracing is disabled. let tracer = with_trace.then(Mutex::default); let executor = EngineExecutor { execution_id, - flow, + flow: Arc::clone(flow), handlers, remote, tracer: tracer.as_ref(), + sub_flow_registry, }; - let result = executor - .execute_from_index(flow.start_idx, value_store) - .await; + let result = executor.execute_from_index(start_idx, value_store).await; let trace = tracer.and_then(|collector| collector.into_inner().ok()?.take_run()); (result.signal, trace) } @@ -75,10 +85,14 @@ struct EngineExecutor<'a> { /// action can correlate a callback (e.g. `respond`) against the run /// that triggered it. execution_id: &'a str, - flow: &'a CompiledFlow, + /// `Arc`-wrapped so minting a sub-flow registry entry (which captures + /// the flow standalone -- see `sub_flow_registry`) is a cheap refcount + /// bump rather than a deep clone of the node graph. + flow: Arc, handlers: &'a FunctionStore, remote: Option<&'a dyn RemoteRuntime>, tracer: Option<&'a Mutex>, + sub_flow_registry: SubFlowRegistry, } impl<'a> EngineExecutor<'a> { @@ -345,6 +359,11 @@ impl<'a> EngineExecutor<'a> { Ok(request) => match block_on(remote_runtime.execute_remote(RemoteExecution { target_service: service.to_string(), request, + // Function-thunk settings are always eagerly resolved + // to literals (see `build_function_thunk_args`) -- + // never a `CompiledThunk::Node` reference -- so this + // path never mints a sub-flow UUID. + sub_flow_activity: None, })) { Ok(result) => remote_result_to_signal(result), Err(err) => Signal::Failure(err), @@ -549,24 +568,33 @@ impl<'a> EngineExecutor<'a> { } }; - let values = match self.resolve_remote_args(&mut args, value_store, frame_id) { - Ok(values) => values, - Err(signal) => { - return self.commit_result( - node.id, - signal, - parameter_results_from_args(&args), - started_at, - now_unix_micros(), - value_store, - ); - } - }; - let parameter_results = parameter_results_from_values(&values); + // Shared by every sub-flow UUID minted while resolving this call's + // parameters (if any): the `sub_flow_execution.*` subscriber bumps + // it on every lookup+run, so the idle timeout below only fires on + // genuine inactivity, not because the call is legitimately long-lived. + let activity = Arc::new(Notify::new()); + let (params, minted_ids) = + match self.resolve_remote_args(&mut args, value_store, frame_id, &activity) { + Ok(resolved) => resolved, + Err(signal) => { + return self.commit_result( + node.id, + signal, + parameter_results_from_args(&args), + started_at, + now_unix_micros(), + value_store, + ); + } + }; + let parameter_results = parameter_results_from_remote_params(¶ms); - let request = match self.build_remote_request(node, values) { + let request = match self.build_remote_request(node, params) { Ok(request) => request, Err(err) => { + for id in &minted_ids { + self.sub_flow_registry.remove(id); + } return self.commit_result( node.id, Signal::Failure(err), @@ -578,13 +606,33 @@ impl<'a> EngineExecutor<'a> { } }; - match remote_runtime + // Only calls that actually minted a sub-flow reference get the + // renewable idle timeout; an ordinary call with no sub-flow + // parameters has no activity to track and keeps today's flat + // from-the-start deadline (see `NATSRemoteRuntime::execute_remote`). + let sub_flow_activity = if minted_ids.is_empty() { + None + } else { + Some(Arc::clone(&activity)) + }; + + let result = remote_runtime .execute_remote(RemoteExecution { target_service: service.to_string(), request, + sub_flow_activity, }) - .await - { + .await; + + // The parent call is done (success or failure) -- every sub-flow + // UUID minted for it is no longer reachable by the action and can + // be dropped, regardless of how many times (if any) it was actually + // invoked while the call was outstanding. + for id in &minted_ids { + self.sub_flow_registry.remove(id); + } + + match result { Ok(result) => self.commit_remote_result( node.id, result, @@ -818,19 +866,66 @@ impl<'a> EngineExecutor<'a> { None } + /// Resolves one remote node's arguments to either a materialized literal + /// or a minted sub-flow UUID, per positional slot. Returns the resolved + /// slots alongside every id minted along the way, so the caller can + /// remove them all once the remote call this request belongs to + /// resolves (see `execute_remote_node`). fn resolve_remote_args( &self, args: &mut [Argument], value_store: &mut ValueStore, frame_id: Option, - ) -> Result, Signal> { - let mut values = Vec::with_capacity(args.len()); + activity: &Arc, + ) -> Result<(Vec, Vec), Signal> { + let mut params = Vec::with_capacity(args.len()); + let mut minted_ids = Vec::new(); for (index, argument) in args.iter_mut().enumerate() { match argument { - Argument::Eval(value) => values.push(value.clone()), - Argument::Thunk(thunk) => { - // Remote execution always receives materialized values, never thunks. + Argument::Eval(value) => params.push(RemoteParam::Literal(value.clone())), + // A `CompiledThunk::Node` sub-flow reference destined for a + // *remote* node's parameter is not resolved here at all -- + // unlike every other thunk in this engine (including the + // exact same variant reached through the local `build_args` + // path used by `std::control::if`/`if_else`, which stays + // eager and synchronous, see `control.rs`), it may need to + // run zero, one, or many times, driven by the action itself + // over `ActionSubFlowExecutionRequest` while this call is + // outstanding. So instead of executing it we mint a UUID + // and hand the action a `SubFlow` reference it can invoke + // on its own schedule (see `sub_flow_registry`). + // + // `CompiledThunk::Function` (the other `Deferred` variant) + // is unaffected and keeps executing eagerly below, exactly + // as before -- only a bare node reference gets this + // treatment. + Argument::Thunk(Thunk::Node(node_id)) => { + // Not executed, so left exactly as `build_args` already + // recorded it: `eager: false, executed: false`. + match self.sub_flow_registry.mint( + &self.flow, + *node_id, + self.execution_id, + Arc::clone(activity), + ) { + Some(id) => { + minted_ids.push(id.clone()); + params.push(RemoteParam::SubFlow(id)); + } + None => { + return Err(Signal::Failure(RuntimeError::new( + "T-CORE-000001", + "NodeNotFound", + format!("Node {} not found", node_id), + ))); + } + } + } + Argument::Thunk(thunk @ Thunk::Function(_)) => { + // Remote execution always receives materialized values for + // function-thunk args -- this mirrors the pre-existing + // eager-resolution behavior unchanged. self.trace_mark_thunk(frame_id, index, true, true); let child = self.execute_thunk(thunk, value_store); if let (Some(parent), Some(child_root)) = (frame_id, child.root_frame) { @@ -843,7 +938,7 @@ impl<'a> EngineExecutor<'a> { match child.signal { Signal::Success(value) => { *argument = Argument::Eval(value.clone()); - values.push(value); + params.push(RemoteParam::Literal(value)); } // Same unwind rule as local eager params: return exits this call frame only. Signal::Return(value) => return Err(Signal::Success(value)), @@ -853,15 +948,15 @@ impl<'a> EngineExecutor<'a> { } } - Ok(values) + Ok((params, minted_ids)) } fn build_remote_request( &self, node: &CompiledNode, - values: Vec, + params: Vec, ) -> Result { - if node.parameters.len() != values.len() { + if node.parameters.len() != params.len() { return Err(RuntimeError::new( "T-CORE-000005", "RemoteParameterMismatch", @@ -872,10 +967,17 @@ impl<'a> EngineExecutor<'a> { // Parameters are matched positionally on the receiving end, not by // key — `node.parameters` must already be in the function's declared // parameter order. - let parameters = values + let parameters = params .into_iter() - .map(|value| ActionNodeValue { - value: Some(action_node_value::Value::LiteralValue(value)), + .map(|param| ActionNodeValue { + value: Some(match param { + RemoteParam::Literal(value) => action_node_value::Value::LiteralValue(value), + RemoteParam::SubFlow(execution_identifier) => { + action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier, + }) + } + }), }) .collect(); @@ -1131,11 +1233,27 @@ fn parameter_results_from_args(args: &[Argument]) -> Vec Vec { - values +/// One resolved remote-call parameter slot: either a materialized literal +/// value, or a minted sub-flow UUID standing in for a `CompiledThunk::Node` +/// reference the action may invoke later (see `resolve_remote_args`). +enum RemoteParam { + Literal(Value), + SubFlow(String), +} + +fn parameter_results_from_remote_params( + params: &[RemoteParam], +) -> Vec { + params .iter() - .map(|value| NodeParameterNodeExecutionResult { - value: Some(value.clone()), + .map(|param| NodeParameterNodeExecutionResult { + value: match param { + RemoteParam::Literal(value) => Some(value.clone()), + // No literal value was materialized for a minted sub-flow + // reference -- same convention as an unresolved `Argument::Thunk` + // in `parameter_results_from_args`. + RemoteParam::SubFlow(_) => None, + }, }) .collect() } diff --git a/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs b/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs new file mode 100644 index 0000000..804e47d --- /dev/null +++ b/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs @@ -0,0 +1,195 @@ +//! Registry of pending sub-flow node ranges. +//! +//! `execute_compiled` holds `EngineExecutor`/`&mut ValueStore` borrowed for +//! the entire span a remote node call is outstanding (parked on one +//! `.await` in `execute_remote_node`) -- a separate NATS subscriber +//! (`sub_flow_execution.*` in `taurus/src/app/worker.rs`) can't reach back +//! into that borrowed state to run a sub-flow node range on demand. So +//! registry entries are self-contained: everything needed to run the +//! sub-flow's node range standalone, captured at mint time. +//! +//! Entries are looked up (never removed) any number of times while the +//! parent remote call is outstanding -- the action may invoke the same +//! sub-flow UUID repeatedly. Only the parent call's own resolution (success +//! or failure) removes the entries it minted; see `EngineExecutor::execute_remote_node`. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use tokio::sync::Notify; + +use crate::runtime::engine::model::CompiledFlow; + +/// Everything needed to run one sub-flow node range standalone. +#[derive(Clone)] +pub struct PendingSubFlow { + /// The compiled flow the referenced node range lives in. `CompiledFlow` + /// is cheap to clone structurally, but it's wrapped in `Arc` at compile + /// time (see `runtime/engine.rs`) so minting never deep-clones the node + /// graph. + pub flow: Arc, + /// Index of the sub-flow's entry node within `flow.nodes`. + pub start_idx: usize, + /// The *parent* flow execution's id -- reused as the `execution_identifier` + /// for any remote call the sub-flow itself makes, and as the key for the + /// idle-timeout activity marker (see `activity` below). + pub parent_execution_id: String, + /// Shared with the in-flight `NATSRemoteRuntime::execute_remote` call + /// that minted this entry (and any sibling entries minted for the same + /// remote call). Every successful lookup+run bumps this so that call's + /// idle timeout resets instead of firing while the action is still + /// actively driving sub-flow traffic. + pub activity: Arc, +} + +/// Cheaply `Clone`-able handle to the shared pending sub-flow map. Safe to +/// hand a clone to both `EngineExecutor` (to mint) and the `sub_flow_execution.*` +/// NATS subscriber (to look up) since all state lives behind the shared `Arc`. +#[derive(Clone, Default)] +pub struct SubFlowRegistry { + entries: Arc>>, +} + +impl SubFlowRegistry { + pub fn new() -> Self { + Self::default() + } + + /// Mint a fresh execution id for `node_id` inside `flow` and register it. + /// + /// Returns `None` if `node_id` isn't a real node in `flow` -- defensive: + /// the compiler is expected to reject a dangling sub-flow reference + /// before this is ever reached, but minting must not panic on a bad id. + pub fn mint( + &self, + flow: &Arc, + node_id: i64, + parent_execution_id: &str, + activity: Arc, + ) -> Option { + let start_idx = *flow.node_idx_by_id.get(&node_id)?; + let id = uuid::Uuid::new_v4().to_string(); + let pending = PendingSubFlow { + flow: Arc::clone(flow), + start_idx, + parent_execution_id: parent_execution_id.to_string(), + activity, + }; + self.entries + .lock() + .expect("sub flow registry mutex should not be poisoned") + .insert(id.clone(), pending); + Some(id) + } + + /// Look up a pending sub-flow without removing it -- the same id can be + /// looked up many times while the parent call is outstanding. + pub fn get(&self, id: &str) -> Option { + self.entries + .lock() + .expect("sub flow registry mutex should not be poisoned") + .get(id) + .cloned() + } + + /// Remove a single entry. Called once per minted id when the *parent* + /// node's own remote call resolves (success or failure) -- not after + /// each individual sub-flow invocation. + pub fn remove(&self, id: &str) { + self.entries + .lock() + .expect("sub flow registry mutex should not be poisoned") + .remove(id); + } + + #[cfg(test)] + pub fn len(&self) -> usize { + self.entries + .lock() + .expect("sub flow registry mutex should not be poisoned") + .len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::runtime::engine::model::CompiledFlow; + use std::collections::HashMap as StdHashMap; + + fn flow_with_node(node_id: i64) -> Arc { + Arc::new(CompiledFlow { + project_id: 1, + start_idx: 0, + nodes: vec![crate::runtime::engine::model::CompiledNode { + id: node_id, + handler_id: "test::handler".to_string(), + execution_target: crate::runtime::engine::model::NodeExecutionTarget::Local, + next_idx: None, + parameters: Vec::new(), + }], + node_idx_by_id: StdHashMap::from([(node_id, 0usize)]), + }) + } + + #[test] + fn mint_get_remove_round_trip() { + let registry = SubFlowRegistry::new(); + let flow = flow_with_node(7); + let activity = Arc::new(Notify::new()); + + let id = registry + .mint(&flow, 7, "parent-1", activity) + .expect("node 7 exists in flow"); + + let pending = registry.get(&id).expect("entry should exist after mint"); + assert_eq!(pending.start_idx, 0); + assert_eq!(pending.parent_execution_id, "parent-1"); + + // Looking up again does not remove the entry. + assert!(registry.get(&id).is_some()); + + registry.remove(&id); + assert!(registry.get(&id).is_none()); + } + + #[test] + fn mint_returns_none_for_unknown_node_id() { + let registry = SubFlowRegistry::new(); + let flow = flow_with_node(7); + let activity = Arc::new(Notify::new()); + + assert!(registry.mint(&flow, 999, "parent-1", activity).is_none()); + assert_eq!(registry.len(), 0); + } + + #[test] + fn concurrent_mints_from_simulated_parallel_parents_all_survive() { + let registry = SubFlowRegistry::new(); + let flow = flow_with_node(7); + + let handles: Vec<_> = (0..32) + .map(|i| { + let registry = registry.clone(); + let flow = Arc::clone(&flow); + std::thread::spawn(move || { + let activity = Arc::new(Notify::new()); + registry + .mint(&flow, 7, &format!("parent-{i}"), activity) + .expect("node 7 exists in flow") + }) + }) + .collect(); + + let ids: Vec = handles.into_iter().map(|h| h.join().unwrap()).collect(); + + assert_eq!(ids.len(), 32); + let unique: std::collections::HashSet<_> = ids.iter().collect(); + assert_eq!(unique.len(), 32, "every mint should produce a unique id"); + assert_eq!(registry.len(), 32); + + for id in &ids { + assert!(registry.get(id).is_some()); + } + } +} diff --git a/crates/taurus-core/src/runtime/remote/mod.rs b/crates/taurus-core/src/runtime/remote/mod.rs index 1cbd36d..a7eb44f 100644 --- a/crates/taurus-core/src/runtime/remote/mod.rs +++ b/crates/taurus-core/src/runtime/remote/mod.rs @@ -3,7 +3,10 @@ //! Local runtime nodes can delegate execution to remote services through this //! trait without coupling the core engine to a specific transport. +use std::sync::Arc; + use async_trait::async_trait; +use tokio::sync::Notify; use tucana::{aquila::ActionExecutionRequest, shared::NodeExecutionResult}; use crate::types::errors::runtime_error::RuntimeError; @@ -13,6 +16,14 @@ pub struct RemoteExecution { pub target_service: String, /// Execution request payload expected by the remote runtime. pub request: ActionExecutionRequest, + /// Set only when `request` carries at least one minted sub-flow + /// reference. A `RemoteRuntime` implementation that supports the + /// renewable idle-timeout keepalive (see `NATSRemoteRuntime`) should + /// reset its wait deadline every time this is notified instead of + /// enforcing a single flat deadline from call start. `None` means this + /// call has no sub-flow traffic to wait on, so implementations must + /// fall back to their ordinary flat timeout unchanged. + pub sub_flow_activity: Option>, } #[async_trait] diff --git a/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs b/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs index 9d471ab..599f896 100644 --- a/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs +++ b/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs @@ -1,15 +1,28 @@ //! Delegates remote node execution to another service over NATS request/reply: //! publishes an `ActionExecutionRequest` on `action..` -//! with a fresh reply inbox, then waits (bounded by `execution_result_timeout`) -//! for the matching `ActionExecutionResponse`. +//! with a fresh reply inbox, then waits for the matching `ActionExecutionResponse`. +//! +//! For an ordinary call (no sub-flow parameters minted) the wait is bounded +//! by one flat `execution_result_timeout` from the moment the request is +//! sent, same as always. A call that *did* mint at least one sub-flow +//! reference (`RemoteExecution::sub_flow_activity` is `Some`) instead waits +//! on a *renewable* idle timeout: every time the `sub_flow_execution.*` +//! subscriber (`taurus/src/app/worker.rs`) looks up and runs one of this +//! call's minted sub-flows, it notifies the shared `Notify`, which resets +//! the deadline here. The call only fails if the reply never arrives *and* +//! nothing on the sub-flow channel happens for a full `execution_result_timeout` +//! window -- so a call meant to stay open for hours survives as long as the +//! action keeps driving sub-flow traffic (or eventually replies). +use std::sync::Arc; use std::time::Duration; -use async_nats::Client; +use async_nats::{Client, Message, Subscriber}; use futures_lite::StreamExt; -use prost::Message; +use prost::Message as _; use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime}; use taurus_core::types::errors::runtime_error::RuntimeError; +use tokio::sync::Notify; use tonic::async_trait; use tucana::aquila::ActionExecutionResponse; use tucana::shared::NodeExecutionResult; @@ -103,9 +116,15 @@ impl RemoteRuntime for NATSRemoteRuntime { } } - let message = match tokio::time::timeout(self.execution_result_timeout, sub.next()).await { - Ok(Some(message)) => message, - Ok(None) => { + let message = match wait_for_reply( + &mut sub, + self.execution_result_timeout, + execution.sub_flow_activity.as_ref(), + ) + .await + { + ReplyOutcome::Message(message) => message, + ReplyOutcome::Closed => { log::error!("RemoteRuntimeException: NATS reply subscription closed"); return Err(RuntimeError::new( "T-PROV-000001", @@ -113,10 +132,9 @@ impl RemoteRuntime for NATSRemoteRuntime { "Failed to receive any response messages from a remote runtime.", )); } - Err(err) => { + ReplyOutcome::TimedOut => { log::error!( - "RemoteRuntimeException: failed to receive NATS response before timeout: {}", - err + "RemoteRuntimeException: failed to receive NATS response before timeout" ); return Err(RuntimeError::new( "T-PROV-000001", @@ -153,3 +171,268 @@ impl RemoteRuntime for NATSRemoteRuntime { } } } + +enum ReplyOutcome { + Message(Message), + /// The reply subscription ended without ever producing a message. + Closed, + /// No reply arrived (and, for a sub-flow-activity call, no activity + /// either) within the timeout window. + TimedOut, +} + +/// Waits for the next message on `sub`. +/// +/// * `activity` is `None` for an ordinary call: waits with one flat +/// `timeout` from the moment this function is called, exactly like the +/// single `tokio::time::timeout(...)` this replaces. +/// * `activity` is `Some` for a call that minted at least one sub-flow +/// reference: the `timeout` window restarts every time `activity` is +/// notified (bumped by the `sub_flow_execution.*` subscriber on every +/// successful lookup+run for this call's parent id), so genuine sub-flow +/// traffic keeps the wait alive indefinitely -- it only fails on a window +/// with no reply *and* no sub-flow activity at all. +async fn wait_for_reply( + sub: &mut Subscriber, + timeout: Duration, + activity: Option<&Arc>, +) -> ReplyOutcome { + let Some(activity) = activity else { + return match tokio::time::timeout(timeout, sub.next()).await { + Ok(Some(message)) => ReplyOutcome::Message(message), + Ok(None) => ReplyOutcome::Closed, + Err(_) => ReplyOutcome::TimedOut, + }; + }; + + loop { + tokio::select! { + message = sub.next() => { + return match message { + Some(message) => ReplyOutcome::Message(message), + None => ReplyOutcome::Closed, + }; + } + // Activity resets the deadline: looping back around creates a + // fresh `sleep` future below instead of reusing the elapsed one. + _ = activity.notified() => continue, + _ = tokio::time::sleep(timeout) => return ReplyOutcome::TimedOut, + } + } +} + +/// These tests need a real NATS server (CI already runs one as a service on +/// `127.0.0.1:4222` -- see `.github/workflows/build-and-test.yml`); they are +/// `#[ignore]`d so `cargo test` stays hermetic by default. Run locally with +/// a NATS server up (`nats-server` or `docker run -p 4222:4222 nats:2`) via +/// `cargo test -p taurus-provider -- --ignored`, optionally overriding +/// `NATS_URL`. +#[cfg(test)] +mod tests { + use super::*; + use std::time::Instant; + use tucana::aquila::ActionExecutionRequest; + use tucana::shared::{node_execution_result, value::Kind}; + + /// Cheap process-unique id for test topics -- avoids colliding with any + /// other concurrently-running test against the same NATS server, without + /// pulling in a UUID dependency this crate doesn't otherwise need. + fn unique_id() -> String { + use std::sync::atomic::{AtomicU64, Ordering}; + static COUNTER: AtomicU64 = AtomicU64::new(0); + format!( + "test-{}-{}", + std::process::id(), + COUNTER.fetch_add(1, Ordering::Relaxed) + ) + } + + async fn test_client() -> Client { + let url = + std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string()); + async_nats::connect(url) + .await + .expect("connect to local NATS test server") + } + + fn build_execution( + service: &str, + execution_identifier: &str, + sub_flow_activity: Option>, + ) -> RemoteExecution { + RemoteExecution { + target_service: service.to_string(), + request: ActionExecutionRequest { + execution_identifier: execution_identifier.to_string(), + function_identifier: "test::fn".to_string(), + parameters: Vec::new(), + project_id: 1, + }, + sub_flow_activity, + } + } + + /// Subscribes to `action..` and receives + /// the request but never replies -- standing in for an action that's + /// gone silent. Subscribing happens synchronously, *before* returning, + /// so the caller can publish the request immediately afterwards without + /// racing the spawned task for the subscription: a subscriber must + /// already exist when the request is published, otherwise NATS core's + /// "no responders" fast-path immediately synthesizes an empty reply, + /// which would make the call fail instantly instead of actually waiting + /// out the timeout under test. + async fn spawn_silent_responder(client: Client, service: &str, execution_identifier: &str) { + let topic = format!("action.{}.{}", service, execution_identifier); + let mut sub = client + .subscribe(topic) + .await + .expect("subscribe to request topic"); + tokio::spawn(async move { + let _ = sub.next().await; + // Deliberately never replies; keep the subscription alive for + // the rest of the test so no further "no responders" notice + // can be synthesized either. + std::future::pending::<()>().await; + }); + } + + /// Waits for the next request on `action..` + /// and answers it with a canned success response after `delay`, standing + /// in for a slow-but-alive action. See `spawn_silent_responder` for why + /// subscribing happens synchronously before returning. + async fn spawn_delayed_responder( + client: Client, + service: &str, + execution_identifier: &str, + delay: Duration, + ) { + let topic = format!("action.{}.{}", service, execution_identifier); + let mut sub = client + .subscribe(topic) + .await + .expect("subscribe to request topic"); + let execution_identifier = execution_identifier.to_string(); + tokio::spawn(async move { + let message = sub.next().await.expect("request should arrive"); + let reply = message.reply.expect("request should carry a reply subject"); + tokio::time::sleep(delay).await; + let response = ActionExecutionResponse { + execution_identifier, + node_result: Some(NodeExecutionResult { + started_at: 0, + finished_at: 0, + parameter_results: Vec::new(), + id: None, + result: Some(node_execution_result::Result::Success( + tucana::shared::Value { + kind: Some(Kind::BoolValue(true)), + }, + )), + }), + }; + client + .publish(reply, response.encode_to_vec().into()) + .await + .expect("publish reply"); + client.flush().await.expect("flush reply"); + }); + } + + #[tokio::test] + #[ignore = "requires a real NATS server, see module docs"] + async fn ordinary_call_times_out_at_flat_deadline_when_no_activity() { + let client = test_client().await; + let runtime = NATSRemoteRuntime::with_execution_result_timeout( + client.clone(), + Duration::from_millis(200), + ); + let execution_identifier = unique_id(); + spawn_silent_responder(client, "svc", &execution_identifier).await; + let execution = build_execution("svc", &execution_identifier, None); + + let started = Instant::now(); + let result = runtime.execute_remote(execution).await; + let elapsed = started.elapsed(); + + assert!(result.is_err(), "expected a timeout failure, got {:?}", result); + assert!( + elapsed >= Duration::from_millis(200), + "should not fail before the flat deadline, elapsed={:?}", + elapsed + ); + assert!( + elapsed < Duration::from_secs(2), + "should fail at roughly the flat deadline, elapsed={:?}", + elapsed + ); + } + + #[tokio::test] + #[ignore = "requires a real NATS server, see module docs"] + async fn sub_flow_activity_keeps_call_alive_past_the_flat_deadline() { + let client = test_client().await; + let flat_timeout = Duration::from_millis(200); + let runtime = + NATSRemoteRuntime::with_execution_result_timeout(client.clone(), flat_timeout); + let execution_identifier = unique_id(); + let activity = Arc::new(Notify::new()); + + // The reply lands well after the flat deadline would have fired. + let reply_delay = flat_timeout * 3; + spawn_delayed_responder(client, "svc", &execution_identifier, reply_delay).await; + + // Simulate the `sub_flow_execution.*` subscriber bumping activity on + // every sub-flow lookup+run, faster than the flat deadline, for + // longer than the flat deadline alone would tolerate. + let notifier_activity = Arc::clone(&activity); + tokio::spawn(async move { + for _ in 0..8 { + tokio::time::sleep(flat_timeout / 3).await; + notifier_activity.notify_one(); + } + }); + + let execution = build_execution("svc", &execution_identifier, Some(activity)); + let started = Instant::now(); + let result = runtime.execute_remote(execution).await; + let elapsed = started.elapsed(); + + assert!( + result.is_ok(), + "sub-flow activity should have kept the call alive, got {:?}", + result + ); + assert!( + elapsed >= reply_delay, + "should only resolve once the reply actually arrives, elapsed={:?}", + elapsed + ); + } + + #[tokio::test] + #[ignore = "requires a real NATS server, see module docs"] + async fn sub_flow_activity_silence_past_window_still_times_out() { + let client = test_client().await; + let runtime = NATSRemoteRuntime::with_execution_result_timeout( + client.clone(), + Duration::from_millis(200), + ); + let execution_identifier = unique_id(); + spawn_silent_responder(client, "svc", &execution_identifier).await; + // `Some` activity handle, but nothing ever notifies it -- must + // behave the same as an ordinary call with no activity to track. + let activity = Arc::new(Notify::new()); + let execution = build_execution("svc", &execution_identifier, Some(activity)); + + let started = Instant::now(); + let result = runtime.execute_remote(execution).await; + let elapsed = started.elapsed(); + + assert!(result.is_err(), "expected a timeout failure, got {:?}", result); + assert!( + elapsed >= Duration::from_millis(200) && elapsed < Duration::from_secs(2), + "should time out at roughly the idle window, elapsed={:?}", + elapsed + ); + } +} diff --git a/crates/taurus/Cargo.toml b/crates/taurus/Cargo.toml index 61adf3b..422424e 100644 --- a/crates/taurus/Cargo.toml +++ b/crates/taurus/Cargo.toml @@ -18,6 +18,8 @@ tonic-health = { workspace = true } tonic = { workspace = true } taurus-core = { workspace = true } taurus-provider = { workspace = true } +uuid = { workspace = true } +env_logger = { workspace = true } [dev-dependencies] serde = { workspace = true } diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs index 4abe7a1..119baa6 100644 --- a/crates/taurus/src/app/worker.rs +++ b/crates/taurus/src/app/worker.rs @@ -3,6 +3,16 @@ //! `taurus_core::runtime::engine::ExecutionEngine`, and (in dynamic mode) //! reports the result back to Aquila via [`TaurusRuntimeExecutionService`]. //! +//! Also subscribes to `sub_flow_execution.*`: Aquila forwards +//! `ActionSubFlowExecutionRequest`s there (NATS request/reply, not the +//! publish-then-reply-subject shape `execution.*` uses) whenever an action +//! invokes a sub-flow reference minted by a remote node call (see +//! `taurus_core::runtime::engine::sub_flow_registry`). The same +//! `execution_identifier` may be invoked any number of times while the +//! parent remote call is outstanding, so this subscriber never removes a +//! registry entry itself -- only the parent call's own resolution does +//! (`EngineExecutor::execute_remote_node`). +//! //! Flows execute concurrently: [`spawn_worker`]'s loop only decodes and //! dispatches messages, spawning one task per execution rather than //! awaiting each one inline, so a slow flow doesn't stall unrelated ones. @@ -11,6 +21,9 @@ //! Once a message is dequeued from NATS we always run it to completion //! (waiting for a permit if needed) rather than dropping it on shutdown, //! since core NATS has no redelivery for an already-claimed message. +//! `sub_flow_execution.*` messages are expected to be quick (Phase 0's +//! aquila-side dispatch timeout keeps each hop short) so they are spawned +//! without going through the same semaphore as full flow executions. //! //! Callers signal shutdown cooperatively (see //! [`crate::app::wait_for_shutdown`]) so the worker stops accepting new @@ -27,6 +40,7 @@ use taurus_core::types::signal::Signal; use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime; use tokio::sync::{Notify, Semaphore}; use tokio::task::{JoinHandle, JoinSet}; +use tucana::aquila::ActionSubFlowExecutionRequest; use tucana::shared::execution_result; use tucana::shared::{ExecutionFlow, ExecutionResult, NodeExecutionResult, Value}; @@ -67,10 +81,31 @@ pub fn spawn_worker( } }; + let mut sub_flow_execution_subscription = match client + .queue_subscribe(String::from("sub_flow_execution.*"), "taurus".into()) + .await + { + Ok(subscription) => { + log::info!("Subscribed to 'sub_flow_execution.*'"); + subscription + } + Err(err) => { + log::error!("Failed to subscribe to 'sub_flow_execution.*': {:?}", err); + errors::record( + "transport", + "nats.subscribe", + &err, + "subject=sub_flow_execution.* queue=taurus", + ); + return; + } + }; + let mut execution_closed = false; + let mut sub_flow_execution_closed = false; let mut in_flight = JoinSet::new(); - while !execution_closed { + while !execution_closed || !sub_flow_execution_closed { tokio::select! { message = execution_subscription.next(), if !execution_closed => { match message { @@ -101,8 +136,37 @@ pub fn spawn_worker( } } } + message = sub_flow_execution_subscription.next(), if !sub_flow_execution_closed => { + match message { + Some(message) => { + // Deliberately not gated by `semaphore`: sub-flow + // hops are meant to be quick (see module docs), + // and gating them behind the same permit pool as + // full flow executions could deadlock a flow + // that's waiting on its own sub-flow traffic + // while holding a permit for the parent call. + let engine = engine.clone(); + let nats_remote = nats_remote.clone(); + let client = client.clone(); + in_flight.spawn(async move { + process_sub_flow_execution_message( + message, + &engine, + &nats_remote, + &client, + with_trace, + ).await; + }); + } + None => { + sub_flow_execution_closed = true; + log::warn!("Subscription 'sub_flow_execution.*' ended"); + } + } + } _ = shutdown.notified() => { execution_closed = true; + sub_flow_execution_closed = true; log::info!("NATS worker received shutdown signal"); } } @@ -196,6 +260,167 @@ async fn process_execution_message( } } +/// Handles one `sub_flow_execution.*` NATS request/reply message: decodes it +/// into an `ActionSubFlowExecutionRequest`, runs the referenced sub-flow +/// node range (if the registry still has an entry for it), and publishes an +/// `ExecutionResult` back to the request's reply subject -- aquila's +/// `nats_bridge::handle_sub_flow_execution` awaits exactly this shape via +/// `nats_client.request(...)`. +async fn process_sub_flow_execution_message( + message: async_nats::Message, + engine: &ExecutionEngine, + nats_remote: &NATSRemoteRuntime, + client: &async_nats::Client, + with_trace: bool, +) { + let Some(reply) = message.reply.clone() else { + log::error!( + "Received 'sub_flow_execution' message without a reply subject on '{}'; dropping", + message.subject + ); + return; + }; + + let request = match ActionSubFlowExecutionRequest::decode(&*message.payload) { + Ok(request) => request, + Err(err) => { + log::error!( + "Failed to deserialize sub flow execution request: {:?}, payload: {:?}", + err, + &message.payload + ); + errors::record( + "serialization", + "action_sub_flow_execution_request.decode", + &err, + format!( + "subject={} payload_bytes={}", + message.subject, + message.payload.len() + ), + ); + publish_sub_flow_execution_result(client, reply, build_sub_flow_decode_error_result()) + .await; + return; + } + }; + + let result = + build_sub_flow_execution_result(request, engine, Some(nats_remote), with_trace).await; + publish_sub_flow_execution_result(client, reply, result).await; +} + +/// Core logic for one sub-flow execution reply, decoupled from NATS wiring +/// (and, via `remote` being the `RemoteRuntime` trait object rather than the +/// concrete `NATSRemoteRuntime`, from a real NATS connection too) so it's +/// directly unit-testable: looks the id up in the engine's sub-flow +/// registry and runs it if found. A lookup miss (already completed, never +/// minted, or minted by a since-restarted process) is reported as a normal +/// error `ExecutionResult`, not a dropped/timed-out request -- so aquila +/// surfaces a clean failure to the action instead of its own request timing +/// out silently (see `nats_bridge::handle_sub_flow_execution`). +async fn build_sub_flow_execution_result( + request: ActionSubFlowExecutionRequest, + engine: &ExecutionEngine, + remote: Option<&dyn RemoteRuntime>, + with_trace: bool, +) -> ExecutionResult { + let execution_identifier = request.execution_identifier.clone(); + let started_at = now_unix_micros(); + + match engine + .execute_sub_flow(&execution_identifier, request.parameters, remote, with_trace) + .await + { + Some(report) => build_execution_result( + execution_id_or_generate(&execution_identifier), + 0, + started_at, + now_unix_micros(), + None, + report.node_execution_results, + report.signal, + ), + None => build_sub_flow_not_found_result(execution_identifier, started_at), + } +} + +/// The registry mints ids via `uuid::Uuid::new_v4()`, so parsing back should +/// always succeed; falls back to a fresh id only defensively (the parse +/// failure would already be visible in the id echoed back on the result). +fn execution_id_or_generate(execution_identifier: &str) -> ExecutionId { + ExecutionId::parse_str(execution_identifier).unwrap_or_else(|_| ExecutionId::new_v4()) +} + +async fn publish_sub_flow_execution_result( + client: &async_nats::Client, + reply: async_nats::Subject, + result: ExecutionResult, +) { + if let Err(err) = client.publish(reply, result.encode_to_vec().into()).await { + log::error!("Failed to publish sub flow execution result: {:?}", err); + errors::record( + "transport", + "nats.publish", + &err, + "subject=sub_flow_execution.reply", + ); + return; + } + if let Err(err) = client.flush().await { + log::error!("Failed to flush sub flow execution result: {:?}", err); + errors::record( + "transport", + "nats.flush", + &err, + "subject=sub_flow_execution.reply", + ); + } +} + +fn build_sub_flow_not_found_result(execution_identifier: String, started_at: i64) -> ExecutionResult { + let now = now_unix_micros(); + let runtime_error = RuntimeError::new( + "T-TAURUS-000002", + "SubFlowExecutionNotFound", + "No pending sub flow execution found for this execution identifier -- it may have \ + already completed, never existed, or the owning process restarted", + ); + + ExecutionResult { + execution_identifier, + flow_id: 0, + started_at, + finished_at: now, + input: None, + node_execution_results: Vec::new(), + result: Some(execution_result::Result::Error( + runtime_error.as_tucana_error(), + )), + } +} + +fn build_sub_flow_decode_error_result() -> ExecutionResult { + let now = now_unix_micros(); + let runtime_error = RuntimeError::new( + "T-TAURUS-000003", + "SubFlowExecutionRequestDecodeError", + "Failed to decode sub flow execution request payload", + ); + + ExecutionResult { + execution_identifier: String::new(), + flow_id: 0, + started_at: now, + finished_at: now, + input: None, + node_execution_results: Vec::new(), + result: Some(execution_result::Result::Error( + runtime_error.as_tucana_error(), + )), + } +} + #[derive(Clone)] struct FlowRunResult { execution_id: ExecutionId, @@ -378,10 +603,19 @@ fn build_decode_error_result(execution_id: ExecutionId) -> ExecutionResult { mod tests { use super::*; + use tonic::async_trait; use serde::Deserialize; + use std::sync::Mutex as StdMutex; + use taurus_core::runtime::engine::ExecutionEngine; + use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime}; + use taurus_core::types::exit_reason::ExitReason; + use tucana::aquila::{ActionNodeSubFlowValue, action_node_value}; use tucana::shared::{ - ValidationFlow, execution_result, + FlowInput, NodeFunction, NodeParameter, ReferencePath, ValidationFlow, execution_result, helper::value::{from_json_value, to_json_value}, + node_value, reference_value, + sub_flow::ExecutionReference, + value::Kind, }; #[derive(Deserialize)] @@ -538,4 +772,259 @@ mod tests { .and_then(|input| input.input.clone().map(from_json_value)), } } + + // --- sub_flow_execution.* subscriber core logic ----------------------- + + fn int_value(value: i64) -> Value { + Value { + kind: Some(Kind::NumberValue(tucana::shared::NumberValue { + number: Some(tucana::shared::number_value::Number::Integer(value)), + })), + } + } + + fn node( + database_id: i64, + runtime_function_id: &str, + parameters: Vec, + next_node_id: Option, + definition_source: Option<&str>, + ) -> NodeFunction { + NodeFunction { + database_id: Some(database_id), + runtime_function_id: runtime_function_id.to_string(), + parameters, + next_node_id, + definition_source: definition_source.map(str::to_string), + } + } + + /// A `SubFlow{starting_node_id}`-valued parameter, exactly what + /// `resolve_remote_args` mints a UUID for when the owning node is + /// dispatched remotely (see `taurus-core::runtime::engine::executor`). + fn sub_flow_param(database_id: i64, runtime_parameter_id: &str, node_id: i64) -> NodeParameter { + NodeParameter { + database_id, + runtime_parameter_id: runtime_parameter_id.to_string(), + value: Some(tucana::shared::NodeValue { + value: Some(node_value::Value::SubFlow(tucana::shared::SubFlow { + input_schema: None, + output_schema: None, + signature: String::new(), + settings: Vec::new(), + execution_reference: Some(ExecutionReference::StartingNodeId(node_id)), + })), + }), + cast: None, + } + } + + /// Reads the first positional value out of the sub-flow's seeded flow + /// input -- `ExecutionEngine::execute_sub_flow` wraps + /// `ActionSubFlowExecutionRequest.parameters` as a single `ListValue` + /// flow input, so this is how a sub-flow node range would read its + /// first action-supplied argument. + fn first_flow_input_param(database_id: i64, runtime_parameter_id: &str) -> NodeParameter { + NodeParameter { + database_id, + runtime_parameter_id: runtime_parameter_id.to_string(), + value: Some(tucana::shared::NodeValue { + value: Some(node_value::Value::ReferenceValue( + tucana::shared::ReferenceValue { + target: Some(reference_value::Target::FlowInput(FlowInput {})), + paths: vec![ReferencePath { + path: None, + array_index: Some(0), + }], + }, + )), + }), + cast: None, + } + } + + /// Captures the minted sub-flow UUID from the outgoing remote request's + /// `SubFlow` parameter, then blocks until `release` is notified -- + /// standing in for the parent remote call staying outstanding while the + /// action drives `sub_flow_execution.*` traffic against the minted id. + /// The registry only drops an entry once the *parent* call resolves + /// (`EngineExecutor::execute_remote_node`), so without this the entry + /// would already be gone by the time the test tries a second lookup. + struct BlockingMintCapturingRuntime { + result: NodeExecutionResult, + minted_id: Arc>>, + release: Arc, + } + + #[async_trait] + impl RemoteRuntime for BlockingMintCapturingRuntime { + async fn execute_remote( + &self, + execution: RemoteExecution, + ) -> Result { + if let Some(parameter) = execution.request.parameters.first() + && let Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier, + })) = ¶meter.value + { + *self + .minted_id + .lock() + .expect("mint recorder should not be poisoned") = Some(execution_identifier.clone()); + } + self.release.notified().await; + Ok(self.result.clone()) + } + } + + async fn wait_for_minted_id(minted_id: &StdMutex>) -> String { + for _ in 0..1000 { + if let Some(id) = minted_id + .lock() + .expect("mint recorder should not be poisoned") + .clone() + { + return id; + } + tokio::task::yield_now().await; + } + panic!("timed out waiting for a sub flow id to be minted"); + } + + #[tokio::test] + async fn sub_flow_execution_result_runs_pending_entry_and_can_be_invoked_repeatedly() { + let engine = Arc::new(ExecutionEngine::new()); + let minted_id = Arc::new(StdMutex::new(None)); + let release = Arc::new(Notify::new()); + let remote = BlockingMintCapturingRuntime { + result: NodeExecutionResult { + started_at: 1, + finished_at: 2, + parameter_results: Vec::new(), + id: Some(tucana::shared::node_execution_result::Id::NodeId(1)), + result: Some(tucana::shared::node_execution_result::Result::Success( + int_value(1), + )), + }, + minted_id: Arc::clone(&minted_id), + release: Arc::clone(&release), + }; + + let remote_node = node( + 1, + "remote::open_stream", + vec![sub_flow_param(100, "on_message", 2)], + None, + Some("action.svc"), + ); + // Echoes the first action-supplied sub-flow-invocation parameter + // straight back out, so a run's result proves which parameters it + // was actually seeded with. + let sub_flow_target = node( + 2, + "std::control::value", + vec![first_flow_input_param(200, "value")], + None, + None, + ); + + let flow = ExecutionFlow { + flow_id: 1, + project_id: 7, + starting_node_id: 1, + node_functions: vec![remote_node, sub_flow_target], + input_value: None, + }; + + // The "parent" remote call is kept outstanding (parked on + // `release.notified()`) for the rest of this test by running it on + // its own task. + let parent_engine = Arc::clone(&engine); + let parent = tokio::spawn(async move { + parent_engine + .execute_flow_report_async("parent-1", flow, Some(&remote), false) + .await + }); + + let execution_identifier = wait_for_minted_id(&minted_id).await; + + let first_request = ActionSubFlowExecutionRequest { + execution_identifier: execution_identifier.clone(), + parameters: vec![int_value(41)], + }; + let first_result = + build_sub_flow_execution_result(first_request, &engine, None, false).await; + assert_eq!(first_result.execution_identifier, execution_identifier); + match first_result.result { + Some(execution_result::Result::Success(value)) => assert_eq!(value, int_value(41)), + other => panic!("expected success result, got {:?}", other), + } + + // The same id can be invoked again while the parent call is still + // outstanding -- the registry entry is not removed by running it. + let second_request = ActionSubFlowExecutionRequest { + execution_identifier: execution_identifier.clone(), + parameters: vec![int_value(99)], + }; + let second_result = + build_sub_flow_execution_result(second_request, &engine, None, false).await; + match second_result.result { + Some(execution_result::Result::Success(value)) => assert_eq!(value, int_value(99)), + other => panic!("expected success result on repeated invocation, got {:?}", other), + } + + // Letting the parent call resolve removes the entry it minted. + release.notify_one(); + let report = parent.await.expect("parent task should not panic"); + assert_eq!(report.exit_reason, ExitReason::Success); + + let after_parent_completion = ActionSubFlowExecutionRequest { + execution_identifier: execution_identifier.clone(), + parameters: Vec::new(), + }; + let after_result = + build_sub_flow_execution_result(after_parent_completion, &engine, None, false).await; + match after_result.result { + Some(execution_result::Result::Error(err)) => { + assert_eq!(err.code, "T-TAURUS-000002"); + } + other => panic!( + "expected the registry entry to be gone once the parent call resolved, got {:?}", + other + ), + } + } + + #[tokio::test] + async fn sub_flow_execution_result_reports_not_found_for_unknown_id() { + let engine = ExecutionEngine::new(); + let request = ActionSubFlowExecutionRequest { + execution_identifier: "does-not-exist".to_string(), + parameters: Vec::new(), + }; + + let result = build_sub_flow_execution_result(request, &engine, None, false).await; + + assert_eq!(result.execution_identifier, "does-not-exist"); + match result.result { + Some(execution_result::Result::Error(err)) => { + assert_eq!(err.code, "T-TAURUS-000002"); + assert_eq!(err.category, "SubFlowExecutionNotFound"); + } + other => panic!("expected not-found error result, got {:?}", other), + } + } + + #[test] + fn sub_flow_decode_error_result_reports_expected_code() { + let result = build_sub_flow_decode_error_result(); + + match result.result { + Some(execution_result::Result::Error(err)) => { + assert_eq!(err.code, "T-TAURUS-000003"); + assert_eq!(err.category, "SubFlowExecutionRequestDecodeError"); + } + other => panic!("expected decode error result, got {:?}", other), + } + } } From 63fe1365db9b7206f01ccbe88d369ad7b32e3ae9 Mon Sep 17 00:00:00 2001 From: Raphael Date: Wed, 12 Aug 2026 13:41:26 +0200 Subject: [PATCH 2/3] fix: input order for sub flow --- crates/taurus-core/src/runtime/engine.rs | 34 +++++++++++++---- .../src/runtime/engine/executor.rs | 2 + .../src/runtime/engine/sub_flow_registry.rs | 21 +++++++++-- crates/taurus/src/app/worker.rs | 37 ++++++++++++------- 4 files changed, 70 insertions(+), 24 deletions(-) diff --git a/crates/taurus-core/src/runtime/engine.rs b/crates/taurus-core/src/runtime/engine.rs index e575d4d..a290b5c 100644 --- a/crates/taurus-core/src/runtime/engine.rs +++ b/crates/taurus-core/src/runtime/engine.rs @@ -12,7 +12,7 @@ use std::sync::Arc; use futures_lite::future::block_on; use tucana::shared::value::Kind; -use tucana::shared::{ExecutionFlow, ListValue, NodeExecutionResult, NodeFunction, Value}; +use tucana::shared::{ExecutionFlow, NodeExecutionResult, NodeFunction, Value}; use crate::handler::registry::FunctionStore; use crate::runtime::execution::trace::TraceRun; @@ -187,9 +187,15 @@ impl ExecutionEngine { /// /// `parameters` are the action-supplied positional values from /// `ActionSubFlowExecutionRequest.parameters` -- bound the same way a - /// normal top-level flow execution binds `ExecutionFlow.input_value`, - /// wrapped as a single `ListValue` so `Target::FlowInput` references - /// inside the sub-flow's node range resolve positionally against them. + /// local consumer callback binds them for a native `for_each`/`map`/etc. + /// (see `functions/array.rs::run_with_unary_input`): each positional + /// value is seeded as `InputType{node_id: caller_node_id, parameter_index: + /// caller_parameter_index, input_index}`, so `Target::InputType` + /// references inside the sub-flow's node range -- which is exactly what + /// the compiler emits for a value the sub-flow was invoked with -- find + /// them keyed the same way regardless of whether the callback ran + /// in-process or, as here, standalone in response to an + /// `ActionSubFlowExecutionRequest`. /// /// Returns `None` if `execution_identifier` doesn't match any pending /// sub-flow -- already completed (parent call resolved and the entry @@ -207,10 +213,22 @@ impl ExecutionEngine { // is itself proof the parent call is still being actively driven. pending.activity.notify_one(); - let flow_input = Value { - kind: Some(Kind::ListValue(ListValue { values: parameters })), - }; - let mut value_store = ValueStore::new(flow_input, with_trace); + let mut value_store = ValueStore::new( + Value { + kind: Some(Kind::NullValue(0)), + }, + with_trace, + ); + for (input_index, value) in parameters.into_iter().enumerate() { + value_store.insert_input_type( + tucana::shared::InputType { + node_id: pending.caller_node_id, + parameter_index: pending.caller_parameter_index, + input_index: input_index as i64, + }, + value, + ); + } // Deliberately *not* `pending.parent_execution_id`: if a node inside // this sub-flow's own node range is itself dispatched remotely, it diff --git a/crates/taurus-core/src/runtime/engine/executor.rs b/crates/taurus-core/src/runtime/engine/executor.rs index 7c152cf..716e0e1 100644 --- a/crates/taurus-core/src/runtime/engine/executor.rs +++ b/crates/taurus-core/src/runtime/engine/executor.rs @@ -908,6 +908,8 @@ impl<'a> EngineExecutor<'a> { *node_id, self.execution_id, Arc::clone(activity), + value_store.get_current_node_id(), + index as i64, ) { Some(id) => { minted_ids.push(id.clone()); diff --git a/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs b/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs index 804e47d..115adca 100644 --- a/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs +++ b/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs @@ -34,6 +34,17 @@ pub struct PendingSubFlow { /// for any remote call the sub-flow itself makes, and as the key for the /// idle-timeout activity marker (see `activity` below). pub parent_execution_id: String, + /// Id of the node whose parameter this sub-flow was minted from (e.g. + /// the `for_each` node), and that parameter's positional index (e.g. + /// `consumer`). Together with each call's positional slot these form the + /// `InputType{node_id, parameter_index, input_index}` key the sub-flow's + /// node range references for its action-supplied arguments -- the same + /// addressing a local consumer callback gets via + /// `ValueStore::insert_input_type` (see `functions/array.rs::run_with_unary_input`). + /// `execute_sub_flow` seeds the store with these before running the + /// node range standalone. + pub caller_node_id: i64, + pub caller_parameter_index: i64, /// Shared with the in-flight `NATSRemoteRuntime::execute_remote` call /// that minted this entry (and any sibling entries minted for the same /// remote call). Every successful lookup+run bumps this so that call's @@ -66,6 +77,8 @@ impl SubFlowRegistry { node_id: i64, parent_execution_id: &str, activity: Arc, + caller_node_id: i64, + caller_parameter_index: i64, ) -> Option { let start_idx = *flow.node_idx_by_id.get(&node_id)?; let id = uuid::Uuid::new_v4().to_string(); @@ -74,6 +87,8 @@ impl SubFlowRegistry { start_idx, parent_execution_id: parent_execution_id.to_string(), activity, + caller_node_id, + caller_parameter_index, }; self.entries .lock() @@ -139,7 +154,7 @@ mod tests { let activity = Arc::new(Notify::new()); let id = registry - .mint(&flow, 7, "parent-1", activity) + .mint(&flow, 7, "parent-1", activity, 1, 1) .expect("node 7 exists in flow"); let pending = registry.get(&id).expect("entry should exist after mint"); @@ -159,7 +174,7 @@ mod tests { let flow = flow_with_node(7); let activity = Arc::new(Notify::new()); - assert!(registry.mint(&flow, 999, "parent-1", activity).is_none()); + assert!(registry.mint(&flow, 999, "parent-1", activity, 1, 1).is_none()); assert_eq!(registry.len(), 0); } @@ -175,7 +190,7 @@ mod tests { std::thread::spawn(move || { let activity = Arc::new(Notify::new()); registry - .mint(&flow, 7, &format!("parent-{i}"), activity) + .mint(&flow, 7, &format!("parent-{i}"), activity, 1, 1) .expect("node 7 exists in flow") }) }) diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs index 119baa6..1970cf3 100644 --- a/crates/taurus/src/app/worker.rs +++ b/crates/taurus/src/app/worker.rs @@ -611,7 +611,7 @@ mod tests { use taurus_core::types::exit_reason::ExitReason; use tucana::aquila::{ActionNodeSubFlowValue, action_node_value}; use tucana::shared::{ - FlowInput, NodeFunction, NodeParameter, ReferencePath, ValidationFlow, execution_result, + NodeFunction, NodeParameter, ValidationFlow, execution_result, helper::value::{from_json_value, to_json_value}, node_value, reference_value, sub_flow::ExecutionReference, @@ -819,23 +819,34 @@ mod tests { } } - /// Reads the first positional value out of the sub-flow's seeded flow - /// input -- `ExecutionEngine::execute_sub_flow` wraps - /// `ActionSubFlowExecutionRequest.parameters` as a single `ListValue` - /// flow input, so this is how a sub-flow node range would read its - /// first action-supplied argument. - fn first_flow_input_param(database_id: i64, runtime_parameter_id: &str) -> NodeParameter { + /// Reads the first positional value the sub-flow was invoked with -- + /// `ExecutionEngine::execute_sub_flow` seeds those as + /// `InputType{node_id: caller_node_id, parameter_index: + /// caller_parameter_index, input_index}`, the same addressing a local + /// consumer callback gets (see `functions/array.rs::run_with_unary_input`), + /// keyed here against the node/parameter that minted the sub-flow + /// reference in the first place (node `caller_node_id`'s parameter at + /// `caller_parameter_index`). + fn first_input_type_param( + database_id: i64, + runtime_parameter_id: &str, + caller_node_id: i64, + caller_parameter_index: i64, + ) -> NodeParameter { NodeParameter { database_id, runtime_parameter_id: runtime_parameter_id.to_string(), value: Some(tucana::shared::NodeValue { value: Some(node_value::Value::ReferenceValue( tucana::shared::ReferenceValue { - target: Some(reference_value::Target::FlowInput(FlowInput {})), - paths: vec![ReferencePath { - path: None, - array_index: Some(0), - }], + target: Some(reference_value::Target::InputType( + tucana::shared::InputType { + node_id: caller_node_id, + parameter_index: caller_parameter_index, + input_index: 0, + }, + )), + paths: vec![], }, )), }), @@ -923,7 +934,7 @@ mod tests { let sub_flow_target = node( 2, "std::control::value", - vec![first_flow_input_param(200, "value")], + vec![first_input_type_param(200, "value", 1, 0)], None, None, ); From dff684d9bcfb2ca395eae6e332f27e6f200813ab Mon Sep 17 00:00:00 2001 From: Raphael Date: Wed, 12 Aug 2026 13:55:28 +0200 Subject: [PATCH 3/3] feat: added more fixtures --- crates/taurus-core/src/fixtures.rs | 15 ++- .../src/runtime/engine/sub_flow_registry.rs | 7 +- crates/taurus-tests/src/main.rs | 98 +++++++++++++++++-- crates/taurus/src/app/worker.rs | 97 +++++++++++++++++- flows/0013_remote_for_each_subflow.json | 98 +++++++++++++++++++ ...14_remote_for_each_empty_list_subflow.json | 68 +++++++++++++ ...ote_for_each_subflow_error_propagates.json | 83 ++++++++++++++++ 7 files changed, 456 insertions(+), 10 deletions(-) create mode 100644 flows/0013_remote_for_each_subflow.json create mode 100644 flows/0014_remote_for_each_empty_list_subflow.json create mode 100644 flows/0015_remote_for_each_subflow_error_propagates.json diff --git a/crates/taurus-core/src/fixtures.rs b/crates/taurus-core/src/fixtures.rs index 3be0f7e..abe359c 100644 --- a/crates/taurus-core/src/fixtures.rs +++ b/crates/taurus-core/src/fixtures.rs @@ -31,7 +31,20 @@ pub struct Case { pub struct RemoteFixture { pub target_service: String, pub function_identifier: String, - pub result_parameter: String, + /// Only meaningful when `sub_flow_calls` is empty: the literal-valued + /// parameter to echo straight back as this remote call's own result + /// (see `0012_remote_function_subflow.json`). + #[serde(default)] + pub result_parameter: Option, + /// Positional values to drive one `ExecutionEngine::execute_sub_flow` + /// call per entry against this request's `SubFlow`-valued parameter, in + /// order -- simulates an action invoking a minted sub-flow reference + /// the same number of times a real action would (e.g. once per element + /// for a remotely-dispatched `for_each`'s consumer callback). When + /// non-empty, the remote call itself resolves to `null` once every call + /// has run, mirroring a `void`-signature remote function. + #[serde(default)] + pub sub_flow_calls: Vec, } #[derive(Clone, Deserialize)] diff --git a/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs b/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs index 115adca..829e1c6 100644 --- a/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs +++ b/crates/taurus-core/src/runtime/engine/sub_flow_registry.rs @@ -153,13 +153,18 @@ mod tests { let flow = flow_with_node(7); let activity = Arc::new(Notify::new()); + // caller_node_id and caller_parameter_index are deliberately distinct + // (3, 5) so a field mix-up in `mint`/`PendingSubFlow` would be + // caught by the assertions below instead of accidentally matching. let id = registry - .mint(&flow, 7, "parent-1", activity, 1, 1) + .mint(&flow, 7, "parent-1", activity, 3, 5) .expect("node 7 exists in flow"); let pending = registry.get(&id).expect("entry should exist after mint"); assert_eq!(pending.start_idx, 0); assert_eq!(pending.parent_execution_id, "parent-1"); + assert_eq!(pending.caller_node_id, 3); + assert_eq!(pending.caller_parameter_index, 5); // Looking up again does not remove the entry. assert!(registry.get(&id).is_some()); diff --git a/crates/taurus-tests/src/main.rs b/crates/taurus-tests/src/main.rs index f16993a..ff7322b 100644 --- a/crates/taurus-tests/src/main.rs +++ b/crates/taurus-tests/src/main.rs @@ -8,21 +8,29 @@ use taurus_core::fixtures::{Case, Cases, Input, RemoteFixture, print_failure, pr use taurus_core::runtime::engine::ExecutionEngine; use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime}; use taurus_core::types::errors::runtime_error::RuntimeError; -use tucana::aquila::action_node_value; +use taurus_core::types::signal::Signal; +use tucana::aquila::{ActionNodeSubFlowValue, action_node_value}; use tucana::shared::node_execution_result::{ Id as NodeExecutionResultId, Result as NodeExecutionOutcome, }; +use tucana::shared::value::Kind; use tucana::shared::{ - NodeExecutionResult, + NodeExecutionResult, Value, helper::value::{from_json_value, to_json_value}, }; -struct FixtureRemoteRuntime { +struct FixtureRemoteRuntime<'a> { fixture: RemoteFixture, + // Needed to simulate an action invoking a minted `SubFlow` reference + // back into `ExecutionEngine::execute_sub_flow` (see `sub_flow_calls`) + // -- the same engine instance the flow itself is running on, so the + // callback resolves against the registry entry the flow's own remote + // call minted. + engine: &'a ExecutionEngine, } #[async_trait::async_trait] -impl RemoteRuntime for FixtureRemoteRuntime { +impl RemoteRuntime for FixtureRemoteRuntime<'_> { async fn execute_remote( &self, execution: RemoteExecution, @@ -48,6 +56,29 @@ impl RemoteRuntime for FixtureRemoteRuntime { )); } + // A `SubFlow`-valued parameter means this request carries a minted + // callback reference (see `resolve_remote_args`/`SubFlowRegistry`) + // -- drive it via `sub_flow_calls` regardless of how many entries + // that list has (zero is a legitimate, real case: an empty input + // list means the reference is minted but never actually invoked). + // Only fall through to the literal-echo path below when no + // `SubFlow` parameter is present at all. + if let Some(execution_identifier) = execution + .request + .parameters + .iter() + .find_map(|parameter| match parameter.value.as_ref()? { + action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier, + }) => Some(execution_identifier.clone()), + _ => None, + }) + { + return self + .drive_sub_flow_calls(&execution, &execution_identifier) + .await; + } + // Parameters are positional on the wire now (no key), so fixtures // with a `resultParameter` only make sense with a single remote // parameter — take it directly rather than looking it up by name. @@ -65,7 +96,7 @@ impl RemoteRuntime for FixtureRemoteRuntime { "T-TEST-000003", "RemoteParameterMissing", format!( - "Remote parameter {} was not provided", + "Remote parameter {:?} was not provided", self.fixture.result_parameter ), ) @@ -83,6 +114,58 @@ impl RemoteRuntime for FixtureRemoteRuntime { } } +impl FixtureRemoteRuntime<'_> { + /// Simulates an action driving `ActionSubFlowExecutionRequest` traffic + /// against `execution_identifier` (the request's minted `SubFlow` + /// reference): one `execute_sub_flow` call per `sub_flow_calls` entry, + /// in order -- exactly as `hercules`'s `for_each` implementation calls + /// back once per list element via `executeSubFlow`. An empty + /// `sub_flow_calls` is valid and drives zero calls (e.g. an empty input + /// list: the reference is minted but never actually invoked). Fails the + /// whole remote call the moment any individual sub-flow run fails, + /// since a real action would abort the same way. Resolves to `null`, + /// matching a `void`-signature consumer-driving remote function (e.g. + /// `for_each` itself). + async fn drive_sub_flow_calls( + &self, + execution: &RemoteExecution, + execution_identifier: &str, + ) -> Result { + for call in &self.fixture.sub_flow_calls { + let value = from_json_value(call.clone()); + let report = self + .engine + .execute_sub_flow(&execution_identifier, vec![value], Some(self), false) + .await + .ok_or_else(|| { + RuntimeError::new( + "T-TEST-000005", + "SubFlowNotFound", + format!( + "Sub flow {} was not minted (or already resolved)", + execution_identifier + ), + ) + })?; + if let Signal::Failure(err) = report.signal { + return Err(err); + } + } + + Ok(NodeExecutionResult { + started_at: 0, + finished_at: 0, + parameter_results: Vec::new(), + id: Some(NodeExecutionResultId::FunctionIdentifier( + execution.request.function_identifier.clone(), + )), + result: Some(NodeExecutionOutcome::Success(Value { + kind: Some(Kind::NullValue(0)), + })), + }) + } +} + pub enum CaseResult { Success, Failure(Input, serde_json::Value), @@ -107,7 +190,10 @@ impl Testable for Case { let remote = self .remote .clone() - .map(|fixture| FixtureRemoteRuntime { fixture }); + .map(|fixture| FixtureRemoteRuntime { + fixture, + engine: &engine, + }); for input in self.inputs.clone() { let flow_input = input.clone().input.map(from_json_value); diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs index 1970cf3..56af0fb 100644 --- a/crates/taurus/src/app/worker.rs +++ b/crates/taurus/src/app/worker.rs @@ -827,11 +827,12 @@ mod tests { /// keyed here against the node/parameter that minted the sub-flow /// reference in the first place (node `caller_node_id`'s parameter at /// `caller_parameter_index`). - fn first_input_type_param( + fn input_type_param( database_id: i64, runtime_parameter_id: &str, caller_node_id: i64, caller_parameter_index: i64, + input_index: i64, ) -> NodeParameter { NodeParameter { database_id, @@ -843,7 +844,7 @@ mod tests { tucana::shared::InputType { node_id: caller_node_id, parameter_index: caller_parameter_index, - input_index: 0, + input_index, }, )), paths: vec![], @@ -854,6 +855,21 @@ mod tests { } } + fn first_input_type_param( + database_id: i64, + runtime_parameter_id: &str, + caller_node_id: i64, + caller_parameter_index: i64, + ) -> NodeParameter { + input_type_param( + database_id, + runtime_parameter_id, + caller_node_id, + caller_parameter_index, + 0, + ) + } + /// Captures the minted sub-flow UUID from the outgoing remote request's /// `SubFlow` parameter, then blocks until `release` is notified -- /// standing in for the parent remote call staying outstanding while the @@ -1006,6 +1022,83 @@ mod tests { } } + /// A sub-flow's action-supplied parameters are seeded one `InputType` + /// slot per positional value (`input_index` 0, 1, ...), not just the + /// first. Drives a two-argument call and reads both slots back through + /// `std::number::add`, so a regression that only seeds `input_index: 0` + /// (as a naive port of the single-argument `for_each` case might) would + /// fail this by treating the second argument as missing. + #[tokio::test] + async fn sub_flow_execution_seeds_every_positional_input_index_independently() { + let engine = Arc::new(ExecutionEngine::new()); + let minted_id = Arc::new(StdMutex::new(None)); + let release = Arc::new(Notify::new()); + let remote = BlockingMintCapturingRuntime { + result: NodeExecutionResult { + started_at: 1, + finished_at: 2, + parameter_results: Vec::new(), + id: Some(tucana::shared::node_execution_result::Id::NodeId(1)), + result: Some(tucana::shared::node_execution_result::Result::Success( + int_value(1), + )), + }, + minted_id: Arc::clone(&minted_id), + release: Arc::clone(&release), + }; + + let remote_node = node( + 1, + "remote::open_stream", + vec![sub_flow_param(100, "on_message", 2)], + None, + Some("action.svc"), + ); + // Reads both action-supplied arguments back independently via their + // own `input_index`, so the assertion below proves neither slot + // leaked into or overwrote the other. + let sub_flow_target = node( + 2, + "std::number::add", + vec![ + input_type_param(200, "first", 1, 0, 0), + input_type_param(201, "second", 1, 0, 1), + ], + None, + None, + ); + + let flow = ExecutionFlow { + flow_id: 1, + project_id: 7, + starting_node_id: 1, + node_functions: vec![remote_node, sub_flow_target], + input_value: None, + }; + + let parent_engine = Arc::clone(&engine); + let parent = tokio::spawn(async move { + parent_engine + .execute_flow_report_async("parent-1", flow, Some(&remote), false) + .await + }); + + let execution_identifier = wait_for_minted_id(&minted_id).await; + + let request = ActionSubFlowExecutionRequest { + execution_identifier: execution_identifier.clone(), + parameters: vec![int_value(3), int_value(4)], + }; + let result = build_sub_flow_execution_result(request, &engine, None, false).await; + match result.result { + Some(execution_result::Result::Success(value)) => assert_eq!(value, int_value(7)), + other => panic!("expected success result, got {:?}", other), + } + + release.notify_one(); + parent.await.expect("parent task should not panic"); + } + #[tokio::test] async fn sub_flow_execution_result_reports_not_found_for_unknown_id() { let engine = ExecutionEngine::new(); diff --git a/flows/0013_remote_for_each_subflow.json b/flows/0013_remote_for_each_subflow.json new file mode 100644 index 0000000..7f6e0b1 --- /dev/null +++ b/flows/0013_remote_for_each_subflow.json @@ -0,0 +1,98 @@ +{ + "name": "0013_remote_for_each_subflow", + "description": "This flow validates that a remotely-dispatched for_each node's SubFlow{startingNodeId} consumer is minted as a callback reference (not eagerly executed) and that each simulated action-driven execute_sub_flow call is seeded with the correct per-element InputType value, exactly the scenario that regressed with a ReferenceValueNotFound on the sub-flow's InputType lookup", + "inputs": [ + { + "input": null, + "expected_result": null + } + ], + "remote": { + "targetService": "websocket-action", + "functionIdentifier": "for_each", + "subFlowCalls": [1, 2, 3] + }, + "flow": { + "flowId": "1", + "projectId": "1", + "startingNodeId": "1", + "nodeFunctions": [ + { + "databaseId": "1", + "runtimeFunctionId": "for_each", + "parameters": [ + { + "databaseId": "1", + "runtimeParameterId": "list", + "value": { + "literalValue": { + "listValue": { + "values": [ + {"numberValue": {"integer": "1"}}, + {"numberValue": {"integer": "2"}}, + {"numberValue": {"integer": "3"}} + ] + } + } + } + }, + { + "databaseId": "2", + "runtimeParameterId": "consumer", + "value": { + "subFlow": { + "startingNodeId": "3" + } + } + } + ], + "definitionSource": "action.websocket-action" + }, + { + "databaseId": "3", + "runtimeFunctionId": "std::number::add", + "parameters": [ + { + "databaseId": "3", + "runtimeParameterId": "first", + "value": { + "literalValue": { + "numberValue": {"integer": "1"} + } + } + }, + { + "databaseId": "4", + "runtimeParameterId": "second", + "value": { + "referenceValue": { + "inputType": { + "nodeId": "1", + "parameterIndex": "1" + } + } + } + } + ], + "nextNodeId": "2", + "definitionSource": "taurus" + }, + { + "databaseId": "2", + "runtimeFunctionId": "std::number::as_text", + "parameters": [ + { + "databaseId": "5", + "runtimeParameterId": "number", + "value": { + "referenceValue": { + "nodeId": "3" + } + } + } + ], + "definitionSource": "taurus" + } + ] + } +} diff --git a/flows/0014_remote_for_each_empty_list_subflow.json b/flows/0014_remote_for_each_empty_list_subflow.json new file mode 100644 index 0000000..d561cad --- /dev/null +++ b/flows/0014_remote_for_each_empty_list_subflow.json @@ -0,0 +1,68 @@ +{ + "name": "0014_remote_for_each_empty_list_subflow", + "description": "This flow validates that a remotely-dispatched for_each over an empty list still mints and immediately resolves cleanly, with the consumer's SubFlow reference never invoked at all -- zero execute_sub_flow calls should still let the parent remote call resolve successfully", + "inputs": [ + { + "input": null, + "expected_result": null + } + ], + "remote": { + "targetService": "websocket-action", + "functionIdentifier": "for_each", + "subFlowCalls": [] + }, + "flow": { + "flowId": "1", + "projectId": "1", + "startingNodeId": "1", + "nodeFunctions": [ + { + "databaseId": "1", + "runtimeFunctionId": "for_each", + "parameters": [ + { + "databaseId": "1", + "runtimeParameterId": "list", + "value": { + "literalValue": { + "listValue": { + "values": [] + } + } + } + }, + { + "databaseId": "2", + "runtimeParameterId": "consumer", + "value": { + "subFlow": { + "startingNodeId": "2" + } + } + } + ], + "definitionSource": "action.websocket-action" + }, + { + "databaseId": "2", + "runtimeFunctionId": "std::number::as_text", + "parameters": [ + { + "databaseId": "5", + "runtimeParameterId": "number", + "value": { + "referenceValue": { + "inputType": { + "nodeId": "1", + "parameterIndex": "1" + } + } + } + } + ], + "definitionSource": "taurus" + } + ] + } +} diff --git a/flows/0015_remote_for_each_subflow_error_propagates.json b/flows/0015_remote_for_each_subflow_error_propagates.json new file mode 100644 index 0000000..ed899c3 --- /dev/null +++ b/flows/0015_remote_for_each_subflow_error_propagates.json @@ -0,0 +1,83 @@ +{ + "name": "0015_remote_for_each_subflow_error_propagates", + "description": "This flow validates that when one of several simulated action-driven execute_sub_flow calls fails, the failure aborts the remaining calls and propagates as the parent remote node's own failure -- exactly how a real action failing mid-loop would surface", + "inputs": [ + { + "input": null, + "expected_result": { + "name": "DivisionByZero", + "message": "You cannot divide by zero" + } + } + ], + "remote": { + "targetService": "websocket-action", + "functionIdentifier": "for_each", + "subFlowCalls": [10, 0] + }, + "flow": { + "flowId": "1", + "projectId": "1", + "startingNodeId": "1", + "nodeFunctions": [ + { + "databaseId": "1", + "runtimeFunctionId": "for_each", + "parameters": [ + { + "databaseId": "1", + "runtimeParameterId": "list", + "value": { + "literalValue": { + "listValue": { + "values": [ + {"numberValue": {"integer": "10"}}, + {"numberValue": {"integer": "0"}} + ] + } + } + } + }, + { + "databaseId": "2", + "runtimeParameterId": "consumer", + "value": { + "subFlow": { + "startingNodeId": "2" + } + } + } + ], + "definitionSource": "action.websocket-action" + }, + { + "databaseId": "2", + "runtimeFunctionId": "std::number::divide", + "parameters": [ + { + "databaseId": "3", + "runtimeParameterId": "first", + "value": { + "literalValue": { + "numberValue": {"integer": "10"} + } + } + }, + { + "databaseId": "4", + "runtimeParameterId": "second", + "value": { + "referenceValue": { + "inputType": { + "nodeId": "1", + "parameterIndex": "1" + } + } + } + } + ], + "definitionSource": "taurus" + } + ] + } +}