diff --git a/.github/workflows/build-and-test.yml b/.github/workflows/build-and-test.yml index 36d3318..30a1347 100644 --- a/.github/workflows/build-and-test.yml +++ b/.github/workflows/build-and-test.yml @@ -21,10 +21,10 @@ jobs: env: RUST_BACKTRACE: 'full' - name: Run workspace tests - run: cargo test --workspace --all-targets + run: cargo test --workspace --all-targets --exclude taurus-bench - name: Run NATS test execution request tests run: cargo test -p taurus test_execution_request -- --ignored --nocapture env: NATS_URL: nats://127.0.0.1:4222 - name: Run tests package flow suite - run: cargo run --package tests + run: cargo run --package taurus-tests diff --git a/Cargo.lock b/Cargo.lock index 6f2a455..f2cf740 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -31,12 +31,27 @@ dependencies = [ "memchr", ] +[[package]] +name = "alloca" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5a7d05ea6aea7e9e64d25b9156ba2fee3fdd659e34e41063cd2fc7cd020d7f4" +dependencies = [ + "cc", +] + [[package]] name = "allocator-api2" version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + [[package]] name = "anstream" version = "1.0.0" @@ -335,6 +350,12 @@ dependencies = [ "serde", ] +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cc" version = "1.3.0" @@ -373,6 +394,33 @@ dependencies = [ "num-traits", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + [[package]] name = "clap" version = "4.6.4" @@ -513,6 +561,72 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "criterion" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "950046b2aa2492f9a536f5f4f9a3de7b9e2476e575e05bd6c333371add4d98f3" +dependencies = [ + "alloca", + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "itertools 0.13.0", + "num-traits", + "oorandom", + "page_size", + "plotters", + "rayon", + "regex", + "serde", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8d80a2f4f5b554395e47b5d8305bc3d27813bacb73493eb1001e8f76dae29ea" +dependencies = [ + "cast", + "itertools 0.13.0", +] + +[[package]] +name = "crossbeam-deque" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" + +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + [[package]] name = "crypto-common" version = "0.1.7" @@ -965,6 +1079,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "zerocopy", +] + [[package]] name = "hashbrown" version = "0.15.5" @@ -1240,6 +1365,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + [[package]] name = "itertools" version = "0.14.0" @@ -1438,24 +1572,6 @@ dependencies = [ "tucana 0.0.75", ] -[[package]] -name = "manual" -version = "0.1.0" -dependencies = [ - "async-nats 0.50.0", - "clap", - "env_logger", - "log", - "prost", - "serde", - "serde_json", - "taurus-core", - "taurus-provider", - "tokio", - "tonic", - "tucana 0.0.76", -] - [[package]] name = "matchers" version = "0.2.0" @@ -1645,6 +1761,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + [[package]] name = "openssl-probe" version = "0.2.1" @@ -1746,6 +1868,16 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" +[[package]] +name = "page_size" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d5b2194ed13191c1999ae0704b7839fb18384fa22e49b57eeaa97d79ce40da" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "parking" version = "2.2.1" @@ -1792,7 +1924,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2ed4d5c6ae95e08ac768883c8401cf0e8deb4e6e1d6a4e1fd3d2ec4f0ec63200" dependencies = [ "heck", - "itertools", + "itertools 0.14.0", "prost", "prost-types", ] @@ -1880,6 +2012,34 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +[[package]] +name = "plotters" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747" +dependencies = [ + "num-traits", + "plotters-backend", + "plotters-svg", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "plotters-backend" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a" + +[[package]] +name = "plotters-svg" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670" +dependencies = [ + "plotters-backend", +] + [[package]] name = "portable-atomic" version = "1.14.0" @@ -1955,7 +2115,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools", + "itertools 0.14.0", "log", "multimap", "petgraph", @@ -1976,7 +2136,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.119", @@ -2108,6 +2268,26 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -2650,6 +2830,16 @@ dependencies = [ "tucana 0.0.76", ] +[[package]] +name = "taurus-bench" +version = "0.1.0" +dependencies = [ + "criterion", + "prost", + "taurus-core", + "tucana 0.0.76", +] + [[package]] name = "taurus-core" version = "0.1.0" @@ -2660,12 +2850,31 @@ dependencies = [ "log", "lupus", "rand 0.10.2", + "serde", "serde_json", + "tokio", "tucana 0.0.76", "ureq", "uuid", ] +[[package]] +name = "taurus-manual" +version = "0.1.0" +dependencies = [ + "async-nats 0.50.0", + "clap", + "env_logger", + "log", + "prost", + "serde_json", + "taurus-core", + "taurus-provider", + "tokio", + "tonic", + "tucana 0.0.76", +] + [[package]] name = "taurus-provider" version = "0.1.0" @@ -2685,6 +2894,18 @@ dependencies = [ "tucana 0.0.76", ] +[[package]] +name = "taurus-tests" +version = "0.1.0" +dependencies = [ + "async-trait", + "env_logger", + "log", + "serde_json", + "taurus-core", + "tucana 0.0.76", +] + [[package]] name = "tempfile" version = "3.27.0" @@ -2698,19 +2919,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "tests" -version = "0.1.0" -dependencies = [ - "async-trait", - "env_logger", - "log", - "serde", - "serde_json", - "taurus-core", - "tucana 0.0.76", -] - [[package]] name = "thiserror" version = "2.0.18" @@ -2780,6 +2988,16 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "tokio" version = "1.53.1" @@ -3402,6 +3620,22 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + [[package]] name = "winapi-util" version = "0.1.11" @@ -3411,6 +3645,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + [[package]] name = "windows-link" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index 9dc90af..2e6a644 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = [ "crates/taurus-core", "crates/taurus-manual", "crates/taurus", "crates/taurus-tests", "crates/taurus-provider"] +members = [ "crates/taurus-core", "crates/taurus-manual", "crates/taurus", "crates/taurus-tests", "crates/taurus-provider", "crates/taurus-bench"] resolver = "3" [workspace.package] @@ -32,3 +32,11 @@ path = "./crates/taurus-core" [workspace.dependencies.taurus-provider] path = "./crates/taurus-provider" + + +# `cargo build --profile profiling` — a release build with debug symbols, +# for flamegraph/sampling work. Kept separate from [profile.release] so +# ordinary release builds stay free of the extra symbol/binary-size cost. +[profile.profiling] +inherits = "release" +debug = true diff --git a/crates/taurus-bench/Cargo.toml b/crates/taurus-bench/Cargo.toml new file mode 100644 index 0000000..a349278 --- /dev/null +++ b/crates/taurus-bench/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "taurus-bench" +version.workspace = true +edition.workspace = true +publish = false + +[dependencies] +taurus-core = { workspace = true } +tucana = { workspace = true } + +[dev-dependencies] +criterion = "0.8" +prost = { workspace = true } + +[[bench]] +name = "engine_execution" +harness = false diff --git a/crates/taurus-bench/benches/engine_execution.rs b/crates/taurus-bench/benches/engine_execution.rs new file mode 100644 index 0000000..0eff7e2 --- /dev/null +++ b/crates/taurus-bench/benches/engine_execution.rs @@ -0,0 +1,246 @@ +//! Flow-level benchmarks for `ExecutionEngine`. +//! +//! `chain_add` measures per-node overhead across a linear chain (each node +//! reads the previous node's result), which is where `ValueStore` result +//! storage and reference resolution cost accumulates. `array_map` measures +//! per-iteration callback overhead inside a single `map` node, which is +//! where unconditional trace-label formatting accumulates independently of +//! node count. Both are run with `with_trace` on and off to isolate the +//! cost of Trace V2 collection. + +use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main}; +use std::collections::HashMap; +use taurus_core::runtime::engine::ExecutionEngine; +use tucana::shared::{ + ListValue, NodeFunction, NodeParameter, NodeValue, ReferenceValue, Struct, SubFlow, Value, + node_value, reference_value, reference_value::Target, sub_flow::ExecutionReference, + value::Kind, +}; + +fn int_value(value: i64) -> Value { + taurus_core::value::value_from_i64(value) +} + +fn literal_param(runtime_parameter_id: &str, value: Value) -> NodeParameter { + NodeParameter { + database_id: 0, + runtime_parameter_id: runtime_parameter_id.to_string(), + value: Some(NodeValue { + value: Some(node_value::Value::LiteralValue(value)), + }), + cast: None, + } +} + +fn reference_param(runtime_parameter_id: &str, target: Target) -> NodeParameter { + NodeParameter { + database_id: 0, + runtime_parameter_id: runtime_parameter_id.to_string(), + value: Some(NodeValue { + value: Some(node_value::Value::ReferenceValue(ReferenceValue { + target: Some(target), + paths: Vec::new(), + })), + }), + cast: None, + } +} + +fn thunk_param(runtime_parameter_id: &str, starting_node_id: i64) -> NodeParameter { + NodeParameter { + database_id: 0, + runtime_parameter_id: runtime_parameter_id.to_string(), + value: Some(NodeValue { + value: Some(node_value::Value::SubFlow(SubFlow { + input_schema: None, + output_schema: None, + signature: String::new(), + settings: Vec::new(), + execution_reference: Some(ExecutionReference::StartingNodeId(starting_node_id)), + })), + }), + cast: None, + } +} + +fn node( + database_id: i64, + runtime_function_id: &str, + parameters: Vec, + next_node_id: Option, +) -> NodeFunction { + NodeFunction { + database_id: Some(database_id), + runtime_function_id: runtime_function_id.to_string(), + parameters, + next_node_id, + definition_source: Some("taurus".to_string()), + } +} + +/// A chain of `n` "std::number::add" nodes; node `i` adds a constant to +/// node `i - 1`'s result. Exercises per-node ValueStore insert/lookup cost. +fn build_chain_flow(n: i64) -> (i64, Vec) { + let mut nodes = Vec::with_capacity(n as usize); + nodes.push(node( + 1, + "std::number::add", + vec![ + literal_param("a", int_value(0)), + literal_param("b", int_value(1)), + ], + if n > 1 { Some(2) } else { None }, + )); + for i in 2..=n { + nodes.push(node( + i, + "std::number::add", + vec![ + reference_param("a", Target::NodeId(i - 1)), + literal_param("b", int_value(1)), + ], + if i < n { Some(i + 1) } else { None }, + )); + } + (1, nodes) +} + +/// A single "std::list::map" node over an array of `n` items, with a +/// callback (node 2) that adds a constant to each item. Exercises +/// per-iteration callback overhead independent of node count. +fn build_map_flow(n: usize) -> (i64, Vec) { + let array = Value { + kind: Some(Kind::ListValue(ListValue { + values: (0..n as i64).map(int_value).collect(), + })), + }; + + let map_node = node( + 1, + "std::list::map", + vec![literal_param("array", array), thunk_param("transform", 2)], + None, + ); + let callback_node = node( + 2, + "std::number::add", + vec![ + reference_param( + "a", + Target::InputType(tucana::shared::InputType { + node_id: 1, + parameter_index: 1, + input_index: 0, + }), + ), + literal_param("b", int_value(1)), + ], + None, + ); + (1, vec![map_node, callback_node]) +} + +fn bench_chain(c: &mut Criterion) { + let mut group = c.benchmark_group("chain_add"); + for &size in &[10_i64, 50, 200] { + for &with_trace in &[false, true] { + let (start, nodes) = build_chain_flow(size); + group.bench_with_input( + BenchmarkId::new(if with_trace { "trace_on" } else { "trace_off" }, size), + &size, + |b, _| { + let engine = ExecutionEngine::new(); + b.iter(|| { + engine.execute_graph(start, nodes.clone(), None, None, None, with_trace) + }); + }, + ); + } + } + group.finish(); +} + +fn bench_array_map(c: &mut Criterion) { + let mut group = c.benchmark_group("array_map"); + for &size in &[10_usize, 100, 1_000] { + for &with_trace in &[false, true] { + let (start, nodes) = build_map_flow(size); + group.bench_with_input( + BenchmarkId::new(if with_trace { "trace_on" } else { "trace_off" }, size), + &size, + |b, _| { + let engine = ExecutionEngine::new(); + b.iter(|| { + engine.execute_graph(start, nodes.clone(), None, None, None, with_trace) + }); + }, + ); + } + } + group.finish(); +} + +/// Isolates `ValueStore::get` reference-resolution cost (Finding 3): a +/// struct field lookup by path, called repeatedly against a pre-populated +/// store, independent of the rest of the executor. +fn bench_value_store_get(c: &mut Criterion) { + use taurus_core::runtime::execution::value_store::ValueStore; + + let mut fields = HashMap::new(); + fields.insert( + "name".to_string(), + Value { + kind: Some(Kind::StringValue("benchmark".to_string())), + }, + ); + let flow_input = Value { + kind: Some(Kind::StructValue(Struct { fields })), + }; + let mut store = ValueStore::new(flow_input, false); + store.insert_success_with_timing(1, int_value(42), Vec::new(), 0, 0); + + c.bench_function("value_store_get_by_node_id", |b| { + b.iter(|| { + store.get(&ReferenceValue { + target: Some(reference_value::Target::NodeId(1)), + paths: Vec::new(), + }) + }); + }); +} + +criterion_group!( + benches, + bench_chain, + bench_array_map, + bench_value_store_get, + bench_compile_vs_encode +); +criterion_main!(benches); + +/// 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 +/// cache key is the flow's serialized content. Compares that serialization +/// cost against the full `execute_graph` (compile + run) cost it would be +/// competing against, for the same flow. +fn bench_compile_vs_encode(c: &mut Criterion) { + use prost::Message as _; + + let (start, nodes) = build_chain_flow(200); + let mut group = c.benchmark_group("compile_vs_encode"); + group.bench_function("encode_to_vec_cache_key/200", |b| { + b.iter(|| { + let mut buf = Vec::new(); + for n in &nodes { + n.encode(&mut buf).unwrap(); + } + buf + }); + }); + group.bench_function("full_execute_graph/200 (for comparison)", |b| { + let engine = ExecutionEngine::new(); + b.iter(|| engine.execute_graph(start, nodes.clone(), None, None, None, false)); + }); + group.finish(); +} diff --git a/crates/taurus-bench/examples/profile_workload.rs b/crates/taurus-bench/examples/profile_workload.rs new file mode 100644 index 0000000..f793c86 --- /dev/null +++ b/crates/taurus-bench/examples/profile_workload.rs @@ -0,0 +1,98 @@ +//! Standalone workload for CPU profiling (flamegraph/samply), separate from +//! the Criterion harness so sampled stacks aren't diluted by Criterion's own +//! measurement loop overhead. Runs the pathological `chain_add` case +//! (Finding 1: O(n^2) trace snapshotting) many times in a tight loop. + +use std::collections::HashMap; +use taurus_core::runtime::engine::ExecutionEngine; +use tucana::shared::{NodeFunction, NodeParameter, NodeValue, ReferenceValue, node_value}; + +fn int_value(value: i64) -> tucana::shared::Value { + taurus_core::value::value_from_i64(value) +} + +fn literal_param(runtime_parameter_id: &str, value: tucana::shared::Value) -> NodeParameter { + NodeParameter { + database_id: 0, + runtime_parameter_id: runtime_parameter_id.to_string(), + value: Some(NodeValue { + value: Some(node_value::Value::LiteralValue(value)), + }), + cast: None, + } +} + +fn reference_param(runtime_parameter_id: &str, node_id: i64) -> NodeParameter { + NodeParameter { + database_id: 0, + runtime_parameter_id: runtime_parameter_id.to_string(), + value: Some(NodeValue { + value: Some(node_value::Value::ReferenceValue(ReferenceValue { + target: Some(tucana::shared::reference_value::Target::NodeId(node_id)), + paths: Vec::new(), + })), + }), + cast: None, + } +} + +fn node( + database_id: i64, + parameters: Vec, + next_node_id: Option, +) -> NodeFunction { + NodeFunction { + database_id: Some(database_id), + runtime_function_id: "std::number::add".to_string(), + parameters, + next_node_id, + definition_source: Some("taurus".to_string()), + } +} + +fn build_chain_flow(n: i64) -> (i64, Vec) { + let mut nodes = Vec::with_capacity(n as usize); + nodes.push(node( + 1, + vec![ + literal_param("a", int_value(0)), + literal_param("b", int_value(1)), + ], + if n > 1 { Some(2) } else { None }, + )); + for i in 2..=n { + nodes.push(node( + i, + vec![ + reference_param("a", i - 1), + literal_param("b", int_value(1)), + ], + if i < n { Some(i + 1) } else { None }, + )); + } + (1, nodes) +} + +fn main() { + let with_trace = std::env::var("PROFILE_TRACE").is_ok(); + let iterations: u32 = std::env::var("PROFILE_ITERS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(300); + + let engine = ExecutionEngine::new(); + let (start, nodes) = build_chain_flow(200); + + let mut total = HashMap::new(); + for i in 0..iterations { + let (signal, _reason) = + engine.execute_graph(start, nodes.clone(), None, None, None, with_trace); + total.insert(i, signal.exit_reason()); + } + // Prevent the compiler from optimizing the loop away. + eprintln!( + "completed {} iterations, last exit_reason={:?}", + iterations, + total.get(&(iterations.saturating_sub(1))) + ); +} diff --git a/crates/taurus-core/Cargo.toml b/crates/taurus-core/Cargo.toml index 8d3c1bb..2bc6c46 100644 --- a/crates/taurus-core/Cargo.toml +++ b/crates/taurus-core/Cargo.toml @@ -12,5 +12,7 @@ lupus = { workspace = true } futures-lite = { workspace = true } async-trait = { workspace = true } uuid = { workspace = true } +serde = { workspace = true } serde_json = { workspace = true } ureq = { workspace = true } +tokio = { workspace = true } diff --git a/crates/taurus-core/src/fixtures.rs b/crates/taurus-core/src/fixtures.rs new file mode 100644 index 0000000..3be0f7e --- /dev/null +++ b/crates/taurus-core/src/fixtures.rs @@ -0,0 +1,114 @@ +//! JSON fixture loading shared by flow validation tooling binaries. +//! +//! `taurus-manual` and `taurus-tests` both load `ValidationFlow` fixtures +//! from JSON files and report pass/fail results in the same format; this +//! module is the single definition of that fixture format. + +use std::path::Path; + +use log::{error, info}; +use serde::Deserialize; +use tucana::shared::ValidationFlow; + +#[derive(Clone, Deserialize)] +pub struct Input { + pub input: Option, + pub expected_result: serde_json::Value, +} + +#[derive(Clone, Deserialize)] +pub struct Case { + pub name: String, + pub description: String, + pub inputs: Vec, + pub flow: ValidationFlow, + #[serde(default)] + pub remote: Option, +} + +#[derive(Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RemoteFixture { + pub target_service: String, + pub function_identifier: String, + pub result_parameter: String, +} + +#[derive(Clone, Deserialize)] +pub struct Cases { + pub cases: Vec, +} + +pub fn print_success(case: &Case) { + info!("test {} ... ok", case.name); +} + +pub fn print_failure(case: &Case, input: &Input, result: serde_json::Value) { + error!("test {} ... FAILED", case.name); + error!(" input: {:?}", input.input); + error!(" expected: {:?}", input.expected_result); + error!(" real_value: {:?}", result); + error!(" message: {}", case.description); +} + +fn get_test_case + std::fmt::Debug>(path: P) -> Option { + let content = match std::fs::read_to_string(&path) { + Ok(it) => it, + Err(err) => { + error!("Cannot read file ({:?}): {:?}", path, err); + return None; + } + }; + + match serde_json::from_str(&content) { + Ok(it) => it, + Err(err) => { + error!("Cannot read json ({:?}): {:?}", path, err); + None + } + } +} + +fn get_test_cases(path: &str) -> Cases { + let mut items = Vec::new(); + let dir = match std::fs::read_dir(path) { + Ok(d) => d, + Err(err) => { + panic!("Cannot open path: {:?}", err) + } + }; + + for entry in dir { + let entry = match entry { + Ok(it) => it, + Err(err) => { + error!("Cannot read entry: {:?}", err); + continue; + } + }; + let file_path = entry.path(); + items.push(match get_test_case(&file_path) { + Some(it) => it, + None => { + continue; + } + }); + } + + Cases { cases: items } +} + +impl Case { + pub fn from_path(path: &str) -> Self { + match get_test_case(path) { + Some(s) => s, + None => panic!("flow was not found"), + } + } +} + +impl Cases { + pub fn from_path(path: &str) -> Self { + get_test_cases(path) + } +} diff --git a/crates/taurus-core/src/lib.rs b/crates/taurus-core/src/lib.rs index a31a888..96efe78 100644 --- a/crates/taurus-core/src/lib.rs +++ b/crates/taurus-core/src/lib.rs @@ -1,9 +1,21 @@ -//! Taurus core runtime library. +//! Taurus core runtime library: the transport-agnostic flow execution engine. +//! Nothing here depends on NATS or gRPC, so it can be driven synchronously +//! (`taurus-tests`) or wired to a live transport (`taurus`, `taurus-manual` +//! via `taurus-provider`) without changes. The one exception is a thin +//! dependency on Tokio's runtime-detection API (not its networking types): +//! the `http` handler runs its blocking call via `block_in_place` when a +//! Tokio multi-thread runtime happens to be active, and calls it directly +//! otherwise -- so callers with no runtime at all (`taurus-tests`, +//! `taurus-manual --offline`) still work unchanged. //! -//! Exposes the runtime engine, strongly-typed execution contracts, and shared -//! value/error utilities used by runtime binaries. +//! See [`runtime::engine`] for the public execution API, [`types`] for the +//! shared signal/error vocabulary, and [`fixtures`]/[`normalize`] for the +//! JSON fixture and proto-value-normalization helpers shared by the runtime +//! binaries. +pub mod fixtures; mod handler; +pub mod normalize; pub mod runtime; pub mod time; pub mod types; diff --git a/crates/taurus-core/src/normalize.rs b/crates/taurus-core/src/normalize.rs new file mode 100644 index 0000000..c7df41c --- /dev/null +++ b/crates/taurus-core/src/normalize.rs @@ -0,0 +1,60 @@ +//! Fills missing/`None` proto value fields with explicit null defaults. +//! +//! Proto3 leaves optional scalar fields unset rather than null; consumers +//! that expect every field to be present (e.g. JSON reporting, debug +//! printing) need `None` normalized to an explicit `NullValue` first. + +use tucana::shared::{NodeExecutionResult, Value, node_execution_result, value::Kind}; + +pub fn null_value() -> Value { + Value { + kind: Some(Kind::NullValue(0)), + } +} + +pub fn normalize_value(value: &mut Value) { + match &mut value.kind { + Some(Kind::StructValue(struct_value)) => { + for field in struct_value.fields.values_mut() { + normalize_value(field); + } + } + Some(Kind::ListValue(list_value)) => { + for item in &mut list_value.values { + normalize_value(item); + } + } + Some(Kind::NumberValue(number)) if number.number.is_none() => { + value.kind = Some(Kind::NullValue(0)); + } + Some(_) => {} + None => { + value.kind = Some(Kind::NullValue(0)); + } + } +} + +pub fn normalize_node_execution_result(result: &mut NodeExecutionResult) { + for parameter_result in &mut result.parameter_results { + match &mut parameter_result.value { + Some(value) => normalize_value(value), + None => { + parameter_result.value = Some(null_value()); + } + } + } + + match &mut result.result { + Some(node_execution_result::Result::Success(value)) => normalize_value(value), + Some(node_execution_result::Result::Error(error)) => { + if let Some(details) = &mut error.details { + for value in details.fields.values_mut() { + normalize_value(value); + } + } + } + None => { + result.result = Some(node_execution_result::Result::Success(null_value())); + } + } +} diff --git a/crates/taurus-core/src/runtime/engine.rs b/crates/taurus-core/src/runtime/engine.rs index 0ba6fd7..7edf79a 100644 --- a/crates/taurus-core/src/runtime/engine.rs +++ b/crates/taurus-core/src/runtime/engine.rs @@ -52,84 +52,6 @@ impl ExecutionEngine { } } - /// Execute an `ExecutionFlow`. - pub fn execute_flow( - &self, - flow: ExecutionFlow, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> (Signal, ExitReason) { - let report = block_on(self.execute_flow_with_execution_id_report_async( - ExecutionId::new_v4(), - flow, - remote, - respond_emitter, - with_trace, - )); - (report.signal, report.exit_reason) - } - - /// Execute an `ExecutionFlow` asynchronously. - pub async fn execute_flow_async( - &self, - flow: ExecutionFlow, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> (Signal, ExitReason) { - let report = self - .execute_flow_with_execution_id_report_async( - ExecutionId::new_v4(), - flow, - remote, - respond_emitter, - with_trace, - ) - .await; - (report.signal, report.exit_reason) - } - - /// Execute an `ExecutionFlow` with a caller-provided execution id. - pub fn execute_flow_with_execution_id( - &self, - execution_id: ExecutionId, - flow: ExecutionFlow, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> (Signal, ExitReason) { - let report = block_on(self.execute_flow_with_execution_id_report_async( - execution_id, - flow, - remote, - respond_emitter, - with_trace, - )); - (report.signal, report.exit_reason) - } - - /// Execute an `ExecutionFlow` asynchronously with a caller-provided execution id. - pub async fn execute_flow_with_execution_id_async( - &self, - execution_id: ExecutionId, - flow: ExecutionFlow, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> (Signal, ExitReason) { - let report = self - .execute_flow_with_execution_id_report_async( - execution_id, - flow, - remote, - respond_emitter, - with_trace, - ) - .await; - (report.signal, report.exit_reason) - } - /// Execute an `ExecutionFlow` and return the final signal plus per-node execution results. pub fn execute_flow_report( &self, @@ -147,42 +69,6 @@ impl ExecutionEngine { )) } - /// Execute an `ExecutionFlow` asynchronously and return per-node execution results. - pub async fn execute_flow_report_async( - &self, - flow: ExecutionFlow, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> EngineExecutionReport { - self.execute_flow_with_execution_id_report_async( - ExecutionId::new_v4(), - flow, - remote, - respond_emitter, - with_trace, - ) - .await - } - - /// Execute an `ExecutionFlow` with a caller-provided execution id and return per-node results. - pub fn execute_flow_with_execution_id_report( - &self, - execution_id: ExecutionId, - flow: ExecutionFlow, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> EngineExecutionReport { - block_on(self.execute_flow_with_execution_id_report_async( - execution_id, - flow, - remote, - respond_emitter, - with_trace, - )) - } - /// Execute an `ExecutionFlow` asynchronously with a caller-provided execution id and return per-node results. pub async fn execute_flow_with_execution_id_report_async( &self, @@ -215,8 +101,9 @@ impl ExecutionEngine { respond_emitter: Option<&dyn RespondEmitter>, with_trace: bool, ) -> (Signal, ExitReason) { - let report = block_on(self.execute_graph_with_execution_id_report_async( + let report = block_on(self.execute_graph_with_project_id_report_async( ExecutionId::new_v4(), + 0, start_node_id, node_functions, flow_input, @@ -227,78 +114,6 @@ impl ExecutionEngine { (report.signal, report.exit_reason) } - /// Execute a graph asynchronously. - pub async fn execute_graph_async( - &self, - start_node_id: i64, - node_functions: Vec, - flow_input: Option, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> (Signal, ExitReason) { - let report = self - .execute_graph_with_execution_id_report_async( - ExecutionId::new_v4(), - start_node_id, - node_functions, - flow_input, - remote, - respond_emitter, - with_trace, - ) - .await; - (report.signal, report.exit_reason) - } - - /// Execute a graph described by node list and start node with a caller-provided execution id. - pub fn execute_graph_with_execution_id( - &self, - execution_id: ExecutionId, - start_node_id: i64, - node_functions: Vec, - flow_input: Option, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> (Signal, ExitReason) { - let report = block_on(self.execute_graph_with_execution_id_report_async( - execution_id, - start_node_id, - node_functions, - flow_input, - remote, - respond_emitter, - with_trace, - )); - (report.signal, report.exit_reason) - } - - /// Execute a graph asynchronously with a caller-provided execution id. - pub async fn execute_graph_with_execution_id_async( - &self, - execution_id: ExecutionId, - start_node_id: i64, - node_functions: Vec, - flow_input: Option, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> (Signal, ExitReason) { - let report = self - .execute_graph_with_execution_id_report_async( - execution_id, - start_node_id, - node_functions, - flow_input, - remote, - respond_emitter, - with_trace, - ) - .await; - (report.signal, report.exit_reason) - } - /// Execute a graph and return the final signal plus per-node execution results. pub fn execute_graph_report( &self, @@ -309,74 +124,8 @@ impl ExecutionEngine { respond_emitter: Option<&dyn RespondEmitter>, with_trace: bool, ) -> EngineExecutionReport { - block_on(self.execute_graph_with_execution_id_report_async( + block_on(self.execute_graph_with_project_id_report_async( ExecutionId::new_v4(), - start_node_id, - node_functions, - flow_input, - remote, - respond_emitter, - with_trace, - )) - } - - /// Execute a graph asynchronously and return per-node execution results. - pub async fn execute_graph_report_async( - &self, - start_node_id: i64, - node_functions: Vec, - flow_input: Option, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> EngineExecutionReport { - self.execute_graph_with_execution_id_report_async( - ExecutionId::new_v4(), - start_node_id, - node_functions, - flow_input, - remote, - respond_emitter, - with_trace, - ) - .await - } - - /// Execute a graph with a caller-provided execution id and return per-node results. - pub fn execute_graph_with_execution_id_report( - &self, - execution_id: ExecutionId, - start_node_id: i64, - node_functions: Vec, - flow_input: Option, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> EngineExecutionReport { - block_on(self.execute_graph_with_execution_id_report_async( - execution_id, - start_node_id, - node_functions, - flow_input, - remote, - respond_emitter, - with_trace, - )) - } - - /// Execute a graph asynchronously with a caller-provided execution id and return per-node results. - pub async fn execute_graph_with_execution_id_report_async( - &self, - execution_id: ExecutionId, - start_node_id: i64, - node_functions: Vec, - flow_input: Option, - remote: Option<&dyn RemoteRuntime>, - respond_emitter: Option<&dyn RespondEmitter>, - with_trace: bool, - ) -> EngineExecutionReport { - self.execute_graph_with_project_id_report_async( - execution_id, 0, start_node_id, node_functions, @@ -384,8 +133,7 @@ impl ExecutionEngine { remote, respond_emitter, with_trace, - ) - .await + )) } async fn execute_graph_with_project_id_report_async( @@ -403,10 +151,7 @@ impl ExecutionEngine { emitter.emit(execution_id, EmitType::StartingExec, null_value()); } - let mut value_store = match flow_input { - Some(v) => ValueStore::new(v), - None => ValueStore::default(), - }; + let mut value_store = ValueStore::new(flow_input.unwrap_or_default(), with_trace); let compiled = match compile_flow(project_id, start_node_id, node_functions) { Ok(plan) => plan, diff --git a/crates/taurus-core/src/runtime/engine/executor.rs b/crates/taurus-core/src/runtime/engine/executor.rs index fffac96..580e50f 100644 --- a/crates/taurus-core/src/runtime/engine/executor.rs +++ b/crates/taurus-core/src/runtime/engine/executor.rs @@ -704,7 +704,7 @@ impl<'a> EngineExecutor<'a> { ); args.push(Argument::Eval(value.clone())); } - CompiledArg::Reference(reference) => match value_store.get(reference.clone()) { + CompiledArg::Reference(reference) => match value_store.get(reference) { ValueStoreResult::Success(value) => { self.trace_record_arg( frame_id, @@ -1258,7 +1258,7 @@ fn resolve_function_setting( paths: Vec::new(), }; - match value_store.get(reference) { + match value_store.get(&reference) { ValueStoreResult::Success(value) => { if is_null_value(&value) && let Some(default_value) = setting.default_value.clone() diff --git a/crates/taurus-core/src/runtime/execution/mod.rs b/crates/taurus-core/src/runtime/execution/mod.rs index 753a755..b178080 100644 --- a/crates/taurus-core/src/runtime/execution/mod.rs +++ b/crates/taurus-core/src/runtime/execution/mod.rs @@ -3,9 +3,7 @@ //! These types are owned by the execution engine lifecycle and are not part of //! the transport-level flow contracts. -pub mod registry; pub mod render; -pub mod store; pub mod trace; pub mod tracer; pub mod value_store; diff --git a/crates/taurus-core/src/runtime/execution/registry.rs b/crates/taurus-core/src/runtime/execution/registry.rs deleted file mode 100644 index 213669d..0000000 --- a/crates/taurus-core/src/runtime/execution/registry.rs +++ /dev/null @@ -1,19 +0,0 @@ -//! Registry metadata types for runtime handler discovery. - -use std::collections::HashMap; - -use crate::types::execution::signature::HandlerSignature; - -/// Static registry entry metadata. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct HandlerRegistration { - pub handler_id: String, - pub signature: HandlerSignature, - pub description: Option, -} - -/// Read-only handler metadata registry. -#[derive(Debug, Clone, Default)] -pub struct HandlerRegistry { - pub handlers: HashMap, -} diff --git a/crates/taurus-core/src/runtime/execution/store.rs b/crates/taurus-core/src/runtime/execution/store.rs deleted file mode 100644 index 3cde097..0000000 --- a/crates/taurus-core/src/runtime/execution/store.rs +++ /dev/null @@ -1,25 +0,0 @@ -//! Mutable execution state for a single flow run. - -use std::collections::HashMap; - -use tucana::shared::{NodeExecutionResult, Value}; - -use crate::types::execution::ids::{FrameId, NodeId}; - -/// Input slot key for runtime-provided temporary inputs (for iterators/predicates). -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct InputSlotKey { - pub node_id: NodeId, - pub parameter_index: i32, - pub input_index: i32, -} - -/// Store that captures mutable runtime execution state. -#[derive(Debug, Clone, Default)] -pub struct ExecutionStore { - pub node_results: HashMap, - pub input_slots: HashMap, - pub flow_input: Option, - pub current_node: Option, - pub frame_stack: Vec, -} diff --git a/crates/taurus-core/src/runtime/execution/value_store.rs b/crates/taurus-core/src/runtime/execution/value_store.rs index a2e7966..48509d9 100644 --- a/crates/taurus-core/src/runtime/execution/value_store.rs +++ b/crates/taurus-core/src/runtime/execution/value_store.rs @@ -1,6 +1,7 @@ //! Mutable value store used by runtime execution to resolve references. use std::collections::HashMap; +use std::sync::Arc; use tucana::shared::node_execution_result::{Id as TucanaNodeResultId, Result as TucanaNodeResult}; use tucana::shared::{ @@ -20,16 +21,22 @@ pub enum ValueStoreResult { #[derive(Default)] pub struct ValueStore { - latest_results: HashMap, - result_history: Vec, + // Arc-wrapped so `insert_node_result` records the same result into both + // maps via a cheap refcount bump instead of a deep clone (each node + // result carries its own parameter_results and value trees). + latest_results: HashMap>, + result_history: Vec>, input_types: HashMap, flow_input: Value, current_node_id: i64, runtime_trace_labels: Vec, + // Lets push_runtime_trace_label skip building a label when nothing + // will read it back (see that method for why this matters). + trace_enabled: bool, } impl ValueStore { - pub fn new(flow_input: Value) -> Self { + pub fn new(flow_input: Value, trace_enabled: bool) -> Self { Self { latest_results: HashMap::new(), result_history: Vec::new(), @@ -37,6 +44,7 @@ impl ValueStore { flow_input, current_node_id: 0, runtime_trace_labels: Vec::new(), + trace_enabled, } } @@ -48,13 +56,13 @@ impl ValueStore { self.current_node_id = node_id; } - pub fn get(&mut self, reference: ReferenceValue) -> ValueStoreResult { - let target = match reference.target { + pub fn get(&mut self, reference: &ReferenceValue) -> ValueStoreResult { + let target = match &reference.target { Some(target) => target, None => return ValueStoreResult::NotFound, }; - let result = match target { + let result = match *target { tucana::shared::reference_value::Target::FlowInput(_) => self.get_flow_input(), tucana::shared::reference_value::Target::NodeId(id) => self.get_result(id), tucana::shared::reference_value::Target::InputType(input_type) => { @@ -68,7 +76,7 @@ impl ValueStore { if let ValueStoreResult::Success(value) = result { let mut current = value; - for path in reference.paths { + for path in &reference.paths { if let Some(index) = path.array_index { match current.kind { Some(ref kind) => match kind { @@ -82,11 +90,11 @@ impl ValueStore { } } - if let Some(field_name) = path.path { + if let Some(field_name) = &path.path { match current.kind { Some(ref kind) => { if let Kind::StructValue(struct_value) = kind { - match struct_value.fields.get(&field_name) { + match struct_value.fields.get(field_name) { Some(item) => current = item.clone(), None => return ValueStoreResult::NotFound, } @@ -188,6 +196,7 @@ impl ValueStore { pub fn insert_node_result(&mut self, id: i64, mut result: NodeExecutionResult) { result.id = Some(TucanaNodeResultId::NodeId(id)); + let result = Arc::new(result); self.latest_results.insert(id, result.clone()); self.result_history.push(result); } @@ -200,13 +209,13 @@ impl ValueStore { started_at: i64, finished_at: i64, ) { - self.result_history.push(NodeExecutionResult { + self.result_history.push(Arc::new(NodeExecutionResult { started_at, finished_at, parameter_results, id: Some(TucanaNodeResultId::FunctionIdentifier(id)), result: Some(TucanaNodeResult::Success(value)), - }); + })); } pub fn insert_function_error_with_timing( @@ -217,21 +226,35 @@ impl ValueStore { started_at: i64, finished_at: i64, ) { - self.result_history.push(NodeExecutionResult { + self.result_history.push(Arc::new(NodeExecutionResult { started_at, finished_at, parameter_results, id: Some(TucanaNodeResultId::FunctionIdentifier(id)), result: Some(TucanaNodeResult::Error(runtime_error.as_tucana_error())), - }); + })); } - pub fn node_execution_results(&self) -> Vec { - self.result_history.clone() + /// Takes ownership of the accumulated result history, leaving it empty. + /// Callers use this once, at report time, after which the store is + /// discarded. Each entry unwraps its `Arc` for free (no clone) once + /// `latest_results` (the only other owner) is cleared first. + pub fn node_execution_results(&mut self) -> Vec { + self.latest_results.clear(); + std::mem::take(&mut self.result_history) + .into_iter() + .map(|entry| Arc::try_unwrap(entry).unwrap_or_else(|arc| (*arc).clone())) + .collect() } - pub fn push_runtime_trace_label(&mut self, label: String) { - self.runtime_trace_labels.push(label); + /// Builds and records a trace label only when tracing is active; the + /// label is otherwise discarded by `trace_link_child` unread, so callers + /// (array/comparator callbacks, in a hot per-iteration loop) pass the + /// formatting as a closure instead of paying for it unconditionally. + pub fn push_runtime_trace_label(&mut self, label: impl FnOnce() -> String) { + if self.trace_enabled { + self.runtime_trace_labels.push(label()); + } } pub fn pop_runtime_trace_label(&mut self) -> Option { @@ -250,7 +273,7 @@ impl ValueStore { }; results.push(StoreResultEntry { node_id: *node_id, - result: result.clone(), + result: (**result).clone(), preview, }); } diff --git a/crates/taurus-core/src/runtime/functions/array.rs b/crates/taurus-core/src/runtime/functions/array.rs index 41780c7..b4f10d3 100644 --- a/crates/taurus-core/src/runtime/functions/array.rs +++ b/crates/taurus-core/src/runtime/functions/array.rs @@ -107,7 +107,7 @@ fn run_with_unary_input( thunk_node: &crate::handler::argument::Thunk, ) -> Signal { ctx.insert_input_type(input_type, item.clone()); - ctx.push_runtime_trace_label(format!("iter={} value={}", iter_index, preview_value(item))); + ctx.push_runtime_trace_label(|| format!("iter={} value={}", iter_index, preview_value(item))); let signal = run(thunk_node, ctx); ctx.clear_input_type(input_type); signal @@ -125,12 +125,14 @@ fn run_with_binary_inputs( ) -> Signal { ctx.insert_input_type(left_input, left.clone()); ctx.insert_input_type(right_input, right.clone()); - ctx.push_runtime_trace_label(format!( - "cmp#{} a={} b={}", - cmp_index, - preview_value(left), - preview_value(right) - )); + ctx.push_runtime_trace_label(|| { + format!( + "cmp#{} a={} b={}", + cmp_index, + preview_value(left), + preview_value(right) + ) + }); let signal = run(thunk_node, ctx); ctx.clear_input_type(left_input); ctx.clear_input_type(right_input); diff --git a/crates/taurus-core/src/runtime/functions/control.rs b/crates/taurus-core/src/runtime/functions/control.rs index 94954df..2c0eb05 100644 --- a/crates/taurus-core/src/runtime/functions/control.rs +++ b/crates/taurus-core/src/runtime/functions/control.rs @@ -69,7 +69,7 @@ fn r#if( if *bool { // Branch execution is delegated to the executor through `run`. - ctx.push_runtime_trace_label("branch=if".to_string()); + ctx.push_runtime_trace_label(|| "branch=if".to_string()); run(if_pointer, ctx) } else { Signal::Success(Value { @@ -99,10 +99,10 @@ fn if_else( }; if *bool { - ctx.push_runtime_trace_label("branch=if".to_string()); + ctx.push_runtime_trace_label(|| "branch=if".to_string()); run(if_pointer, ctx) } else { - ctx.push_runtime_trace_label("branch=else".to_string()); + ctx.push_runtime_trace_label(|| "branch=else".to_string()); run(else_pointer, ctx) } } diff --git a/crates/taurus-core/src/runtime/functions/http.rs b/crates/taurus-core/src/runtime/functions/http.rs index 82b44ba..86b46ed 100644 --- a/crates/taurus-core/src/runtime/functions/http.rs +++ b/crates/taurus-core/src/runtime/functions/http.rs @@ -29,6 +29,19 @@ fn fail(category: &str, message: impl Into) -> Signal { Signal::Failure(RuntimeError::new("T-STD-00001", category, message)) } +/// 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). +fn run_blocking(f: impl FnOnce() -> R) -> R { + if tokio::runtime::Handle::try_current().is_ok() { + tokio::task::block_in_place(f) + } else { + f() + } +} + fn respond( args: &[Argument], _ctx: &mut ValueStore, @@ -141,6 +154,12 @@ fn headers_from_value(value: &Value) -> Result { } } +/// The `http::request::send` handler. The actual request is a +/// synchronous, blocking `ureq` call; when running inside a Tokio +/// multi-thread runtime (true for the `taurus` service, false for +/// `taurus-tests` and `taurus-manual --offline`, which call the engine +/// with no runtime at all) it runs via `block_in_place` so it doesn't +/// stall a shared async worker thread for its duration. fn send_request( args: &[Argument], _ctx: &mut ValueStore, @@ -213,12 +232,14 @@ fn send_request( ); } }; - request - .with_default_agent() - .configure() - .http_status_as_error(false) - .allow_non_standard_methods(true) - .run() + run_blocking(|| { + request + .with_default_agent() + .configure() + .http_status_as_error(false) + .allow_non_standard_methods(true) + .run() + }) } None => { let request = match request_builder.body(()) { @@ -230,12 +251,14 @@ fn send_request( ); } }; - request - .with_default_agent() - .configure() - .http_status_as_error(false) - .allow_non_standard_methods(true) - .run() + run_blocking(|| { + request + .with_default_agent() + .configure() + .http_status_as_error(false) + .allow_non_standard_methods(true) + .run() + }) } }; diff --git a/crates/taurus-core/src/types/errors/error.rs b/crates/taurus-core/src/types/errors/error.rs deleted file mode 100644 index bdec273..0000000 --- a/crates/taurus-core/src/types/errors/error.rs +++ /dev/null @@ -1,236 +0,0 @@ -//! General (application-level) errors that can occur outside pure node execution. -//! -//! Every variant can be converted into a [`RuntimeError`] to unify reporting. - -use std::collections::HashMap; -use std::error::Error as StdError; -use std::fmt::{Display, Formatter}; - -use tucana::shared::Value; -use tucana::shared::value::Kind::StringValue; - -use crate::types::errors::runtime_error::RuntimeError; - -/// Application-layer failures that should still be reportable as runtime failures. -#[derive(Debug, Clone)] -pub enum Error { - /// Invalid or missing runtime configuration. - Configuration { - message: String, - details: HashMap, - }, - /// Invalid application state transition. - State { - message: String, - details: HashMap, - }, - /// Failed communication with dependency or transport layer. - Transport { - dependency: String, - message: String, - details: HashMap, - }, - /// Failed serialization/deserialization. - Serialization { - format: String, - message: String, - details: HashMap, - }, - /// Catch-all internal application error. - Internal { - message: String, - details: HashMap, - }, -} - -impl Error { - /// Build a configuration error with optional structured details. - pub fn configuration(message: impl Into) -> Self { - Self::Configuration { - message: message.into(), - details: HashMap::new(), - } - } - - /// Build an invalid state error with optional structured details. - pub fn state(message: impl Into) -> Self { - Self::State { - message: message.into(), - details: HashMap::new(), - } - } - - /// Build a transport/dependency error with optional structured details. - pub fn transport(dependency: impl Into, message: impl Into) -> Self { - Self::Transport { - dependency: dependency.into(), - message: message.into(), - details: HashMap::new(), - } - } - - /// Build a serialization error with optional structured details. - pub fn serialization(format: impl Into, message: impl Into) -> Self { - Self::Serialization { - format: format.into(), - message: message.into(), - details: HashMap::new(), - } - } - - /// Build an internal application error with optional structured details. - pub fn internal(message: impl Into) -> Self { - Self::Internal { - message: message.into(), - details: HashMap::new(), - } - } - - /// Attach a detail entry to this error. - pub fn with_detail(mut self, key: impl Into, value: Value) -> Self { - match &mut self { - Error::Configuration { details, .. } - | Error::State { details, .. } - | Error::Transport { details, .. } - | Error::Serialization { details, .. } - | Error::Internal { details, .. } => { - details.insert(key.into(), value); - } - } - self - } -} - -impl From for RuntimeError { - fn from(value: Error) -> Self { - match value { - Error::Configuration { message, details } => { - let mut err = RuntimeError::with_code( - "T-CORE-000301".to_string(), - "Configuration".to_string(), - message, - ); - err.details.extend(details); - err - } - Error::State { message, details } => { - let mut err = RuntimeError::with_code( - "T-CORE-000302".to_string(), - "State".to_string(), - message, - ); - err.details.extend(details); - err - } - Error::Transport { - dependency, - message, - details, - } => { - let mut err = RuntimeError::with_code( - "T-CORE-000303".to_string(), - "Transport".to_string(), - message, - ) - .with_detail( - "dependency".to_string(), - Value { - kind: Some(StringValue(dependency)), - }, - ); - err.details.extend(details); - err - } - Error::Serialization { - format, - message, - details, - } => { - let mut err = RuntimeError::with_code( - "T-CORE-000304".to_string(), - "Serialization".to_string(), - message, - ) - .with_detail( - "format".to_string(), - Value { - kind: Some(StringValue(format)), - }, - ); - err.details.extend(details); - err - } - Error::Internal { message, details } => { - let mut err = RuntimeError::with_code( - "T-CORE-000399".to_string(), - "Internal".to_string(), - message, - ); - err.details.extend(details); - err - } - } - } -} - -impl StdError for Error {} - -impl Display for Error { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - match self { - Error::Configuration { message, .. } => write!(f, "Configuration error: {message}"), - Error::State { message, .. } => write!(f, "State error: {message}"), - Error::Transport { - dependency, - message, - .. - } => write!(f, "Transport error ({dependency}): {message}"), - Error::Serialization { - format, message, .. - } => write!(f, "Serialization error ({format}): {message}"), - Error::Internal { message, .. } => write!(f, "Internal error: {message}"), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use tucana::shared::{Struct, value::Kind::StructValue}; - - #[test] - fn app_error_converts_to_runtime_error_with_expected_code_and_category() { - let app_err = Error::transport("nats", "connection lost").with_detail( - "subject", - Value { - kind: Some(StringValue("execution.*".to_string())), - }, - ); - - let runtime_error: RuntimeError = app_err.into(); - - assert_eq!(runtime_error.code, "T-CORE-000303"); - assert_eq!(runtime_error.category, "Transport"); - assert_eq!(runtime_error.message, "connection lost"); - assert!(runtime_error.details.contains_key("dependency")); - assert!(runtime_error.details.contains_key("subject")); - } - - #[test] - fn runtime_error_value_contains_required_struct_fields() { - let runtime_error: RuntimeError = Error::internal("boom").into(); - let value = runtime_error.as_value(); - - let Some(StructValue(Struct { fields })) = value.kind else { - panic!("expected struct value"); - }; - - assert!(fields.contains_key("code")); - assert!(fields.contains_key("category")); - assert!(fields.contains_key("message")); - assert!(fields.contains_key("timestamp")); - assert!(fields.contains_key("version")); - assert!(fields.contains_key("dependencies")); - assert!(fields.contains_key("details")); - } -} diff --git a/crates/taurus-core/src/types/errors/mod.rs b/crates/taurus-core/src/types/errors/mod.rs index 4e6145c..6563fc9 100644 --- a/crates/taurus-core/src/types/errors/mod.rs +++ b/crates/taurus-core/src/types/errors/mod.rs @@ -1,4 +1,3 @@ //! Error types shared across Taurus runtime execution and application layers. -pub mod error; pub mod runtime_error; diff --git a/crates/taurus-core/src/types/execution/bindings.rs b/crates/taurus-core/src/types/execution/bindings.rs deleted file mode 100644 index 25efcfb..0000000 --- a/crates/taurus-core/src/types/execution/bindings.rs +++ /dev/null @@ -1,54 +0,0 @@ -//! Argument binding and expression model for node execution. - -use tucana::shared::Value; - -use crate::types::execution::ids::{NodeId, ParameterId}; - -/// Path segment for nested lookups inside values. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum ValuePathSegment { - /// Select list element by index. - Index(usize), - /// Select object field by key. - Field(String), -} - -/// Source reference used by argument expressions. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum ReferenceSource { - /// Input value of the overall flow. - FlowInput, - /// Output result of another node. - NodeResult(NodeId), - /// Runtime input slot (used by iterators/predicates). - InputSlot { - node_id: NodeId, - parameter_index: i32, - input_index: i32, - }, -} - -/// Read expression from execution state. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ValueReference { - pub source: ReferenceSource, - pub path: Vec, -} - -/// Argument expression bound to a node parameter. -#[derive(Debug, Clone, PartialEq)] -pub enum ArgumentExpr { - /// Constant value literal. - ValueLiteral(Value), - /// Value resolved from runtime references. - Reference(ValueReference), - /// Deferred execution entry point (lazy function parameter). - DeferredCall(NodeId), -} - -/// Argument binding for one parameter. -#[derive(Debug, Clone, PartialEq)] -pub struct InputBinding { - pub parameter_id: ParameterId, - pub expression: ArgumentExpr, -} diff --git a/crates/taurus-core/src/types/execution/flow_ir.rs b/crates/taurus-core/src/types/execution/flow_ir.rs deleted file mode 100644 index 32d4305..0000000 --- a/crates/taurus-core/src/types/execution/flow_ir.rs +++ /dev/null @@ -1,33 +0,0 @@ -//! Static flow graph representation used by the execution engine. - -use std::collections::HashMap; - -use crate::types::execution::bindings::InputBinding; -use crate::types::execution::ids::{FlowId, NodeId}; - -/// Node execution location kind. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum NodeKind { - /// Node executes in the local Taurus runtime. - Local, - /// Node executes in a remote runtime/service. - Remote { service: String }, -} - -/// Node invocation metadata and wiring. -#[derive(Debug, Clone, PartialEq)] -pub struct FlowNode { - pub id: NodeId, - pub kind: NodeKind, - pub handler_id: String, - pub next: Option, - pub bindings: Vec, -} - -/// Immutable flow graph passed to the executor. -#[derive(Debug, Clone, PartialEq)] -pub struct FlowGraph { - pub id: FlowId, - pub start_node: NodeId, - pub nodes: HashMap, -} diff --git a/crates/taurus-core/src/types/execution/ids.rs b/crates/taurus-core/src/types/execution/ids.rs deleted file mode 100644 index 1fa558e..0000000 --- a/crates/taurus-core/src/types/execution/ids.rs +++ /dev/null @@ -1,53 +0,0 @@ -//! Identifier newtypes for execution-domain entities. - -use std::fmt::{Display, Formatter}; - -/// Flow identifier. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct FlowId(pub i64); - -/// Node identifier in a flow graph. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct NodeId(pub i64); - -/// Frame identifier for nested call execution. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct FrameId(pub u64); - -/// Parameter identifier for handler signatures. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct ParameterId(pub i32); - -/// Unique execution run identifier. -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -pub struct ExecutionId(pub String); - -impl Display for FlowId { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.0) - } -} - -impl Display for NodeId { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.0) - } -} - -impl Display for FrameId { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.0) - } -} - -impl Display for ParameterId { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.0) - } -} - -impl Display for ExecutionId { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.0) - } -} diff --git a/crates/taurus-core/src/types/execution/mod.rs b/crates/taurus-core/src/types/execution/mod.rs deleted file mode 100644 index bb081a8..0000000 --- a/crates/taurus-core/src/types/execution/mod.rs +++ /dev/null @@ -1,11 +0,0 @@ -//! Structured execution model types. -//! -//! This module provides a clean, runtime-oriented vocabulary for: -//! - flow graph representation -//! - parameter signatures and argument bindings -//! - stable identifiers - -pub mod bindings; -pub mod flow_ir; -pub mod ids; -pub mod signature; diff --git a/crates/taurus-core/src/types/execution/signature.rs b/crates/taurus-core/src/types/execution/signature.rs deleted file mode 100644 index 44a46ac..0000000 --- a/crates/taurus-core/src/types/execution/signature.rs +++ /dev/null @@ -1,28 +0,0 @@ -//! Handler/function signature model for runtime execution. - -use crate::types::execution::ids::ParameterId; - -/// How a parameter argument should be evaluated by the executor. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum EvaluationMode { - /// Argument is resolved before invoking the handler. - Eager, - /// Argument is provided as deferred executable call/input expression. - Deferred, -} - -/// Single parameter contract for a handler. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ParameterSpec { - pub id: ParameterId, - pub name: String, - pub evaluation_mode: EvaluationMode, - pub required: bool, -} - -/// Complete handler contract used by registry and runtime checks. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct HandlerSignature { - pub handler_id: String, - pub parameters: Vec, -} diff --git a/crates/taurus-core/src/types/mod.rs b/crates/taurus-core/src/types/mod.rs index 376ff00..0987f79 100644 --- a/crates/taurus-core/src/types/mod.rs +++ b/crates/taurus-core/src/types/mod.rs @@ -3,6 +3,5 @@ //! Split by concern: execution model, signal vocabulary, and error contracts. pub mod errors; -pub mod execution; pub mod exit_reason; pub mod signal; diff --git a/crates/taurus-manual/Cargo.toml b/crates/taurus-manual/Cargo.toml index 799ab33..998b8c4 100644 --- a/crates/taurus-manual/Cargo.toml +++ b/crates/taurus-manual/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "manual" +name = "taurus-manual" version.workspace = true edition.workspace = true @@ -10,7 +10,6 @@ taurus-provider = { workspace = true } log = { workspace = true } env_logger = { workspace = true } serde_json = { workspace = true } -serde = { workspace = true } prost = { workspace = true } tonic = { workspace = true } tokio = { workspace = true } diff --git a/crates/taurus-manual/src/main.rs b/crates/taurus-manual/src/main.rs index 19f30cc..99c6e43 100644 --- a/crates/taurus-manual/src/main.rs +++ b/crates/taurus-manual/src/main.rs @@ -1,11 +1,16 @@ -use std::path::Path; +//! CLI tool for running a single flow fixture by hand: offline through the +//! engine directly, live against a NATS remote/emitter, or queued onto a +//! running `taurus` instance's `execution.*` subscription. See [`Args`] for +//! the mode flags and `taurus_core::fixtures` for the fixture format shared +//! with `taurus-tests`. + use std::time::{Duration, Instant}; use clap::Parser; -use log::error; use log::info; use prost::Message; -use serde::Deserialize; +use taurus_core::fixtures::Case; +use taurus_core::normalize::normalize_node_execution_result; use taurus_core::runtime::engine::{ExecutionEngine, ExecutionId}; use taurus_core::time::now_unix_micros; use taurus_core::types::signal::Signal; @@ -13,115 +18,9 @@ use taurus_provider::providers::emitter::nats_emitter::NATSRespondEmitter; use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime; use tucana::shared::ExecutionFlow; use tucana::shared::NodeExecutionResult; -use tucana::shared::ValidationFlow; -use tucana::shared::Value; use tucana::shared::helper::value::from_json_value; use tucana::shared::helper::value::to_json_value; use tucana::shared::node_execution_result::Id as NodeExecutionResultId; -use tucana::shared::node_execution_result::Result as NodeExecutionResultResult; -use tucana::shared::value::Kind; - -#[derive(Clone, Deserialize)] -pub struct Input { - pub input: Option, - pub expected_result: serde_json::Value, -} - -#[derive(Clone, Deserialize)] -pub struct Case { - pub name: String, - pub description: String, - pub inputs: Vec, - pub flow: ValidationFlow, -} - -pub enum CaseResult { - Success, - Failure(Input, serde_json::Value), -} - -pub trait Testable { - fn run(&self) -> CaseResult; -} - -#[derive(Clone, Deserialize)] -pub struct Cases { - pub cases: Vec, -} - -pub fn print_success(case: &Case) { - info!("test {} ... ok", case.name); -} - -pub fn print_failure(case: &Case, input: &Input, result: serde_json::Value) { - error!("test {} ... FAILED", case.name); - error!(" input: {:?}", input.input); - error!(" expected: {:?}", input.expected_result); - error!(" real_value: {:?}", result); - error!(" message: {}", case.description); -} - -fn get_test_case + std::fmt::Debug>(path: P) -> Option { - let content = match std::fs::read_to_string(&path) { - Ok(it) => it, - Err(err) => { - log::error!("Cannot read file ({:?}): {:?}", path, err); - return None; - } - }; - - match serde_json::from_str(&content) { - Ok(it) => it, - Err(err) => { - log::error!("Cannot read json ({:?}): {:?}", path, err); - None - } - } -} - -fn get_test_cases(path: &str) -> Cases { - let mut items = Vec::new(); - let dir = match std::fs::read_dir(path) { - Ok(d) => d, - Err(err) => { - panic!("Cannot open path: {:?}", err) - } - }; - - for entry in dir { - let entry = match entry { - Ok(it) => it, - Err(err) => { - log::error!("Cannot read entry: {:?}", err); - continue; - } - }; - let file_path = entry.path(); - items.push(match get_test_case(&file_path) { - Some(it) => it, - None => { - continue; - } - }); - } - - Cases { cases: items } -} - -impl Case { - pub fn from_path(path: &str) -> Self { - match get_test_case(path) { - Some(s) => s, - None => panic!("flow was not found"), - } - } -} - -impl Cases { - pub fn from_path(path: &str) -> Self { - get_test_cases(path) - } -} #[derive(clap::Parser, Debug)] #[command(author, version, about)] @@ -209,7 +108,8 @@ async fn main() { let remote = NATSRemoteRuntime::with_execution_result_timeout(client.clone(), Duration::from_secs(30)); - let emitter = NATSRespondEmitter::new(client); + // A single-flow CLI tool never needs real backpressure headroom here. + let emitter = NATSRespondEmitter::new(client, 64); let engine = ExecutionEngine::new(); let started_at = now_unix_micros(); @@ -365,59 +265,6 @@ fn normalize_node_execution_results(node_results: &mut [NodeExecutionResult]) { } } -fn normalize_node_execution_result(result: &mut NodeExecutionResult) { - for parameter_result in &mut result.parameter_results { - match &mut parameter_result.value { - Some(value) => normalize_value(value), - None => { - parameter_result.value = Some(null_value()); - } - } - } - - match &mut result.result { - Some(NodeExecutionResultResult::Success(value)) => normalize_value(value), - Some(NodeExecutionResultResult::Error(error)) => { - if let Some(details) = &mut error.details { - for value in details.fields.values_mut() { - normalize_value(value); - } - } - } - None => { - result.result = Some(NodeExecutionResultResult::Success(null_value())); - } - } -} - -fn normalize_value(value: &mut Value) { - match &mut value.kind { - Some(Kind::StructValue(struct_value)) => { - for field in struct_value.fields.values_mut() { - normalize_value(field); - } - } - Some(Kind::ListValue(list_value)) => { - for item in &mut list_value.values { - normalize_value(item); - } - } - Some(Kind::NumberValue(number)) if number.number.is_none() => { - value.kind = Some(Kind::NullValue(0)); - } - Some(_) => {} - None => { - value.kind = Some(Kind::NullValue(0)); - } - } -} - -fn null_value() -> Value { - Value { - kind: Some(Kind::NullValue(0)), - } -} - fn execution_result_id_label(result: &NodeExecutionResult) -> String { match &result.id { Some(NodeExecutionResultId::NodeId(id)) => format!("node_id={}", id), diff --git a/crates/taurus-provider/src/lib.rs b/crates/taurus-provider/src/lib.rs index 9ce92fb..5f90b84 100644 --- a/crates/taurus-provider/src/lib.rs +++ b/crates/taurus-provider/src/lib.rs @@ -1 +1,7 @@ +//! NATS implementations of the `taurus_core::runtime::remote::RemoteRuntime` +//! and `taurus_core::runtime::engine::RespondEmitter` traits. `taurus-core` +//! has no dependency on this crate or on NATS at all; wiring these +//! implementations into an `ExecutionEngine` run is entirely up to the +//! binary crate (`taurus`, `taurus-manual`) that constructs them. + pub mod providers; diff --git a/crates/taurus-provider/src/providers/emitter/mod.rs b/crates/taurus-provider/src/providers/emitter/mod.rs index f3bb57e..d50c771 100644 --- a/crates/taurus-provider/src/providers/emitter/mod.rs +++ b/crates/taurus-provider/src/providers/emitter/mod.rs @@ -1 +1,3 @@ +//! NATS-backed `RespondEmitter` implementation. See [`nats_emitter`]. + pub mod nats_emitter; diff --git a/crates/taurus-provider/src/providers/emitter/nats_emitter.rs b/crates/taurus-provider/src/providers/emitter/nats_emitter.rs index 7b75b6f..ef6c2df 100644 --- a/crates/taurus-provider/src/providers/emitter/nats_emitter.rs +++ b/crates/taurus-provider/src/providers/emitter/nats_emitter.rs @@ -1,3 +1,7 @@ +//! Streams execution lifecycle events (`starting`/`ongoing`/`finished`/`failed`) +//! to `.` on NATS. See [`NATSRespondEmitter`] for +//! why publishing happens on a background task rather than inline in `emit`. + use async_nats::Client; use prost::Message; use std::collections::HashMap; @@ -8,8 +12,19 @@ use tucana::shared::{Struct, Value}; const DEFAULT_TOPIC_PREFIX: &str = "runtime.emitter"; +/// Keeps the synchronous `RespondEmitter::emit` API on the hot execution +/// path non-blocking by handing events to a background task over a bounded +/// channel; that task does the actual NATS publish. The channel is bounded +/// (rather than unbounded) because with concurrent flow execution, many +/// tasks can call `emit` in a burst while NATS publishing is comparatively +/// slow -- an unbounded queue would grow without limit if publishing ever +/// lags. Once full, `emit` drops the event rather than blocking the +/// caller; these are best-effort lifecycle notifications, not the +/// authoritative execution result (that goes back via gRPC separately). +/// Call [`NATSRespondEmitter::shutdown`] to drain queued events before +/// dropping this, since an aborted owner leaves them unpublished. pub struct NATSRespondEmitter { - tx: mpsc::UnboundedSender, + tx: mpsc::Sender, worker_task: tokio::task::JoinHandle<()>, } @@ -20,13 +35,17 @@ struct NATSEmitMessage { } impl NATSRespondEmitter { - pub fn new(client: Client) -> Self { - Self::with_topic_prefix(client, DEFAULT_TOPIC_PREFIX) + pub fn new(client: Client, channel_capacity: usize) -> Self { + Self::with_topic_prefix(client, DEFAULT_TOPIC_PREFIX, channel_capacity) } - pub fn with_topic_prefix(client: Client, topic_prefix: impl Into) -> Self { + pub fn with_topic_prefix( + client: Client, + topic_prefix: impl Into, + channel_capacity: usize, + ) -> Self { let topic_prefix = topic_prefix.into(); - let (tx, mut rx) = mpsc::unbounded_channel::(); + let (tx, mut rx) = mpsc::channel::(channel_capacity.max(1)); // Keep the public emitter API synchronous while publishing asynchronously. // This worker serializes outbound lifecycle events to one NATS topic per execution: @@ -74,15 +93,25 @@ impl NATSRespondEmitter { impl RespondEmitter for NATSRespondEmitter { fn emit(&self, execution_id: ExecutionId, emit_type: EmitType, value: Value) { - if let Err(err) = self.tx.send(NATSEmitMessage { + match self.tx.try_send(NATSEmitMessage { execution_id, emit_type, value, }) { - log::debug!( - "Dropped runtime emit message because NATS emitter worker is unavailable: {:?}", - err - ); + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + log::warn!( + "Dropped runtime emit message for execution {}: emitter channel full \ + (publishing is lagging behind emit rate)", + execution_id + ); + } + Err(mpsc::error::TrySendError::Closed(_)) => { + log::debug!( + "Dropped runtime emit message for execution {}: emitter worker is unavailable", + execution_id + ); + } } } } diff --git a/crates/taurus-provider/src/providers/mod.rs b/crates/taurus-provider/src/providers/mod.rs index 8501ddc..7f3cd97 100644 --- a/crates/taurus-provider/src/providers/mod.rs +++ b/crates/taurus-provider/src/providers/mod.rs @@ -1,5 +1,6 @@ -/// Provider for Remote Runtimes -pub mod remote; +//! The two NATS-backed provider implementations: [`remote`] for delegating +//! node execution to remote services, [`emitter`] for streaming execution +//! lifecycle events. -/// Provider for Remote Emitters pub mod emitter; +pub mod remote; diff --git a/crates/taurus-provider/src/providers/remote/mod.rs b/crates/taurus-provider/src/providers/remote/mod.rs index e862955..84c07cb 100644 --- a/crates/taurus-provider/src/providers/remote/mod.rs +++ b/crates/taurus-provider/src/providers/remote/mod.rs @@ -1 +1,3 @@ +//! NATS-backed `RemoteRuntime` implementation. See [`nats_remote_runtime`]. + pub mod nats_remote_runtime; 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 e68676d..9d471ab 100644 --- a/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs +++ b/crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs @@ -1,3 +1,8 @@ +//! 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`. + use std::time::Duration; use async_nats::Client; @@ -9,6 +14,9 @@ use tonic::async_trait; use tucana::aquila::ActionExecutionResponse; use tucana::shared::NodeExecutionResult; +// `Client` is a cheap Arc-backed handle, so this is cheap to clone per +// concurrently-executing flow. +#[derive(Clone)] pub struct NATSRemoteRuntime { client: Client, execution_result_timeout: Duration, diff --git a/crates/taurus-tests/Cargo.toml b/crates/taurus-tests/Cargo.toml index ddfa6c8..701814b 100644 --- a/crates/taurus-tests/Cargo.toml +++ b/crates/taurus-tests/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "tests" +name = "taurus-tests" version.workspace = true edition.workspace = true @@ -9,5 +9,4 @@ taurus-core = { workspace = true } log = { workspace = true } env_logger = { workspace = true } serde_json = { workspace = true } -serde = { workspace = true } async-trait = { workspace = true } diff --git a/crates/taurus-tests/src/main.rs b/crates/taurus-tests/src/main.rs index 375d6ff..d556b86 100644 --- a/crates/taurus-tests/src/main.rs +++ b/crates/taurus-tests/src/main.rs @@ -1,8 +1,10 @@ -use std::path::Path; +//! Batch fixture runner: loads every `ValidationFlow` fixture under `./flows/` +//! (see `taurus_core::fixtures`), runs each through `ExecutionEngine::execute_graph` +//! directly (no NATS involved), and logs a pass/fail line per case. Not wired +//! into `cargo test`; run explicitly via `cargo run --package taurus-tests`. -use log::{error, info}; -use serde::Deserialize; use serde_json::json; +use taurus_core::fixtures::{Case, Cases, Input, RemoteFixture, print_failure, print_success}; use taurus_core::runtime::engine::ExecutionEngine; use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime}; use taurus_core::types::errors::runtime_error::RuntimeError; @@ -10,34 +12,10 @@ use tucana::shared::node_execution_result::{ Id as NodeExecutionResultId, Result as NodeExecutionOutcome, }; use tucana::shared::{ - NodeExecutionResult, ValidationFlow, + NodeExecutionResult, helper::value::{from_json_value, to_json_value}, }; -#[derive(Clone, Deserialize)] -pub struct Input { - pub input: Option, - pub expected_result: serde_json::Value, -} - -#[derive(Clone, Deserialize)] -pub struct Case { - pub name: String, - pub description: String, - pub inputs: Vec, - pub flow: ValidationFlow, - #[serde(default)] - pub remote: Option, -} - -#[derive(Clone, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct RemoteFixture { - pub target_service: String, - pub function_identifier: String, - pub result_parameter: String, -} - struct FixtureRemoteRuntime { fixture: RemoteFixture, } @@ -107,85 +85,6 @@ pub trait Testable { fn run(&self) -> CaseResult; } -#[derive(Clone, Deserialize)] -pub struct Cases { - pub cases: Vec, -} - -pub fn print_success(case: &Case) { - info!("test {} ... ok", case.name); -} - -pub fn print_failure(case: &Case, input: &Input, result: serde_json::Value) { - error!("test {} ... FAILED", case.name); - error!(" input: {:?}", input.input); - error!(" expected: {:?}", input.expected_result); - error!(" real_value: {:?}", result); - error!(" message: {}", case.description); -} - -fn get_test_case + std::fmt::Debug>(path: P) -> Option { - let content = match std::fs::read_to_string(&path) { - Ok(it) => it, - Err(err) => { - log::error!("Cannot read file ({:?}): {:?}", path, err); - return None; - } - }; - - match serde_json::from_str(&content) { - Ok(it) => it, - Err(err) => { - log::error!("Cannot read json ({:?}): {:?}", path, err); - None - } - } -} - -fn get_test_cases(path: &str) -> Cases { - let mut items = Vec::new(); - let dir = match std::fs::read_dir(path) { - Ok(d) => d, - Err(err) => { - panic!("Cannot open path: {:?}", err) - } - }; - - for entry in dir { - let entry = match entry { - Ok(it) => it, - Err(err) => { - log::error!("Cannot read entry: {:?}", err); - continue; - } - }; - let file_path = entry.path(); - items.push(match get_test_case(&file_path) { - Some(it) => it, - None => { - continue; - } - }); - } - - Cases { cases: items } -} - -impl Case { - pub fn from_path(path: &str) -> Self { - match get_test_case(path) { - Some(s) => s, - None => panic!("flow was not found"), - } - } -} - -impl Cases { - pub fn from_path(path: &str) -> Self { - get_test_cases(path) - } -} - fn run_tests(cases: Cases) { for case in &cases.cases { match case.run() { diff --git a/crates/taurus/src/app/mod.rs b/crates/taurus/src/app/mod.rs index a339d4e..4e81e18 100644 --- a/crates/taurus/src/app/mod.rs +++ b/crates/taurus/src/app/mod.rs @@ -1,7 +1,16 @@ +//! Startup, shutdown, and (in dynamic mode) Aquila registration for the +//! Taurus service. [`worker`] is the actual NATS-driven flow execution loop +//! this module spawns and supervises; everything else here is one-time setup +//! or the shutdown sequence that tears it back down. +//! +//! Shutdown is cooperative rather than an abort: [`wait_for_shutdown`] signals +//! the worker task to stop via a `Notify` and waits for it to exit on its own, +//! so the worker gets a chance to drain its emitter (publish any queued +//! lifecycle events) before the process exits. + mod worker; use code0_flow::flow_config::environment::Environment; -use code0_flow::flow_config::load_env_file; use code0_flow::flow_config::mode::Mode::{DYNAMIC, STATIC}; use code0_flow::flow_definition::Reader; use code0_flow::flow_service::FlowUpdateService; @@ -11,6 +20,7 @@ use taurus_core::runtime::engine::ExecutionEngine; use taurus_provider::providers::emitter::nats_emitter::NATSRespondEmitter; use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime; use tokio::signal; +use tokio::sync::Notify; use tokio::task::JoinHandle; use tokio::time::sleep; use tonic_health::pb::health_server::HealthServer; @@ -22,8 +32,7 @@ use crate::config::Config; use crate::telemetry::{self, TelemetrySettings, errors}; pub async fn run() { - load_env_file(); - + // .env is loaded in main(), before the Tokio runtime is built. let config = Config::new(); let telemetry = init_telemetry(&config); install_panic_logging(); @@ -38,7 +47,8 @@ pub async fn run() { client.clone(), Duration::from_secs(config.remote_runtime_timeout_secs), ); - let runtime_emitter = NATSRespondEmitter::new(client.clone()); + let runtime_emitter = NATSRespondEmitter::new(client.clone(), config.emitter_channel_capacity); + let worker_shutdown = Arc::new(Notify::new()); let mut worker_task = worker::spawn_worker( client, engine, @@ -46,9 +56,11 @@ pub async fn run() { runtime_emitter, runtime_execution_service, mode_label(&config).to_string(), + worker_shutdown.clone(), + config.max_concurrent_executions, ); - wait_for_shutdown(&mut worker_task, &mut health_task).await; + wait_for_shutdown(&mut worker_task, &mut health_task, &worker_shutdown).await; if let Some(handle) = runtime_status_heartbeat_task.take() { handle.abort(); if let Err(err) = handle.await @@ -311,6 +323,7 @@ async fn update_stopped_status(runtime_status_service: Option<&Arc, health_task: &mut Option>, + worker_shutdown: &Arc, ) { #[cfg(unix)] let sigterm = async { @@ -323,24 +336,27 @@ async fn wait_for_shutdown( #[cfg(not(unix))] let sigterm = std::future::pending::<()>(); + let mut worker_already_finished = false; + if let Some(health_task) = health_task.as_mut() { tokio::select! { _ = &mut *worker_task => { log::warn!("NATS worker task finished, shutting down"); + worker_already_finished = true; health_task.abort(); } _ = &mut *health_task => { log::warn!("Health server task finished, shutting down"); - worker_task.abort(); + worker_shutdown.notify_one(); } _ = signal::ctrl_c() => { log::info!("Ctrl+C/Exit signal received, shutting down"); - worker_task.abort(); + worker_shutdown.notify_one(); health_task.abort(); } _ = sigterm => { log::info!("SIGTERM received, shutting down"); - worker_task.abort(); + worker_shutdown.notify_one(); health_task.abort(); } } @@ -348,17 +364,27 @@ async fn wait_for_shutdown( tokio::select! { _ = &mut *worker_task => { log::warn!("NATS worker task finished, shutting down"); + worker_already_finished = true; } _ = signal::ctrl_c() => { log::info!("Ctrl+C/Exit signal received, shutting down"); - worker_task.abort(); + worker_shutdown.notify_one(); } _ = sigterm => { log::info!("SIGTERM received, shutting down"); - worker_task.abort(); + worker_shutdown.notify_one(); } } } + + // Let the worker task drain its emitter (publish queued lifecycle events) + // instead of aborting it, so in-flight results aren't silently dropped. + if !worker_already_finished && let Err(err) = worker_task.await { + log::warn!( + "NATS worker task ended unexpectedly during shutdown: {}", + err + ); + } } #[cfg(test)] diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs index 23c337d..1a2b229 100644 --- a/crates/taurus/src/app/worker.rs +++ b/crates/taurus/src/app/worker.rs @@ -1,5 +1,27 @@ +//! The NATS-driven flow execution loop: subscribes to `execution.*`, decodes +//! each message into an `ExecutionFlow`, runs it through +//! `taurus_core::runtime::engine::ExecutionEngine`, and (in dynamic mode) +//! reports the result back to Aquila via [`TaurusRuntimeExecutionService`]. +//! +//! 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. +//! Concurrency is bounded by a semaphore (`Config::max_concurrent_executions`) +//! so a burst of NATS messages can't spawn unbounded in-flight executions. +//! 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. +//! +//! [`spawn_worker`] owns the `NATSRespondEmitter` (shared with per-message +//! tasks via `Arc`) for the lifetime of the task; it is only drained (via +//! `NATSRespondEmitter::shutdown`) once every in-flight execution has +//! finished and the loop exits, so callers must signal shutdown +//! cooperatively (see [`crate::app::wait_for_shutdown`]) rather than +//! aborting this task. + use futures_lite::StreamExt; use prost::Message; +use std::sync::Arc; use taurus_core::runtime::engine::{EmitType, ExecutionEngine, ExecutionId, RespondEmitter}; use taurus_core::runtime::remote::RemoteRuntime; use taurus_core::time::now_unix_micros; @@ -7,7 +29,8 @@ use taurus_core::types::errors::runtime_error::RuntimeError; use taurus_core::types::signal::Signal; use taurus_provider::providers::emitter::nats_emitter::NATSRespondEmitter; use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime; -use tokio::task::JoinHandle; +use tokio::sync::{Notify, Semaphore}; +use tokio::task::{JoinHandle, JoinSet}; use tucana::shared::execution_result; use tucana::shared::{ExecutionFlow, ExecutionResult, NodeExecutionResult, Value}; @@ -19,9 +42,15 @@ pub fn spawn_worker( engine: ExecutionEngine, nats_remote: NATSRemoteRuntime, runtime_emitter: NATSRespondEmitter, - mut runtime_execution_service: Option, + runtime_execution_service: Option, flow_type: String, + shutdown: Arc, + max_concurrent_executions: usize, ) -> JoinHandle<()> { + let engine = Arc::new(engine); + let runtime_emitter = Arc::new(runtime_emitter); + let semaphore = Arc::new(Semaphore::new(max_concurrent_executions.max(1))); + tokio::spawn(async move { let mut execution_subscription = match client .queue_subscribe(String::from("execution.*"), "taurus".into()) @@ -39,25 +68,39 @@ pub fn spawn_worker( &err, "subject=execution.* queue=taurus", ); + shutdown_emitter(runtime_emitter).await; return; } }; let mut execution_closed = false; + let mut in_flight = JoinSet::new(); while !execution_closed { tokio::select! { message = execution_subscription.next(), if !execution_closed => { match message { Some(message) => { - process_execution_message( - message, - &engine, - &nats_remote, - &runtime_emitter, - runtime_execution_service.as_mut(), - flow_type.as_str(), - ).await; + let Ok(permit) = semaphore.clone().acquire_owned().await else { + // Semaphore is never explicitly closed; unreachable in practice. + continue; + }; + let engine = engine.clone(); + let nats_remote = nats_remote.clone(); + let runtime_emitter = runtime_emitter.clone(); + let runtime_execution_service = runtime_execution_service.clone(); + let flow_type = flow_type.clone(); + in_flight.spawn(async move { + let _permit = permit; + process_execution_message( + message, + &engine, + &nats_remote, + &runtime_emitter, + runtime_execution_service, + flow_type.as_str(), + ).await; + }); } None => { execution_closed = true; @@ -65,19 +108,42 @@ pub fn spawn_worker( } } } + _ = shutdown.notified() => { + execution_closed = true; + log::info!("NATS worker received shutdown signal"); + } } } - log::info!("NATS worker loop ended"); + log::info!( + "NATS worker loop ended, draining {} in-flight execution(s)", + in_flight.len() + ); + while in_flight.join_next().await.is_some() {} + + shutdown_emitter(runtime_emitter).await; }) } +/// Drains the emitter's publish queue. Only ever called once every +/// in-flight execution (each holding its own `Arc` clone) has finished, so +/// this is always the sole remaining owner in practice; falls back to +/// leaving queued events unpublished (logged) rather than panicking if not. +async fn shutdown_emitter(runtime_emitter: Arc) { + match Arc::try_unwrap(runtime_emitter) { + Ok(emitter) => emitter.shutdown().await, + Err(_) => log::warn!( + "NATS emitter still has outstanding references at shutdown; skipping graceful drain" + ), + } +} + async fn process_execution_message( message: async_nats::Message, engine: &ExecutionEngine, nats_remote: &NATSRemoteRuntime, runtime_emitter: &NATSRespondEmitter, - mut runtime_execution_service: Option<&mut TaurusRuntimeExecutionService>, + mut runtime_execution_service: Option, flow_type: &str, ) { let requested_execution_id = parse_execution_id_from_subject(&message.subject, "execution") @@ -181,13 +247,16 @@ async fn execute_flow( let flow_id = flow.flow_id; let project_id = flow.project_id; let input = flow.input_value.clone(); + // Trace V2 collection is O(n^2) in executed nodes (see taurus-core's + // ValueStore::trace_snapshot) and nothing here consumes the trace_run + // engine.rs prints when with_trace is true; leave it off in production. let report = engine .execute_flow_with_execution_id_report_async( execution_id, flow, remote, respond_emitter, - true, + false, ) .await; let finished_at = now_unix_micros(); diff --git a/crates/taurus/src/client/mod.rs b/crates/taurus/src/client/mod.rs index a953406..44db6b7 100644 --- a/crates/taurus/src/client/mod.rs +++ b/crates/taurus/src/client/mod.rs @@ -1,2 +1,8 @@ +//! gRPC clients Taurus uses to talk *to* Aquila in dynamic mode: execution +//! results ([`runtime_execution`]) and runtime status/heartbeats +//! ([`runtime_status`]). Both are no-ops from the caller's perspective in +//! static mode, since [`crate::app`] only constructs them when +//! `MODE=dynamic`. + pub mod runtime_execution; pub mod runtime_status; diff --git a/crates/taurus/src/client/runtime_execution.rs b/crates/taurus/src/client/runtime_execution.rs index 6166605..d86d74e 100644 --- a/crates/taurus/src/client/runtime_execution.rs +++ b/crates/taurus/src/client/runtime_execution.rs @@ -1,18 +1,27 @@ +//! gRPC client that reports flow execution results back to Aquila. +//! +//! Results are normalized (via `taurus_core::normalize`) before being sent so +//! every optional proto field Aquila reads is explicitly populated rather +//! than left `None`. + use code0_flow::flow_service::{ auth::get_authorization_metadata, retry::create_channel_with_retry, }; use std::time::Duration; +use taurus_core::normalize::{normalize_node_execution_result, normalize_value, null_value}; use tonic::{Extensions, Request, transport::Channel}; use tucana::{ aquila::{ExecutionRequest, execution_service_client::ExecutionServiceClient}, - shared::{ - ExecutionResult, NodeExecutionResult, Value, execution_result, node_execution_result, - value::Kind, - }, + shared::{ExecutionResult, execution_result}, }; use crate::telemetry::errors; +// tonic clients over `Channel` are cheap to clone (the channel itself is a +// multiplexed, Arc-backed HTTP/2 connection); cloning per concurrently +// executing flow is the intended usage pattern rather than sharing one +// client behind a lock. +#[derive(Clone)] pub struct TaurusRuntimeExecutionService { client: ExecutionServiceClient, aquila_token: String, @@ -100,65 +109,12 @@ fn normalize_execution_result(result: &mut ExecutionResult) { } } -fn normalize_node_execution_result(result: &mut NodeExecutionResult) { - for parameter_result in &mut result.parameter_results { - match &mut parameter_result.value { - Some(value) => normalize_value(value), - None => { - parameter_result.value = Some(null_value()); - } - } - } - - match &mut result.result { - Some(node_execution_result::Result::Success(value)) => normalize_value(value), - Some(node_execution_result::Result::Error(error)) => { - if let Some(details) = &mut error.details { - for value in details.fields.values_mut() { - normalize_value(value); - } - } - } - None => { - result.result = Some(node_execution_result::Result::Success(null_value())); - } - } -} - -fn normalize_value(value: &mut Value) { - match &mut value.kind { - Some(Kind::StructValue(struct_value)) => { - for field in struct_value.fields.values_mut() { - normalize_value(field); - } - } - Some(Kind::ListValue(list_value)) => { - for item in &mut list_value.values { - normalize_value(item); - } - } - Some(Kind::NumberValue(number)) if number.number.is_none() => { - value.kind = Some(Kind::NullValue(0)); - } - Some(_) => {} - None => { - value.kind = Some(Kind::NullValue(0)); - } - } -} - -fn null_value() -> Value { - Value { - kind: Some(Kind::NullValue(0)), - } -} - #[cfg(test)] mod tests { use super::*; use tucana::shared::{ - Error, ListValue, NodeParameterNodeExecutionResult, NumberValue, Struct, - node_execution_result, + Error, ListValue, NodeExecutionResult, NodeParameterNodeExecutionResult, NumberValue, + Struct, Value, node_execution_result, value::Kind, }; #[test] diff --git a/crates/taurus/src/client/runtime_status.rs b/crates/taurus/src/client/runtime_status.rs index c9ee66d..6b017c3 100644 --- a/crates/taurus/src/client/runtime_status.rs +++ b/crates/taurus/src/client/runtime_status.rs @@ -1,3 +1,7 @@ +//! gRPC client that reports Taurus's own running/stopped status (and periodic +//! heartbeats) to Aquila. Aquila uses this to know which registered modules +//! currently have a live runtime backing them. + use std::time::{Duration, SystemTime, UNIX_EPOCH}; use code0_flow::flow_service::{ diff --git a/crates/taurus/src/config/mod.rs b/crates/taurus/src/config/mod.rs index a197a73..e029c3a 100644 --- a/crates/taurus/src/config/mod.rs +++ b/crates/taurus/src/config/mod.rs @@ -1,3 +1,7 @@ +//! Process-wide [`Config`], loaded once from environment variables (with an +//! optional `.env` file) at startup and read by every other module that +//! needs a URL, timeout, or mode flag. + use code0_flow::flow_config::env_with_default; use code0_flow::flow_config::environment::Environment; use code0_flow::flow_config::mode::Mode; @@ -42,6 +46,24 @@ pub struct Config { /// Timeout in seconds for remote runtime NATS flush and response waits. pub remote_runtime_timeout_secs: u64, + /// Maximum number of flow executions the worker runs concurrently. + /// Bounds memory/connection use under a burst of NATS messages; the + /// worker keeps pulling from the subscription but stops spawning new + /// executions once this many are in flight. Defaults to 4x the + /// available CPU parallelism (flows spend much of their time waiting + /// on remote calls, not on CPU, so oversubscribing cores is reasonable + /// as a starting point -- tune via env for your actual workload). + pub max_concurrent_executions: usize, + + /// Capacity of the NATS respond emitter's outbound event channel. + /// `emit` drops events (rather than blocking the caller) once this + /// many are queued waiting to be published -- these are best-effort + /// lifecycle notifications, not the authoritative execution result. + /// Defaults to 8x max_concurrent_executions, since a burst of + /// concurrent flows can each emit several events (starting/ongoing/ + /// finished) before the publish worker drains them. + pub emitter_channel_capacity: usize, + /// OpenTelemetry exporter configuration. pub opentelemetry: OpenTelemetry, } @@ -52,6 +74,11 @@ pub struct Config { /// Searches for the env. file at root level. Filename: `.env` impl Config { pub fn new() -> Self { + let max_concurrent_executions: usize = env_with_default( + "MAX_CONCURRENT_EXECUTIONS", + default_max_concurrent_executions(), + ); + Config { environment: env_with_default("ENVIRONMENT", Environment::Development), mode: env_with_default("MODE", Mode::DYNAMIC), @@ -75,6 +102,11 @@ impl Config { 10_u64, ), remote_runtime_timeout_secs: env_with_default("REMOTE_RUNTIME_TIMEOUT_SECS", 30_u64), + max_concurrent_executions, + emitter_channel_capacity: env_with_default( + "EMITTER_CHANNEL_CAPACITY", + max_concurrent_executions.saturating_mul(8), + ), opentelemetry: OpenTelemetry { enabled: env_with_default("OPENTELEMETRY_ENABLED", false), service_name: env_with_default( @@ -89,6 +121,13 @@ impl Config { } } +fn default_max_concurrent_executions() -> usize { + std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(4) + .saturating_mul(4) +} + fn optional_env(key: &str) -> Option { std::env::var(key) .ok() diff --git a/crates/taurus/src/main.rs b/crates/taurus/src/main.rs index f470245..cfd8f33 100644 --- a/crates/taurus/src/main.rs +++ b/crates/taurus/src/main.rs @@ -1,9 +1,31 @@ +//! Taurus is the CodeZero execution runtime: it consumes flow-execution +//! requests from NATS, runs them through `taurus_core::runtime::engine`, and +//! reports results back to Aquila over gRPC (dynamic mode) or purely via the +//! emitter (static mode). +//! +//! See [`app`] for the startup/shutdown sequence and the NATS worker loop, +//! [`client`] for the gRPC clients that talk back to Aquila, and [`config`] +//! for the environment-driven `Config` every other module reads from. + mod app; mod client; mod config; mod telemetry; -#[tokio::main] -async fn main() { - app::run().await; +fn main() { + // .env must be loaded, and MAX_BLOCKING_THREADS read, before the Tokio + // runtime is built below -- Tokio only accepts this setting at + // construction time, so it can't live in the `Config` that app::run() + // loads afterward. + code0_flow::flow_config::load_env_file(); + let max_blocking_threads: usize = + code0_flow::flow_config::env_with_default("MAX_BLOCKING_THREADS", 512_usize); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .max_blocking_threads(max_blocking_threads) + .build() + .unwrap_or_else(|error| panic!("failed to build Tokio runtime: {error}")); + + runtime.block_on(app::run()); } diff --git a/crates/taurus/src/telemetry/metrics.rs b/crates/taurus/src/telemetry/metrics.rs index bfec8f2..3e7ad52 100644 --- a/crates/taurus/src/telemetry/metrics.rs +++ b/crates/taurus/src/telemetry/metrics.rs @@ -1,3 +1,8 @@ +//! Flow and function execution metrics, recorded once per processed message +//! by [`crate::app::worker`]. `METRICS` is populated once at startup by +//! [`initialize`] (wired through `TelemetrySettings::initialize_metrics`) and +//! read thereafter without locking. + use std::sync::OnceLock; use opentelemetry::{ diff --git a/crates/taurus/src/telemetry/mod.rs b/crates/taurus/src/telemetry/mod.rs index 98f9dc3..a7790a2 100644 --- a/crates/taurus/src/telemetry/mod.rs +++ b/crates/taurus/src/telemetry/mod.rs @@ -1,3 +1,7 @@ +//! Observability wiring. `Telemetry`/`TelemetrySettings`/`errors` are +//! re-exported from `code0_flow`; [`metrics`] adds the flow/function +//! execution counters and histograms specific to Taurus. + pub mod metrics; pub use code0_flow::flow_telemetry::{OpenTelemetry, Telemetry, TelemetrySettings, errors};