diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml index 30a1347..2cacbae 100644 --- a/.github/workflows/build-and-test.yml +++ b/.github/workflows/build-and-test.yml @@ -26,5 +26,9 @@ jobs: run: cargo test -p taurus test_execution_request -- --ignored --nocapture env: NATS_URL: nats://127.0.0.1:4222 + - name: Run NATS remote runtime tests + run: cargo test -p taurus-provider -- --ignored --nocapture + env: + NATS_URL: nats://127.0.0.1:4222 - name: Run tests package flow suite run: cargo run --package taurus-tests diff --git a/Cargo.lock b/Cargo.lock index 0be5b64..03a12ee 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1090,6 +1090,11 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash 0.2.0", +] [[package]] name = "heck" @@ -1580,6 +1585,15 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "lru" +version = "0.18.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d2f2f9b4ba7e6b24d95e7e899329d35be83bcded72c8540cdd5368932d1d90a" +dependencies = [ + "hashbrown 0.17.1", +] + [[package]] name = "lupus" version = "0.0.4" @@ -2872,7 +2886,9 @@ dependencies = [ "futures-lite", "inventory", "log", + "lru", "lupus", + "prost", "rand 0.10.2", "serde", "serde_json", @@ -2942,7 +2958,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", diff --git a/Cargo.toml b/Cargo.toml index e5f689b..552f6d8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,6 +28,7 @@ uuid = { version = "1.23.0", features = ["v4"] } ureq = "3.0.0" chrono = { version = "0.4.42", default-features = false, features = ["std", "clock"] } inventory = "0.3.24" +lru = "0.18.2" syn = { version = "3", features = ["full", "extra-traits"] } quote = "1" proc-macro2 = "1" diff --git a/crates/taurus-bench/benches/engine_execution.rs b/crates/taurus-bench/benches/engine_execution.rs index 9ef1a5c..d3d0713 100644 --- a/crates/taurus-bench/benches/engine_execution.rs +++ b/crates/taurus-bench/benches/engine_execution.rs @@ -217,10 +217,39 @@ criterion_group!( bench_chain, bench_array_map, bench_value_store_get, - bench_compile_vs_encode + bench_compile_vs_encode, + bench_compiled_flow_cache ); criterion_main!(benches); +/// Verifies the compiled-flow cache's real-world win: repeatedly executing +/// the *same* flow (as a long-running `taurus` worker does across many NATS +/// messages) with the cache enabled (default capacity) vs. disabled +/// (capacity 0, i.e. today's pre-cache behavior). This is the number that +/// should improve if the cache is doing its job -- `compile_vs_encode` +/// above only checks that hashing is cheap relative to compiling, not that +/// the cache actually pays off end to end. +fn bench_compiled_flow_cache(c: &mut Criterion) { + let (start, nodes) = build_chain_flow(200); + let mut group = c.benchmark_group("compiled_flow_cache"); + + group.bench_function("cache_disabled/200", |b| { + let engine = ExecutionEngine::with_compiled_flow_cache_capacity(0); + b.iter(|| engine.execute_graph("bench", start, nodes.clone(), None, None, false)); + }); + + group.bench_function("cache_enabled/200", |b| { + let engine = ExecutionEngine::with_compiled_flow_cache_capacity( + taurus_core::runtime::engine::DEFAULT_COMPILED_FLOW_CACHE_CAPACITY, + ); + // Warm the cache before measuring steady-state cache-hit cost. + let _ = engine.execute_graph("warmup", start, nodes.clone(), None, None, false); + b.iter(|| engine.execute_graph("bench", start, nodes.clone(), None, None, false)); + }); + + group.finish(); +} + /// Gate-check for compiled-flow caching: is a correctness-safe cache key /// even cheap? `NodeFunction`/`Value` only derive `PartialEq`, not `Hash`, /// and `ExecutionFlow` has no version field, so the only correctness-safe diff --git a/crates/taurus-core/Cargo.toml b/crates/taurus-core/Cargo.toml index a96ada4..86934e2 100644 --- a/crates/taurus-core/Cargo.toml +++ b/crates/taurus-core/Cargo.toml @@ -20,3 +20,5 @@ tokio = { workspace = true } chrono = { workspace = true } inventory = { workspace = true } taurus-macros = { workspace = true } +lru = { workspace = true } +prost = { workspace = true } diff --git a/crates/taurus-core/src/runtime/engine.rs b/crates/taurus-core/src/runtime/engine.rs index 6fa3fb3..40cc947 100644 --- a/crates/taurus-core/src/runtime/engine.rs +++ b/crates/taurus-core/src/runtime/engine.rs @@ -8,9 +8,14 @@ mod executor; pub(crate) mod model; mod sub_flow_registry; -use std::sync::Arc; +use std::collections::hash_map::DefaultHasher; +use std::hash::{Hash, Hasher}; +use std::num::NonZeroUsize; +use std::sync::{Arc, Mutex}; use futures_lite::future::block_on; +use lru::LruCache; +use prost::Message as _; use tucana::shared::value::Kind; use tucana::shared::{ExecutionFlow, NodeExecutionResult, NodeFunction, Value}; @@ -21,11 +26,129 @@ use crate::runtime::remote::RemoteRuntime; use crate::types::exit_reason::ExitReason; use crate::types::signal::Signal; use compiler::compile_flow; +use model::CompiledFlow; use sub_flow_registry::SubFlowRegistry; /// Unique identifier for one top-level flow execution. pub type ExecutionId = uuid::Uuid; +/// Number of distinct compiled flows kept warm by default; see +/// [`ExecutionEngine::with_compiled_flow_cache_limits`]. +pub const DEFAULT_COMPILED_FLOW_CACHE_CAPACITY: usize = 512; + +/// Default total memory budget for the compiled-flow cache. A flow's raw +/// protobuf size is a lower bound on its compiled footprint, not an +/// estimate of it -- see `COMPILED_SIZE_WEIGHT_MULTIPLIER` -- so this caps +/// *estimated* compiled bytes, not wire bytes. +pub const DEFAULT_COMPILED_FLOW_CACHE_MAX_BYTES: usize = 256 * 1024 * 1024; + +/// `CompiledFlow` re-derives a graph of owned `String`s (handler ids, +/// parameter ids, template signatures, remote service names), a `Vec` per +/// node, a secondary `HashMap` index alongside the node list, +/// and boxed nested args for templates -- all heap allocations the raw +/// encoded protobuf bytes don't pay for. There's no exact measurement of +/// this in the codebase; 3x is a middle-of-the-road estimate for "many +/// small strings and nested collections" workloads (plausible range 2-5x) +/// used only to size the cache's byte budget conservatively. +const COMPILED_SIZE_WEIGHT_MULTIPLIER: usize = 3; + +/// `(project_id, start_node_id, content_hash_of_node_functions)`. +/// +/// `compile_flow`'s output depends only on these three inputs, so hashing +/// the encoded `NodeFunction` bytes (rather than requiring `Hash`/`Eq` on +/// the protobuf types, which they don't derive) gives a correctness-safe +/// key: any edit to a node, its parameters, or the graph shape changes the +/// encoded bytes and therefore the key, forcing a recompile. +type CompiledFlowCacheKey = (i64, i64, u64); + +/// Cache key plus an estimated in-memory weight (see +/// `COMPILED_SIZE_WEIGHT_MULTIPLIER`), both derived from a single pass over +/// the encoded node bytes. +fn compiled_flow_cache_key_and_weight( + project_id: i64, + start_node_id: i64, + nodes: &[NodeFunction], +) -> (CompiledFlowCacheKey, usize) { + let mut buf = Vec::new(); + for node in nodes { + node.encode(&mut buf) + .expect("Vec buffer writes are infallible"); + } + let weight_bytes = buf.len().saturating_mul(COMPILED_SIZE_WEIGHT_MULTIPLIER); + let mut hasher = DefaultHasher::new(); + buf.hash(&mut hasher); + ((project_id, start_node_id, hasher.finish()), weight_bytes) +} + +#[cfg(test)] +fn compiled_flow_cache_key( + project_id: i64, + start_node_id: i64, + nodes: &[NodeFunction], +) -> CompiledFlowCacheKey { + compiled_flow_cache_key_and_weight(project_id, start_node_id, nodes).0 +} + +struct CachedCompiledFlow { + plan: Arc, + weight_bytes: usize, +} + +/// LRU cache of compiled flows, bounded by both entry count and an +/// estimated total byte weight -- whichever limit is hit first evicts the +/// least-recently-used entry. In-process only, not shared across replicas +/// or persisted across restarts: a network hop would cost more than the +/// compile it's avoiding (see `ExecutionEngine::with_compiled_flow_cache_limits`). +struct CompiledFlowCache { + entries: LruCache, + total_bytes: usize, + max_bytes: usize, +} + +impl CompiledFlowCache { + fn new(capacity: NonZeroUsize, max_bytes: usize) -> Self { + Self { + entries: LruCache::new(capacity), + total_bytes: 0, + max_bytes, + } + } + + fn get(&mut self, key: &CompiledFlowCacheKey) -> Option> { + self.entries.get(key).map(|entry| Arc::clone(&entry.plan)) + } + + /// No-op if `weight_bytes` alone exceeds the whole budget -- caching a + /// single flow that big would just immediately evict everything else + /// (including itself, next insert), so it's simplest to skip caching it + /// and let it recompile every time instead. + fn insert(&mut self, key: CompiledFlowCacheKey, plan: Arc, weight_bytes: usize) { + if weight_bytes > self.max_bytes { + return; + } + if let Some((_, evicted)) = self.entries.push(key, CachedCompiledFlow { plan, weight_bytes }) { + self.total_bytes -= evicted.weight_bytes; + } + self.total_bytes += weight_bytes; + while self.total_bytes > self.max_bytes { + match self.entries.pop_lru() { + Some((_, evicted)) => self.total_bytes -= evicted.weight_bytes, + None => break, + } + } + } + + #[cfg(test)] + fn peek(&self, key: &CompiledFlowCacheKey) -> Option> { + self.entries.peek(key).map(|entry| Arc::clone(&entry.plan)) + } + + #[cfg(test)] + fn len(&self) -> usize { + self.entries.len() + } +} + /// Runtime engine entrypoint used by runtime binaries and CLI tools. pub struct ExecutionEngine { handlers: FunctionStore, @@ -34,6 +157,9 @@ pub struct ExecutionEngine { /// 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, + /// `None` means the cache is disabled (zero entry capacity or zero byte + /// budget) -- every execution recompiles, matching pre-cache behavior. + compiled_flow_cache: Mutex>, } /// Full result of one engine execution, including per-node results for reporting. @@ -51,11 +177,40 @@ impl Default for ExecutionEngine { } impl ExecutionEngine { - /// Build a new execution engine with default handler registry. + /// Build a new execution engine with default handler registry and a + /// compiled-flow cache bounded by both + /// [`DEFAULT_COMPILED_FLOW_CACHE_CAPACITY`] entries and + /// [`DEFAULT_COMPILED_FLOW_CACHE_MAX_BYTES`] estimated bytes. pub fn new() -> Self { + Self::with_compiled_flow_cache_limits( + DEFAULT_COMPILED_FLOW_CACHE_CAPACITY, + DEFAULT_COMPILED_FLOW_CACHE_MAX_BYTES, + ) + } + + /// Build a new execution engine with a compiled-flow cache bounded to + /// `cache_capacity` entries, using the default byte budget + /// ([`DEFAULT_COMPILED_FLOW_CACHE_MAX_BYTES`]). Pass `0` to disable the + /// cache and recompile every execution, as before. + pub fn with_compiled_flow_cache_capacity(cache_capacity: usize) -> Self { + Self::with_compiled_flow_cache_limits(cache_capacity, DEFAULT_COMPILED_FLOW_CACHE_MAX_BYTES) + } + + /// Build a new execution engine with a compiled-flow cache bounded by + /// *both* `cache_capacity` entries and `cache_max_bytes` estimated + /// total bytes (see `COMPILED_SIZE_WEIGHT_MULTIPLIER`) -- whichever + /// limit is hit first evicts the least-recently-used entry. Pass `0` + /// for either to disable the cache and recompile every execution. + pub fn with_compiled_flow_cache_limits(cache_capacity: usize, cache_max_bytes: usize) -> Self { + let cache = if cache_max_bytes == 0 { + None + } else { + NonZeroUsize::new(cache_capacity).map(|cap| CompiledFlowCache::new(cap, cache_max_bytes)) + }; Self { handlers: FunctionStore::default(), sub_flow_registry: SubFlowRegistry::new(), + compiled_flow_cache: Mutex::new(cache), } } @@ -152,19 +307,38 @@ 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) => Arc::new(plan), - Err(err) => { - let runtime_error = err.as_runtime_error(); - let signal = Signal::Failure(runtime_error); - return EngineExecutionReport { - signal, - exit_reason: ExitReason::Failure, - node_execution_results: Vec::new(), + let (cache_key, cache_weight) = + compiled_flow_cache_key_and_weight(project_id, start_node_id, &node_functions); + let cached = self + .compiled_flow_cache + .lock() + .unwrap() + .as_mut() + .and_then(|cache| cache.get(&cache_key)); + + // Wrapped in `Arc` here (whether freshly compiled or cloned from + // cache), 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 cached { + Some(plan) => plan, + None => { + let plan = match compile_flow(project_id, start_node_id, node_functions) { + Ok(plan) => Arc::new(plan), + Err(err) => { + let runtime_error = err.as_runtime_error(); + let signal = Signal::Failure(runtime_error); + return EngineExecutionReport { + signal, + exit_reason: ExitReason::Failure, + node_execution_results: Vec::new(), + }; + } }; + if let Some(cache) = self.compiled_flow_cache.lock().unwrap().as_mut() { + cache.insert(cache_key, Arc::clone(&plan), cache_weight); + } + plan } }; let start_idx = compiled.start_idx; @@ -1131,6 +1305,9 @@ mod tests { let engine = ExecutionEngine { handlers, sub_flow_registry: SubFlowRegistry::new(), + compiled_flow_cache: Mutex::new(NonZeroUsize::new(DEFAULT_COMPILED_FLOW_CACHE_CAPACITY).map( + |cap| CompiledFlowCache::new(cap, DEFAULT_COMPILED_FLOW_CACHE_MAX_BYTES), + )), }; let add_node = node( @@ -1693,6 +1870,9 @@ mod tests { let engine = ExecutionEngine { handlers, sub_flow_registry: SubFlowRegistry::new(), + compiled_flow_cache: Mutex::new(NonZeroUsize::new(DEFAULT_COMPILED_FLOW_CACHE_CAPACITY).map( + |cap| CompiledFlowCache::new(cap, DEFAULT_COMPILED_FLOW_CACHE_MAX_BYTES), + )), }; let sleep_node = node(1, "test::sleep", vec![], None); @@ -1884,4 +2064,434 @@ mod tests { assert_node_result_id(&report.node_execution_results[3], 2); assert_node_result_id(&report.node_execution_results[4], 1); } + + /// Proves the compiled-flow cache actually short-circuits recompilation + /// (not just "still works") by asserting the second execution's cached + /// `Arc` is the *same allocation* as the first, rather than + /// timing anything -- a wall-clock assertion here would be flaky under + /// CI load. Performance numbers live in `taurus-bench` (criterion), + /// where noise is handled statistically instead of by a hand assertion. + #[test] + fn compiled_flow_cache_reuses_arc_across_executions_of_the_same_flow() { + let engine = ExecutionEngine::new(); + let add_node = node( + 1, + "std::number::add", + vec![ + literal_param(0, "a", int_value(1)), + literal_param(0, "b", int_value(2)), + ], + None, + ); + let nodes = vec![add_node]; + let key = compiled_flow_cache_key(0, 1, &nodes); + + let (signal, reason) = + engine.execute_graph("run-1", 1, nodes.clone(), None, None, false); + assert_eq!(reason, ExitReason::Success); + assert_eq!(expect_success(signal), int_value(3)); + let first = engine + .compiled_flow_cache + .lock() + .unwrap() + .as_ref() + .and_then(|cache| cache.peek(&key)) + .expect("flow should be cached after first execution"); + + let (signal, reason) = engine.execute_graph("run-2", 1, nodes, None, None, false); + assert_eq!(reason, ExitReason::Success); + assert_eq!(expect_success(signal), int_value(3)); + let second = engine + .compiled_flow_cache + .lock() + .unwrap() + .as_ref() + .and_then(|cache| cache.peek(&key)) + .expect("flow should still be cached after second execution"); + + assert!( + Arc::ptr_eq(&first, &second), + "second execution should reuse the cached Arc, not recompile" + ); + } + + /// A structurally different flow (different node id / handler) must not + /// collide with an unrelated cached entry. + #[test] + fn compiled_flow_cache_key_differs_for_different_flows() { + let nodes_a = vec![node( + 1, + "std::number::add", + vec![ + literal_param(0, "a", int_value(1)), + literal_param(0, "b", int_value(2)), + ], + None, + )]; + let nodes_b = vec![node( + 1, + "std::number::add", + vec![ + literal_param(0, "a", int_value(1)), + literal_param(0, "b", int_value(99)), + ], + None, + )]; + + assert_ne!( + compiled_flow_cache_key(0, 1, &nodes_a), + compiled_flow_cache_key(0, 1, &nodes_b) + ); + } + + /// Capacity 0 must disable the cache: nothing is ever stored, so every + /// execution recompiles, matching pre-cache behavior exactly. + #[test] + fn compiled_flow_cache_capacity_zero_disables_caching() { + let engine = ExecutionEngine::with_compiled_flow_cache_capacity(0); + let nodes = vec![node( + 1, + "std::number::add", + vec![ + literal_param(0, "a", int_value(1)), + literal_param(0, "b", int_value(2)), + ], + None, + )]; + + let (signal, reason) = + engine.execute_graph("run-1", 1, nodes.clone(), None, None, false); + assert_eq!(reason, ExitReason::Success); + assert_eq!(expect_success(signal), int_value(3)); + + assert!( + engine.compiled_flow_cache.lock().unwrap().is_none(), + "capacity 0 should leave the cache disabled (None), never populated" + ); + } + + /// A byte budget too small for every flow must evict the + /// least-recently-used entry, not just refuse new inserts -- proves the + /// cache is self-bounding by size, not only by entry count (the entry + /// count alone can't prevent unbounded memory growth if individual + /// flows are multi-MB). + #[test] + fn compiled_flow_cache_evicts_lru_entry_when_byte_budget_exceeded() { + fn add_node(b: i64) -> NodeFunction { + node( + 1, + "std::number::add", + vec![ + literal_param(0, "a", int_value(1)), + literal_param(0, "b", int_value(b)), + ], + None, + ) + } + + let flow_a = vec![add_node(2)]; + let flow_b = vec![add_node(3)]; + let flow_c = vec![add_node(4)]; + + let (key_a, weight_a) = compiled_flow_cache_key_and_weight(0, 1, &flow_a); + let (key_b, weight_b) = compiled_flow_cache_key_and_weight(0, 1, &flow_b); + let (key_c, weight_c) = compiled_flow_cache_key_and_weight(0, 1, &flow_c); + assert_eq!( + weight_a, weight_b, + "structurally identical flows should weigh the same" + ); + assert_eq!( + weight_a, weight_c, + "structurally identical flows should weigh the same" + ); + + // Room for exactly two entries; a generous entry-count cap so only + // the byte budget is actually under test here. + let max_bytes = weight_a + weight_b; + let engine = ExecutionEngine::with_compiled_flow_cache_limits(100, max_bytes); + + let _ = engine.execute_graph("a", 1, flow_a, None, None, false); + let _ = engine.execute_graph("b", 1, flow_b, None, None, false); + { + let cache = engine.compiled_flow_cache.lock().unwrap(); + assert_eq!(cache.as_ref().unwrap().len(), 2); + } + + // A third distinct flow pushes total weight past the budget. `a` + // is the least-recently-used entry (never touched since its own + // insert) and should be the one evicted, not `b`. + let _ = engine.execute_graph("c", 1, flow_c, None, None, false); + + let cache = engine.compiled_flow_cache.lock().unwrap(); + let cache = cache.as_ref().unwrap(); + assert!( + cache.peek(&key_a).is_none(), + "least-recently-used entry should have been evicted" + ); + assert!( + cache.peek(&key_b).is_some(), + "more recently used entry should survive" + ); + assert!( + cache.peek(&key_c).is_some(), + "newly inserted entry should be present" + ); + assert_eq!(cache.len(), 2); + } + + /// A single flow bigger than the entire byte budget must not be cached + /// at all -- caching it would just evict everything else (including + /// itself, on the very next insert), so it's simplest to let it always + /// recompile instead of thrashing the cache. + #[test] + fn compiled_flow_cache_skips_a_single_flow_larger_than_the_whole_budget() { + let nodes = vec![node( + 1, + "std::number::add", + vec![ + literal_param(0, "a", int_value(1)), + literal_param(0, "b", int_value(2)), + ], + None, + )]; + let (key, weight) = compiled_flow_cache_key_and_weight(0, 1, &nodes); + + let engine = ExecutionEngine::with_compiled_flow_cache_limits(100, weight - 1); + let (signal, reason) = engine.execute_graph("run", 1, nodes, None, None, false); + assert_eq!(reason, ExitReason::Success); + assert_eq!(expect_success(signal), int_value(3)); + + let cache = engine.compiled_flow_cache.lock().unwrap(); + assert!( + cache.as_ref().unwrap().peek(&key).is_none(), + "a flow bigger than the whole budget should not be cached" + ); + } + + /// `if`'s `runnable` branch re-enters the executor through the + /// synchronous thunk path (`execute_from_index_sync`). A `Remote` + /// node inside that branch used to hard-fail with + /// `RemoteRuntimeRequiresAsyncExecution` -- it now bridges through + /// `block_on`, the same pattern already used for a local + /// function-thunk's remote call. + #[test] + fn if_branch_can_execute_a_remote_node() { + let engine = ExecutionEngine::new(); + let target_services = Arc::new(Mutex::new(Vec::new())); + let remote = StubRemoteRuntime { + result: NodeExecutionResult { + started_at: 1, + finished_at: 2, + parameter_results: Vec::new(), + id: Some(node_execution_result::Id::NodeId(2)), + result: Some(node_execution_result::Result::Success(int_value(42))), + }, + target_services: Some(Arc::clone(&target_services)), + project_ids: None, + requests: None, + }; + + let if_node = node( + 1, + "std::control::if", + vec![ + literal_param( + 100, + "condition", + Value { + kind: Some(Kind::BoolValue(true)), + }, + ), + thunk_param(101, "runnable", 2), + ], + None, + ); + let mut remote_branch_node = node( + 2, + "remote::branch_add", + vec![literal_param(200, "payload", int_value(1))], + None, + ); + remote_branch_node.definition_source = Some("action.example".to_string()); + + let report = engine.execute_graph_report( + "test", + 1, + vec![if_node, remote_branch_node], + None, + Some(&remote), + false, + ); + + assert_eq!(report.exit_reason, ExitReason::Success); + assert_eq!(expect_success(report.signal), int_value(42)); + assert_eq!( + *target_services + .lock() + .expect("target service recorder should not be poisoned"), + vec!["example".to_string()] + ); + } + + /// Same as above but for `if_else`'s `else_runnable` branch, to prove + /// the fix isn't `if`-specific (both share the same sync thunk path). + #[test] + fn if_else_branch_can_execute_a_remote_node() { + let engine = ExecutionEngine::new(); + let remote = StubRemoteRuntime { + result: NodeExecutionResult { + started_at: 1, + finished_at: 2, + parameter_results: Vec::new(), + id: Some(node_execution_result::Id::NodeId(3)), + result: Some(node_execution_result::Result::Success(int_value(7))), + }, + target_services: None, + project_ids: None, + requests: None, + }; + + let if_else_node = node( + 1, + "std::control::if_else", + vec![ + literal_param( + 100, + "condition", + Value { + kind: Some(Kind::BoolValue(false)), + }, + ), + thunk_param(101, "runnable", 2), + thunk_param(102, "else_runnable", 3), + ], + None, + ); + let then_branch_node = node( + 2, + "std::control::value", + vec![literal_param(200, "value", int_value(999))], + None, + ); + let mut else_branch_node = node( + 3, + "remote::branch_add", + vec![literal_param(300, "payload", int_value(1))], + None, + ); + else_branch_node.definition_source = Some("action.example".to_string()); + + let report = engine.execute_graph_report( + "test", + 1, + vec![if_else_node, then_branch_node, else_branch_node], + None, + Some(&remote), + false, + ); + + assert_eq!(report.exit_reason, ExitReason::Success); + assert_eq!(expect_success(report.signal), int_value(7)); + } + + /// `stop` used to vanish entirely from `node_execution_results` + /// (`commit_result`'s `other => other` branch skipped recording any + /// non-Success/Failure signal). It's now recorded as `Success(null)` + /// -- while execution still halts exactly as before, proven here by + /// asserting node 2 never runs. + #[test] + fn stop_node_is_recorded_as_success_and_still_halts_execution() { + let engine = ExecutionEngine::new(); + let stop_node = node(1, "std::control::stop", vec![], Some(2)); + let unreachable_node = node( + 2, + "std::control::value", + vec![literal_param(100, "value", int_value(99))], + None, + ); + + let report = engine.execute_graph_report( + "test", + 1, + vec![stop_node, unreachable_node], + None, + None, + false, + ); + + assert_eq!(report.exit_reason, ExitReason::Stop); + assert_eq!(report.node_execution_results.len(), 1); + assert_eq!( + report.node_execution_results[0].id, + Some(node_execution_result::Id::NodeId(1)) + ); + assert_eq!( + report.node_execution_results[0].result, + Some(node_execution_result::Result::Success(Value { + kind: Some(Kind::NullValue(0)), + })) + ); + } + + /// Same root cause, one level up: `if`'s handler tail-returns whatever + /// its branch returns, so a branch calling `stop` used to make `if`'s + /// *own* node result vanish too (it was committing the same + /// unconverted `Signal::Stop`). Both `if` and `stop` must now show up. + #[test] + fn if_wrapping_stop_records_both_if_and_stop_nodes() { + let engine = ExecutionEngine::new(); + let if_node = node( + 1, + "std::control::if", + vec![ + literal_param( + 100, + "condition", + Value { + kind: Some(Kind::BoolValue(true)), + }, + ), + thunk_param(101, "runnable", 2), + ], + Some(3), + ); + let stop_node = node(2, "std::control::stop", vec![], None); + let unreachable_node = node( + 3, + "std::control::value", + vec![literal_param(300, "value", int_value(99))], + None, + ); + + let report = engine.execute_graph_report( + "test", + 1, + vec![if_node, stop_node, unreachable_node], + None, + None, + false, + ); + + assert_eq!(report.exit_reason, ExitReason::Stop); + let recorded_ids: Vec<_> = report + .node_execution_results + .iter() + .map(|result| result.id.clone()) + .collect(); + assert_eq!( + recorded_ids, + vec![ + Some(node_execution_result::Id::NodeId(2)), + Some(node_execution_result::Id::NodeId(1)), + ] + ); + for result in &report.node_execution_results { + assert_eq!( + result.result, + Some(node_execution_result::Result::Success(Value { + kind: Some(Kind::NullValue(0)), + })) + ); + } + } } diff --git a/crates/taurus-core/src/runtime/engine/executor.rs b/crates/taurus-core/src/runtime/engine/executor.rs index 6466465..1573688 100644 --- a/crates/taurus-core/src/runtime/engine/executor.rs +++ b/crates/taurus-core/src/runtime/engine/executor.rs @@ -454,20 +454,19 @@ impl<'a> EngineExecutor<'a> { ); NodeResult { signal, frame_id } } - NodeExecutionTarget::Remote { .. } => { - let started_at = now_unix_micros(); - let signal = self.commit_result( - node.id, - Signal::Failure(RuntimeError::new( - "T-CORE-000004", - "RemoteRuntimeRequiresAsyncExecution", - "Remote runtime nodes cannot be executed from a synchronous thunk callback", - )), - Vec::new(), - started_at, - now_unix_micros(), - value_store, - ); + NodeExecutionTarget::Remote { service } => { + // Branch bodies (`if`/`if_else`) and other lazy-arg + // callbacks re-enter the executor synchronously (see + // `execute_from_index_sync`), so a `Remote` node reached + // this way has no `.await` point to hang off of. Bridge it + // the same way `execute_remote_function_thunk` already + // bridges a local-function-thunk's remote call: block only + // this flow invocation's thread while the one genuine + // `.await` inside `execute_remote_node` (the actual remote + // request) completes. Safe to nest under the multi-thread + // runtime this service always runs under -- other worker + // threads keep servicing the reactor. + let signal = block_on(self.execute_remote_node(node, service, value_store, frame_id)); NodeResult { signal, frame_id } } }; @@ -1240,7 +1239,33 @@ impl<'a> EngineExecutor<'a> { ); Signal::Failure(err) } - // Control signals are transient and should not be cached as node outputs. + // `Stop` carries no value and the `NodeExecutionResult` schema + // (tucana) has no dedicated variant for it, so it's recorded as + // a `Success(null)` -- the node still gets an entry in the + // report instead of silently vanishing, including every node + // that merely relayed a nested `Stop` upward (e.g. `if`/`if_else` + // wrapping a branch that called `stop`). The *returned* signal + // stays `Signal::Stop`, unconverted -- only the recorded value + // is Success-shaped; execution still halts exactly as before. + Signal::Stop => { + value_store.insert_success_with_timing( + node_id, + Value { + kind: Some(Kind::NullValue(0)), + }, + parameter_results, + started_at, + finished_at, + ); + Signal::Stop + } + // `Return` is left transient/unrecorded for now -- scoped out + // of this fix. It's often converted to `Success` before + // reaching here (e.g. inside an eager-argument thunk, see + // `force_eager_args`), but a top-level node whose own handler + // is `std::control::return` hits this same `other` branch and + // would have the identical missing-entry symptom as `Stop` did. + // Not addressed here since it wasn't part of what was reported. other => other, } } @@ -1275,6 +1300,21 @@ impl<'a> EngineExecutor<'a> { ); Signal::Failure(err) } + // See `commit_result` -- same rationale, recorded as + // `Success(null)` while still returning `Signal::Stop` + // unconverted so control flow halts exactly as before. + Signal::Stop => { + value_store.insert_function_success_with_timing( + function_id.to_string(), + Value { + kind: Some(Kind::NullValue(0)), + }, + parameter_results, + started_at, + finished_at, + ); + Signal::Stop + } other => other, } } diff --git a/crates/taurus-core/src/runtime/functions/http.rs b/crates/taurus-core/src/runtime/functions/http.rs index 54cbd2b..bb1c816 100644 --- a/crates/taurus-core/src/runtime/functions/http.rs +++ b/crates/taurus-core/src/runtime/functions/http.rs @@ -138,14 +138,19 @@ fn fail(category: &str, message: impl Into) -> Signal { /// Runs a blocking call via `block_in_place` when inside a Tokio /// multi-thread runtime, so it doesn't stall a shared async worker thread; -/// calls it directly otherwise (`taurus-tests`/`taurus-manual --offline` -/// run the engine with no Tokio runtime at all, where `block_in_place` -/// would panic). +/// calls it directly otherwise. `block_in_place` panics ("can't +/// block_in_place from a current_thread runtime") on anything other than a +/// multi-thread runtime, so a bare `Handle::try_current().is_ok()` check +/// isn't enough -- it only tells you *a* runtime exists, not which flavor. +/// `taurus-tests`/`taurus-manual --offline` run the engine with no Tokio +/// runtime at all, and a `current_thread` runtime (e.g. default-flavor +/// `#[tokio::test]`) is possible too; both fall back to calling directly. fn run_blocking(f: impl FnOnce() -> R) -> R { - if tokio::runtime::Handle::try_current().is_ok() { - tokio::task::block_in_place(f) - } else { - f() + match tokio::runtime::Handle::try_current() { + Ok(handle) if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread => { + tokio::task::block_in_place(f) + } + _ => f(), } } @@ -1174,4 +1179,31 @@ mod tests { panic!("server thread join failed: {:?}", err); } } + + #[test] + fn run_blocking_calls_directly_when_no_tokio_runtime_is_present() { + // No runtime at all -- `taurus-tests`/`taurus-manual --offline` + // shape. `block_in_place` would panic here; falling through to a + // direct call must not. + assert_eq!(run_blocking(|| 1 + 1), 2); + } + + #[tokio::test(flavor = "current_thread")] + async fn run_blocking_calls_directly_under_a_current_thread_runtime() { + // Before the runtime-flavor check, this used to unconditionally + // take the `block_in_place` branch (since a runtime *is* present) + // and panic with "can't block_in_place from a current_thread + // runtime". It must now fall back to a direct call instead. + assert_eq!(run_blocking(|| 1 + 1), 2); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn run_blocking_uses_block_in_place_under_a_multi_thread_runtime() { + // On a multi-thread runtime, block_in_place is used so the + // (potentially long) blocking call doesn't stall a shared async + // worker thread. block_in_place panics if called on a runtime + // that isn't multi-thread, so this succeeding at all is itself + // proof the multi-thread branch was taken, not the direct-call one. + assert_eq!(run_blocking(|| 1 + 1), 2); + } } 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 3179463..a0f5004 100644 --- a/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs +++ b/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs @@ -75,6 +75,14 @@ impl RemoteRuntime for NATSRemoteRuntime { )); } }; + // No explicit `.flush()` here -- matches `async-nats`'s own default + // `Client::request()`, which enqueues the publish and starts + // waiting without flushing first (see `lib.rs`'s `Command::Request` + // handling: it calls `enqueue_write_op` only, same as an ordinary + // publish). The connection task's normal write cadence sends this + // well within any realistic `execution_result_timeout` (seconds), + // so the flush bought negligible correctness margin while costing + // a real, measured amount of latency on every remote call. if let Err(err) = self .client .publish_with_reply(topic, inbox, payload.into()) @@ -90,31 +98,6 @@ impl RemoteRuntime for NATSRemoteRuntime { "Failed to receive any response messages from a remote runtime.", )); } - match tokio::time::timeout(self.execution_result_timeout, self.client.flush()).await { - Ok(Ok(())) => {} - Ok(Err(err)) => { - log::error!( - "RemoteRuntimeException: failed to flush NATS request: {}", - err - ); - return Err(RuntimeError::new( - "T-PROV-000001", - "RemoteRuntimeException", - "Failed to receive any response messages from a remote runtime.", - )); - } - Err(err) => { - log::error!( - "RemoteRuntimeException: failed to flush NATS request before timeout: {}", - err - ); - return Err(RuntimeError::new( - "T-PROV-000001", - "RemoteRuntimeException", - "Failed to receive any response messages from a remote runtime.", - )); - } - } let message = match wait_for_reply( &mut sub, @@ -442,4 +425,38 @@ mod tests { elapsed ); } + + /// Guards the removal of the explicit `.flush()` that used to sit + /// between `publish_with_reply` and waiting for the response: an + /// ordinary call with no delay and no sub-flow activity must still + /// reliably reach the responder and get its reply back well within + /// the timeout, relying only on the connection task's normal write + /// cadence (same as `async-nats`'s own default `Client::request()`, + /// which never flushes either). + #[tokio::test] + #[ignore = "requires a real NATS server, see module docs"] + async fn ordinary_call_succeeds_without_explicit_flush() { + let client = test_client().await; + let runtime = NATSRemoteRuntime::with_execution_result_timeout( + client.clone(), + Duration::from_secs(5), + ); + let execution_identifier = unique_id(); + spawn_delayed_responder(client, "svc", &execution_identifier, Duration::ZERO).await; + let execution = build_execution("svc", &execution_identifier, None); + + let result = runtime.execute_remote(execution).await; + + match result { + Ok(node_result) => assert_eq!( + node_result.result, + Some(node_execution_result::Result::Success( + tucana::shared::Value { + kind: Some(Kind::BoolValue(true)), + } + )) + ), + Err(err) => panic!("expected a successful round trip, got {:?}", err), + } + } } diff --git a/crates/taurus/src/app/mod.rs b/crates/taurus/src/app/mod.rs index 0798fee..574dcb9 100644 --- a/crates/taurus/src/app/mod.rs +++ b/crates/taurus/src/app/mod.rs @@ -32,7 +32,10 @@ pub async fn run() { let config = Config::new(); let telemetry = init_telemetry(&config); install_panic_logging(); - let engine = ExecutionEngine::new(); + let engine = ExecutionEngine::with_compiled_flow_cache_limits( + config.compiled_flow_cache_capacity, + config.compiled_flow_cache_max_bytes, + ); let client = connect_nats(&config).await; let mut health_task = spawn_health_task(&config); diff --git a/crates/taurus/src/config/mod.rs b/crates/taurus/src/config/mod.rs index 14905e7..a1f0464 100644 --- a/crates/taurus/src/config/mod.rs +++ b/crates/taurus/src/config/mod.rs @@ -53,6 +53,19 @@ pub struct Config { /// as a starting point -- tune via env for your actual workload). pub max_concurrent_executions: usize, + /// Number of distinct compiled flows kept warm in the in-process LRU + /// cache, avoiding a recompile on every execution of the same flow. + /// Set to 0 to disable the cache and recompile every execution. + pub compiled_flow_cache_capacity: usize, + + /// Estimated total bytes the compiled-flow cache may hold before + /// evicting the least-recently-used entry, independent of + /// `compiled_flow_cache_capacity` -- whichever limit is hit first wins. + /// Bounds worst-case memory when individual flows are large (nothing + /// in this service caps flow/message size upstream). Set to 0 to + /// disable the cache and recompile every execution. + pub compiled_flow_cache_max_bytes: usize, + /// OpenTelemetry exporter configuration. pub opentelemetry: OpenTelemetry, } @@ -91,6 +104,14 @@ impl Config { ), remote_runtime_timeout_secs: env_with_default("REMOTE_RUNTIME_TIMEOUT_SECS", 30_u64), max_concurrent_executions, + compiled_flow_cache_capacity: env_with_default( + "COMPILED_FLOW_CACHE_CAPACITY", + taurus_core::runtime::engine::DEFAULT_COMPILED_FLOW_CACHE_CAPACITY, + ), + compiled_flow_cache_max_bytes: env_with_default( + "COMPILED_FLOW_CACHE_MAX_BYTES", + taurus_core::runtime::engine::DEFAULT_COMPILED_FLOW_CACHE_MAX_BYTES, + ), opentelemetry: OpenTelemetry { enabled: env_with_default("OPENTELEMETRY_ENABLED", false), service_name: env_with_default(