From 93c4a854cfc14adc1cfcc10f30abd9154b3b0308 Mon Sep 17 00:00:00 2001 From: Raphael Date: Sun, 1 Nov 2026 11:34:20 +0100 Subject: [PATCH 1/4] deps: add apus-protocol local path dependency Wires apus-protocol (sibling repo ../apus/rust/apus-protocol) into the workspace as a path dependency, and adds it plus serde_json as runtime dependencies of the taurus crate (serde_json moves from dev-only to a real dependency: exec.rs::load_flows_file uses it at runtime). Adds rcgen and time as dev-dependencies for the Apus integration test suite's in-process TLS certificate generation. --- Cargo.lock | 158 +++++++++++++++++++++++++++++++++++++-- Cargo.toml | 3 + crates/taurus/Cargo.toml | 4 + 3 files changed, 158 insertions(+), 7 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a203064..a3d1043 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -117,6 +117,60 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "apus-protocol" +version = "0.1.0" +dependencies = [ + "base64 0.22.1", + "hmac", + "rustls", + "serde", + "serde_json", + "sha2", + "tokio", + "tokio-rustls", + "x509-parser", +] + +[[package]] +name = "asn1-rs" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f43a50ac4fdca5df8e885c21b835997f0a1cdee65494a6847694a98652d9d8" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom", + "num-traits", + "rusticata-macros", + "thiserror", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + +[[package]] +name = "asn1-rs-impl" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "async-nats" version = "0.50.0" @@ -705,6 +759,20 @@ dependencies = [ "zeroize", ] +[[package]] +name = "der-parser" +version = "10.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "deranged" version = "0.5.8" @@ -860,7 +928,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1768,6 +1836,12 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -1810,6 +1884,16 @@ dependencies = [ "signatory", ] +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -1929,6 +2013,15 @@ dependencies = [ "libm", ] +[[package]] +name = "oid-registry" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -2349,7 +2442,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.13.0", + "itertools 0.14.0", "log", "multimap", "petgraph", @@ -2370,7 +2463,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.13.0", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.119", @@ -2522,6 +2615,19 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "rcgen" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75e669e5202259b5314d1ea5397316ad400819437857b90861765f24c4cf80a2" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "yasna", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -2689,6 +2795,15 @@ dependencies = [ "semver", ] +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom", +] + [[package]] name = "rustix" version = "1.1.4" @@ -2699,7 +2814,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2757,7 +2872,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3108,6 +3223,7 @@ dependencies = [ name = "taurus" version = "0.1.0" dependencies = [ + "apus-protocol", "async-nats", "base64 0.23.1", "code0-flow", @@ -3118,10 +3234,12 @@ dependencies = [ "opentelemetry", "prost", "rand 0.10.3", + "rcgen", "serde", "serde_json", "taurus-core", "taurus-provider", + "time", "tokio", "tonic", "tonic-health", @@ -3225,7 +3343,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3932,7 +4050,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4094,6 +4212,32 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "x509-parser" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4569f339c0c402346d4a75a9e39cf8dad310e287eef1ff56d4c68e5067f53460" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom", + "oid-registry", + "rusticata-macros", + "thiserror", + "time", +] + +[[package]] +name = "yasna" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd" +dependencies = [ + "time", +] + [[package]] name = "yoke" version = "0.8.3" diff --git a/Cargo.toml b/Cargo.toml index 8c223b0..59ee9f1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,9 @@ path = "./crates/taurus-core" [workspace.dependencies.taurus-provider] path = "./crates/taurus-provider" +[workspace.dependencies.apus-protocol] +path = "../apus/rust/apus-protocol" + [workspace.dependencies.taurus-macros] path = "./crates/taurus-macros" diff --git a/crates/taurus/Cargo.toml b/crates/taurus/Cargo.toml index 7d99694..d485593 100644 --- a/crates/taurus/Cargo.toml +++ b/crates/taurus/Cargo.toml @@ -22,6 +22,10 @@ uuid = { workspace = true } env_logger = { workspace = true } jsonwebtoken = { workspace = true } serde = { workspace = true } +apus-protocol = { workspace = true } +serde_json = { workspace = true } [dev-dependencies] serde_json = { workspace = true } +rcgen = "0.13" +time = "0.3" From 6753db49fee21a3f60ae3c573749936b95e27990 Mon Sep 17 00:00:00 2001 From: Raphael Date: Sun, 1 Nov 2026 11:34:37 +0100 Subject: [PATCH 2/4] feat: port Apus server hub/convert/exec/remote Ports the Apus v1 data-plane server from the apus repo's reference implementation (apus/worktrees/taurus-083, branch apus, built on tucana-0.0.83-feature) into crates/taurus/src/app/apus/: - hub.rs: the Apus server (ApusHub, Connection, CallReply, ServerSettings, serve/serve_on). Mutual TLS (action identity from the client certificate's URI SAN) or plain TCP; handshake timeout; connection limits; ROUTES/STATUS; call slots and admission control; deadlines + CANCEL; bidirectional streams; heartbeats; GOAWAY; RETURN of a function becomes a tucana SubFlowFunctionResult. - convert.rs: hardened Apus <-> tucana value conversion, template forwarding, automatic streaming of large strings. - exec.rs: flow execution requested over Apus (ExecContext, ExecutionMode {Nats, Local}, FlowSource {Kv, Memory}), flow lookup from the NATS KV store (cached + watched), ownership checks, local execution via worker::execute_flow, NATS execution via request/reply, load_flows_file. - remote.rs: RoutingRemote implementing taurus_core's RemoteRuntime -- Apus when the action is directly connected, NATS fallback otherwise. - tests.rs: the integration test suite (raw-frame mock action driving serve_on) -- handshake, routing, overload, hostile input, streams both directions, sub-flows, returned functions, deadlines, CANCEL, heartbeats, client-certificate identity and expiry. No changes needed porting onto current origin/main: the reference implementation's mod.rs already excludes the earlier experimental "Glue" gRPC hub. --- crates/taurus/src/app/apus/convert.rs | 441 ++++++++ crates/taurus/src/app/apus/exec.rs | 312 ++++++ crates/taurus/src/app/apus/hub.rs | 1406 +++++++++++++++++++++++++ crates/taurus/src/app/apus/mod.rs | 14 + crates/taurus/src/app/apus/remote.rs | 134 +++ crates/taurus/src/app/apus/tests.rs | 941 +++++++++++++++++ 6 files changed, 3248 insertions(+) create mode 100644 crates/taurus/src/app/apus/convert.rs create mode 100644 crates/taurus/src/app/apus/exec.rs create mode 100644 crates/taurus/src/app/apus/hub.rs create mode 100644 crates/taurus/src/app/apus/mod.rs create mode 100644 crates/taurus/src/app/apus/remote.rs create mode 100644 crates/taurus/src/app/apus/tests.rs diff --git a/crates/taurus/src/app/apus/convert.rs b/crates/taurus/src/app/apus/convert.rs new file mode 100644 index 0000000..25d8baf --- /dev/null +++ b/crates/taurus/src/app/apus/convert.rs @@ -0,0 +1,441 @@ +//! Conversions between Taurus/Tucana values and the Apus wire encoding. +//! Encoding writes straight from `tucana::shared::Value` into the frame +//! buffer; decoding builds `Value`s straight from the frame bytes. + +use std::collections::HashMap; + +use apus_protocol::value::tag; +use apus_protocol::{ApusValue, DecodeError, ErrorInfo, Reader, StreamOut, Writer, limits}; +use tucana::aquila::{ActionNodeSubFlowValue, ActionNodeValue, action_node_value}; +use tucana::shared::number_value::Number; +use tucana::shared::value::Kind; +use tucana::shared::{ + Error as TucanaError, ListValue, NumberValue, Struct, SubFlowFunction, SubFlowFunctionResult, + Value, +}; + +/// Large strings go out as streams when `streams` is set (automatic streaming). +pub type Streams<'a, 'b, 'c> = &'a mut Option<&'b mut StreamOut<'c>>; + +pub fn write_value(w: &mut Writer, value: &Value, streams: Streams<'_, '_, '_>) { + match &value.kind { + None | Some(Kind::NullValue(_)) => w.null(), + Some(Kind::BoolValue(b)) => w.bool(*b), + Some(Kind::NumberValue(NumberValue { number })) => match number { + Some(Number::Integer(i)) => w.int(*i), + Some(Number::Float(f)) => w.float(*f), + None => w.null(), + }, + Some(Kind::StringValue(s)) => w.string_or_stream(s, streams.as_deref_mut()), + Some(Kind::ListValue(list)) => { + w.list_header(list.values.len()); + for item in &list.values { + write_value(w, item, streams); + } + } + Some(Kind::StructValue(st)) => write_struct(w, st, streams), + } +} + +fn write_struct(w: &mut Writer, st: &Struct, streams: Streams<'_, '_, '_>) { + w.object_header(st.fields.len()); + for (key, item) in &st.fields { + w.key(key); + write_value(w, item, streams); + } +} + +fn write_schema(w: &mut Writer, schema: &Option) { + match schema { + Some(st) => write_struct(w, st, &mut None), + None => w.null(), + } +} + +/// Writes one `CALL` parameter. A literal without references is a plain value; +/// a literal with inline `${…}` references is forwarded as a template (the +/// action resolves it, as with 0.0.82); a sub-flow becomes a handle. +pub fn write_param(w: &mut Writer, node: &ActionNodeValue, streams: Streams<'_, '_, '_>) { + match &node.value { + Some(action_node_value::Value::LiteralValue(literal)) => { + if literal.references.is_empty() { + match &literal.value { + Some(value) => write_value(w, value, streams), + None => w.null(), + } + return; + } + w.template_header(); + match &literal.value { + Some(value) => write_value(w, value, streams), + None => w.null(), + } + w.u16(literal.references.len() as u16); + for reference in &literal.references { + w.str(&reference.signature); + match &reference.value { + Some(value) => write_param(w, value, streams), + None => w.null(), + } + } + } + Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier, + input_schema, + output_schema, + })) => { + w.u8(tag::SUB_FLOW); + w.str(execution_identifier); + write_schema(w, input_schema); + write_schema(w, output_schema); + } + None => w.null(), + } +} + +/// Placeholder for a stream inside a decoded value, replaced by the stream's +/// content once it has been received (see [`fill_streams`]). +const STREAM_MARKER: &str = "\u{0}apus-stream:"; + +/// Reads a value sent by an action. Stream values become placeholders; their +/// `(id, length hint)` are pushed to `streams` so the caller can collect them +/// before using the value. Hardened like `apus_protocol::Reader::value`: depth +/// limit, counts checked against the remaining bytes, unique keys, valid UTF-8. +pub fn read_value(r: &mut Reader<'_>, streams: &mut Vec<(u32, i64)>) -> Result { + read_at(r, streams, 0) +} + +fn read_at( + r: &mut Reader<'_>, + streams: &mut Vec<(u32, i64)>, + depth: usize, +) -> Result { + if depth > limits::MAX_DEPTH { + return Err(DecodeError::TooDeep); + } + let kind = match r.u8()? { + tag::NULL => Kind::NullValue(0), + tag::FALSE => Kind::BoolValue(false), + tag::TRUE => Kind::BoolValue(true), + tag::INT => Kind::NumberValue(NumberValue { + number: Some(Number::Integer(r.i64()?)), + }), + tag::FLOAT => Kind::NumberValue(NumberValue { + number: Some(Number::Float(r.f64()?)), + }), + tag::STRING => { + let text = r.str()?; + // The placeholder prefix is reserved: a literal string must never be + // mistaken for (and replaced by) a stream's content. + if text.starts_with(STREAM_MARKER) { + return Err(DecodeError::Invalid("reserved string prefix")); + } + Kind::StringValue(text.to_owned()) + } + tag::LIST => { + let count = r.count(1)?; + let mut values = Vec::with_capacity(count.min(1024)); + for _ in 0..count { + values.push(read_at(r, streams, depth + 1)?); + } + Kind::ListValue(ListValue { values }) + } + tag::OBJECT => { + let count = r.count(5)?; + let mut fields = HashMap::with_capacity(count.min(1024)); + for _ in 0..count { + let key = r.str()?.to_owned(); + let value = read_at(r, streams, depth + 1)?; + if fields.insert(key, value).is_some() { + return Err(DecodeError::DuplicateKey); + } + } + Kind::StructValue(Struct { fields }) + } + tag::STREAM => { + let id = r.u32()?; + let _content_type = r.str()?; + let length = r.i64()?; + streams.push((id, length)); + Kind::StringValue(format!("{STREAM_MARKER}{id}")) + } + other => return Err(DecodeError::InvalidTag(other)), + }; + Ok(Value { kind: Some(kind) }) +} + +/// Reads a function an action returned as the whole value of a `RETURN` +/// (tucana `SubFlowFunctionResult`); the `FUNCTION` tag is already consumed. +/// Schemas are objects or null and may not contain streams. +pub fn read_function(r: &mut Reader<'_>) -> Result { + let function_identifier = r.str()?.to_owned(); + let definition_source = r.str()?.to_owned(); + let mut streams = Vec::new(); + let input_schema = schema(read_at(r, &mut streams, 1)?)?; + let output_schema = schema(read_at(r, &mut streams, 1)?)?; + if !streams.is_empty() { + return Err(DecodeError::Invalid("stream in a function schema")); + } + Ok(SubFlowFunctionResult { + function: Some(SubFlowFunction { + function_identifier, + definition_source: (!definition_source.is_empty()).then_some(definition_source), + }), + input_schema, + output_schema, + name: r.str()?.to_owned(), + description: r.str()?.to_owned(), + }) +} + +fn schema(value: Value) -> Result, DecodeError> { + match value.kind { + None | Some(Kind::NullValue(_)) => Ok(None), + Some(Kind::StructValue(schema)) => Ok(Some(schema)), + Some(_) => Err(DecodeError::Invalid( + "function schema must be an object or null", + )), + } +} + +/// Replaces stream placeholders with the received bytes. Flow values have no +/// bytes type, so a stream stands for a string and must be valid UTF-8. +pub fn fill_streams(value: &mut Value, contents: &mut HashMap>) -> Result<(), String> { + match &mut value.kind { + Some(Kind::StringValue(s)) => { + if let Some(id) = s + .strip_prefix(STREAM_MARKER) + .and_then(|id| id.parse::().ok()) + { + let bytes = contents + .remove(&id) + .ok_or_else(|| format!("stream {id} is not part of this value"))?; + *s = String::from_utf8(bytes) + .map_err(|_| format!("stream {id} is not valid UTF-8 text"))?; + } + Ok(()) + } + Some(Kind::ListValue(list)) => list + .values + .iter_mut() + .try_for_each(|v| fill_streams(v, contents)), + Some(Kind::StructValue(st)) => st + .fields + .values_mut() + .try_for_each(|v| fill_streams(v, contents)), + _ => Ok(()), + } +} + +pub fn from_apus(value: &ApusValue) -> Value { + let kind = match value { + ApusValue::Null => Kind::NullValue(0), + ApusValue::Bool(b) => Kind::BoolValue(*b), + ApusValue::Int(i) => Kind::NumberValue(NumberValue { + number: Some(Number::Integer(*i)), + }), + ApusValue::Float(f) => Kind::NumberValue(NumberValue { + number: Some(Number::Float(*f)), + }), + ApusValue::String(s) => Kind::StringValue(s.clone()), + ApusValue::List(items) => Kind::ListValue(ListValue { + values: items.iter().map(from_apus).collect(), + }), + ApusValue::Object(fields) => Kind::StructValue(Struct { + fields: fields + .iter() + .map(|(k, v)| (k.clone(), from_apus(v))) + .collect(), + }), + // Params, streams and returned functions never reach the engine as such. + ApusValue::Template { .. } + | ApusValue::SubFlow { .. } + | ApusValue::Stream { .. } + | ApusValue::Function { .. } => Kind::NullValue(0), + }; + Value { kind: Some(kind) } +} + +/// Apus error (`ERROR`, `EXEC_ERROR`, `SUBFLOW_ERROR`) -> `shared.Error`. +pub fn to_tucana_error(error: ErrorInfo) -> TucanaError { + TucanaError { + code: error.code, + category: error.category, + message: error.message, + timestamp: error.timestamp_ms, + version: error.version, + dependencies: error.dependencies.into_iter().collect(), + details: match &error.details { + ApusValue::Object(_) => match from_apus(&error.details).kind { + Some(Kind::StructValue(st)) => Some(st), + _ => None, + }, + _ => None, + }, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tucana::aquila::{ActionInlineReferenceValue, ActionLiteralValue}; + + fn string(s: &str) -> Value { + Value { + kind: Some(Kind::StringValue(s.into())), + } + } + + fn int(i: i64) -> Value { + Value { + kind: Some(Kind::NumberValue(NumberValue { + number: Some(Number::Integer(i)), + })), + } + } + + #[test] + fn value_round_trips() { + let value = Value { + kind: Some(Kind::StructValue(Struct { + fields: HashMap::from([ + ("a".to_string(), int(7)), + ( + "b".to_string(), + Value { + kind: Some(Kind::ListValue(ListValue { + values: vec![ + string("x"), + Value { + kind: Some(Kind::BoolValue(true)), + }, + ], + })), + }, + ), + ]), + })), + }; + let mut w = Writer::default(); + write_value(&mut w, &value, &mut None); + let bytes = w.into_inner(); + let decoded = read_value(&mut Reader::new(&bytes), &mut Vec::new()).unwrap(); + assert_eq!(decoded, value); + } + + #[test] + fn templates_and_sub_flows_are_forwarded_as_params() { + let template = ActionNodeValue { + value: Some(action_node_value::Value::LiteralValue(ActionLiteralValue { + value: Some(string("id-${n}")), + references: vec![ActionInlineReferenceValue { + signature: "n".into(), + value: Some(ActionNodeValue { + value: Some(action_node_value::Value::LiteralValue(ActionLiteralValue { + value: Some(int(5)), + references: vec![], + })), + }), + }], + })), + }; + let sub_flow = ActionNodeValue { + value: Some(action_node_value::Value::SubFlow(ActionNodeSubFlowValue { + execution_identifier: "sf-1".into(), + input_schema: None, + output_schema: None, + })), + }; + let mut w = Writer::default(); + write_param(&mut w, &template, &mut None); + write_param(&mut w, &sub_flow, &mut None); + let bytes = w.into_inner(); + let mut r = Reader::new(&bytes); + let template = r.value().unwrap(); + assert_eq!( + apus_protocol::template::resolve(template).unwrap(), + ApusValue::string("id-5") + ); + assert_eq!( + r.value().unwrap(), + ApusValue::SubFlow { + id: "sf-1".into(), + input_schema: Box::new(ApusValue::Null), + output_schema: Box::new(ApusValue::Null), + } + ); + } + + #[test] + fn hostile_values_are_rejected() { + let mut deep = Vec::new(); + for _ in 0..1000 { + deep.push(tag::LIST); + deep.extend_from_slice(&1u32.to_le_bytes()); + } + deep.push(tag::NULL); + assert_eq!( + read_value(&mut Reader::new(&deep), &mut Vec::new()), + Err(DecodeError::TooDeep) + ); + + let mut huge = vec![tag::OBJECT]; + huge.extend_from_slice(&u32::MAX.to_le_bytes()); + assert_eq!( + read_value(&mut Reader::new(&huge), &mut Vec::new()), + Err(DecodeError::CountTooLarge) + ); + + let mut w = Writer::default(); + w.object_header(2); + w.key("a"); + w.null(); + w.key("a"); + w.int(1); + assert_eq!( + read_value(&mut Reader::new(&w.into_inner()), &mut Vec::new()), + Err(DecodeError::DuplicateKey) + ); + } + + #[test] + fn literal_placeholders_and_invalid_text_are_rejected() { + let mut w = Writer::default(); + w.string(&format!("{STREAM_MARKER}3")); + assert!(read_value(&mut Reader::new(&w.into_inner()), &mut Vec::new()).is_err()); + + let mut w = Writer::default(); + w.stream(3, "text/plain", 2); + let mut value = read_value(&mut Reader::new(&w.into_inner()), &mut Vec::new()).unwrap(); + assert!(fill_streams(&mut value, &mut HashMap::from([(3, vec![0xc3, 0x28])])).is_err()); + } + + #[test] + fn large_strings_are_streamed() { + let next = std::sync::atomic::AtomicU32::new(2); + let mut out = StreamOut::new(4, &next); + let mut w = Writer::default(); + write_value(&mut w, &string("a long one"), &mut Some(&mut out)); + assert_eq!(out.pending, vec![(2, b"a long one".to_vec())]); + let mut streams = Vec::new(); + read_value(&mut Reader::new(&w.into_inner()), &mut streams).unwrap(); + assert_eq!(streams, vec![(2, 10)]); + } + + #[test] + fn streams_become_placeholders_until_filled() { + let mut w = Writer::default(); + w.list_header(2); + w.int(1); + w.stream(3, "text/plain", 5); + let bytes = w.into_inner(); + let mut streams = Vec::new(); + let mut value = read_value(&mut Reader::new(&bytes), &mut streams).unwrap(); + assert_eq!(streams, vec![(3, 5)]); + fill_streams(&mut value, &mut HashMap::from([(3, b"hello".to_vec())])).unwrap(); + let Some(Kind::ListValue(list)) = value.kind else { + panic!("expected list") + }; + assert_eq!(list.values, vec![int(1), string("hello")]); + } +} diff --git a/crates/taurus/src/app/apus/exec.rs b/crates/taurus/src/app/apus/exec.rs new file mode 100644 index 0000000..4804220 --- /dev/null +++ b/crates/taurus/src/app/apus/exec.rs @@ -0,0 +1,312 @@ +//! Flow executions requested by a directly connected action (`EXEC` frames). +//! Replaces Aquila's `handle_flow_execution`: look up the flow (Aquila keeps +//! them in its flow store; Taurus caches and watches them), build the +//! `ExecutionFlow`, then run it in this process (`local`) or put it on the NATS +//! execution queue (`nats`). + +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use apus_protocol::ErrorInfo; +use futures_lite::StreamExt; +use prost::Message; +use taurus_core::runtime::engine::{ExecutionEngine, ExecutionId}; +use taurus_core::types::signal::Signal; +use tucana::shared::{ExecutionFlow, ExecutionResult, ValidationFlow, Value, execution_result}; + +use super::remote::RoutingRemote; +use crate::app::worker; +use crate::client::runtime_usage::RuntimeUsageRecorder; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ExecutionMode { + /// Publish on `execution.` with a reply inbox; any Taurus worker runs it. + Nats, + /// Run the flow directly in this process (no NATS hop). + Local, +} + +pub struct ExecFailure { + pub code: String, + pub category: String, + pub message: String, +} + +impl ExecFailure { + pub fn new(code: &str, category: &str, message: impl Into) -> Self { + Self { + code: code.to_string(), + category: category.to_string(), + message: message.into(), + } + } + + pub fn into_error(self) -> ErrorInfo { + ErrorInfo { + timestamp_ms: taurus_core::time::now_unix_micros() / 1000, + version: "taurus".to_string(), + ..ErrorInfo::new(self.code, self.category, self.message) + } + } +} + +/// Where Taurus gets flows from when an action asks to execute one. +pub enum FlowSource { + /// NATS KV `flow_store` (Aquila writes it), optionally cached; the cache + /// follows the store through a KV watch (see [`ExecContext::watch_flows`]). + Kv { + kv: async_nats::jetstream::kv::Store, + cache: Option>>>>, + }, + /// Held in memory (no NATS). Production: pushed by Aquila over the control plane. + Memory(HashMap>), +} + +pub struct ExecContext { + pub client: Option, + pub flows: FlowSource, + pub mode: ExecutionMode, + pub engine: Arc, + pub remote: RoutingRemote, + pub timeout: Duration, + pub flow_type: String, + pub with_trace: bool, + /// Runtime usage of local executions (dynamic mode), like the NATS worker + /// records it; taken at shutdown so the usage publisher can flush and exit. + pub runtime_usage: Mutex>, +} + +impl ExecContext { + pub fn stop_runtime_usage(&self) { + self.runtime_usage + .lock() + .unwrap_or_else(|e| e.into_inner()) + .take(); + } + + pub async fn execute( + &self, + action: &str, + flow_id: i64, + project_id: i64, + execution_id: &str, + input: Value, + projects: Option<&HashSet>, + ) -> Result { + let flow = self.load_flow(flow_id).await.ok_or_else(|| { + ExecFailure::new( + "T-APUS-FLOW-000001", + "NOT_FOUND", + format!("flow {flow_id} was not found"), + ) + })?; + if flow.project_id != project_id { + // The action routed by the project it claimed; it must be the flow's. + return Err(ExecFailure::new( + "T-APUS-FLOW-000004", + "INVALID_ARGUMENT", + format!("flow {flow_id} does not belong to project {project_id}"), + )); + } + if flow.definition_source.as_deref() != Some(format!("action.{action}").as_str()) { + return Err(ExecFailure::new( + "T-APUS-FLOW-000002", + "PERMISSION_DENIED", + format!("flow {flow_id} does not belong to action {action}"), + )); + } + if let Some(projects) = projects + && !projects.contains(&flow.project_id) + { + return Err(ExecFailure::new( + "APUS-FORBIDDEN", + "PERMISSION_DENIED", + format!("the ticket does not cover project {}", flow.project_id), + )); + } + if let Some(reason) = &flow.disable_reason { + return Err(ExecFailure::new( + "T-APUS-FLOW-000003", + "FAILED_PRECONDITION", + format!("flow {flow_id} has been disabled: {reason}"), + )); + } + + let execution_flow = ExecutionFlow { + flow_id, + project_id: flow.project_id, + starting_node_id: flow.starting_node_id, + node_functions: flow.node_functions.clone(), + input_value: Some(input), + }; + + match self.mode { + ExecutionMode::Nats => self.execute_via_nats(execution_id, execution_flow).await, + ExecutionMode::Local => Ok(self.execute_locally(execution_id, execution_flow).await?), + } + } + + async fn execute_via_nats( + &self, + execution_id: &str, + flow: ExecutionFlow, + ) -> Result { + let client = self.client.as_ref().ok_or_else(|| { + ExecFailure::new("T-APUS-EXEC-000003", "UNAVAILABLE", "NATS is disabled") + })?; + let request = async_nats::Request::new() + .payload(flow.encode_to_vec().into()) + .timeout(Some(self.timeout)); + let reply = client + .send_request(format!("execution.{execution_id}"), request) + .await + .map_err(|err| { + ExecFailure::new( + "T-APUS-EXEC-000001", + "UNAVAILABLE", + format!("execution request failed: {err}"), + ) + })?; + let result = ExecutionResult::decode(reply.payload).map_err(|err| { + ExecFailure::new( + "T-APUS-EXEC-000002", + "INTERNAL", + format!("invalid execution result: {err}"), + ) + })?; + match result.result { + Some(execution_result::Result::Success(value)) => Ok(value), + Some(execution_result::Result::Error(err)) => Err(ExecFailure { + code: err.code, + category: err.category, + message: err.message, + }), + None => Ok(null()), + } + } + + async fn execute_locally( + &self, + execution_id: &str, + flow: ExecutionFlow, + ) -> Result { + let execution_id = + ExecutionId::parse_str(execution_id).unwrap_or_else(|_| ExecutionId::new_v4()); + let function_identifiers = worker::function_identifiers_by_node_id(&flow); + let run = worker::execute_flow( + execution_id, + flow, + &self.engine, + Some(&self.remote), + &self.flow_type, + function_identifiers, + self.with_trace, + ) + .await; + if let Some(runtime_usage) = self + .runtime_usage + .lock() + .unwrap_or_else(|e| e.into_inner()) + .as_ref() + { + runtime_usage.record(run.flow_id, run.finished_at.saturating_sub(run.started_at)); + } + match run.signal { + Signal::Success(value) | Signal::Return(value) => Ok(value), + Signal::Stop => Ok(null()), + Signal::Failure(err) => Err(ExecFailure { + code: err.code, + category: err.category, + message: err.message, + }), + } + } + + pub(crate) async fn load_flow(&self, flow_id: i64) -> Option> { + let (kv, cache) = match &self.flows { + FlowSource::Memory(flows) => return flows.get(&flow_id).cloned(), + FlowSource::Kv { kv, cache } => (kv, cache), + }; + if let Some(cache) = cache + && let Some(flow) = cache + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(&flow_id) + { + return Some(flow.clone()); + } + let bytes = kv.get(flow_id.to_string()).await.ok()??; + let flow = Arc::new(ValidationFlow::decode(bytes).ok()?); + if let Some(cache) = cache { + cache + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert(flow_id, flow.clone()); + } + Some(flow) + } + + /// Keeps the flow cache in sync with Aquila's flow store: updated flows are + /// replaced, deleted ones removed (no stale flows after an edit). + pub fn watch_flows(&self) { + let FlowSource::Kv { + kv, + cache: Some(cache), + } = &self.flows + else { + return; + }; + let (kv, cache) = (kv.clone(), cache.clone()); + tokio::spawn(async move { + let mut watch = match kv.watch_all().await { + Ok(watch) => watch, + Err(err) => { + log::warn!("Apus: cannot watch the flow store: {}", err); + return; + } + }; + while let Some(entry) = watch.next().await { + let Ok(entry) = entry else { continue }; + let Ok(flow_id) = entry.key.parse::() else { + continue; + }; + let mut cache = cache.lock().unwrap_or_else(|e| e.into_inner()); + match entry.operation { + async_nats::jetstream::kv::Operation::Put => { + match ValidationFlow::decode(entry.value) { + Ok(flow) => { + cache.insert(flow_id, Arc::new(flow)); + } + Err(_) => { + cache.remove(&flow_id); + } + } + } + _ => { + cache.remove(&flow_id); + } + } + } + }); + } +} + +/// Loads a JSON flow export (`tucana::shared::Flows`, same format as Aquila's static mode). +pub fn load_flows_file(path: &str) -> HashMap> { + let data = std::fs::read_to_string(path) + .unwrap_or_else(|err| panic!("Apus: cannot read flows file {path}: {err}")); + let flows: tucana::shared::Flows = serde_json::from_str(&data) + .unwrap_or_else(|err| panic!("Apus: cannot parse flows file {path}: {err}")); + flows + .flows + .into_iter() + .map(|flow| (flow.flow_id, Arc::new(flow))) + .collect() +} + +fn null() -> Value { + Value { + kind: Some(tucana::shared::value::Kind::NullValue(0)), + } +} diff --git a/crates/taurus/src/app/apus/hub.rs b/crates/taurus/src/app/apus/hub.rs new file mode 100644 index 0000000..ec40af6 --- /dev/null +++ b/crates/taurus/src/app/apus/hub.rs @@ -0,0 +1,1406 @@ +//! The Apus v1 server (data plane): authenticates direct action connections, +//! multiplexes calls over them and serves the executions, sub-flow invocations +//! and streams the action starts. Routing (which Taurus instance serves which +//! project) is announced with `ROUTES`; configuration and flows stay in Aquila. + +use std::collections::{HashMap, HashSet}; +use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::{Duration, Instant}; + +use apus_protocol::message::{self, Route, Routes, reset, status}; +use apus_protocol::stream::{DataOutcome, Incoming, Outgoing, Side, StreamResult}; +use apus_protocol::{ + ErrorInfo, FrameBuilder, Hello, PROTOCOL_VERSION, Reader, StreamOut, TraceContext, Welcome, + cancel, feature, kind, limits, read_frame_limited, reject, scaling, tls, +}; +use taurus_core::types::errors::runtime_error::RuntimeError; +use taurus_core::types::signal::Signal; +use taurus_core::value::raw_value_result; +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, BufReader, BufWriter}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::{Notify, Semaphore, mpsc, oneshot}; +use tucana::aquila::ActionExecutionRequest; +use tucana::shared::{NodeValueExecutionResult, Value, node_value_execution_result}; + +use super::convert; +use super::exec::{ExecContext, ExecFailure}; + +type BoxRead = Box; +type BoxWrite = Box; +type StreamReceivers = Vec<(u32, oneshot::Receiver)>; + +/// Largest `HELLO` accepted. +const MAX_HELLO: usize = 64 * 1024; +/// Delay before answering a failed `HELLO` (slows down guessing). +const REJECT_DELAY: Duration = Duration::from_millis(50); +/// How long an action may finish in-flight work after its client certificate expired. +const CERTIFICATE_GRACE: Duration = Duration::from_secs(30); + +/// Reply to a `CALL`: the action's return value or its reported error, with the +/// action's own timestamps (as in `NodeExecutionResult`). +pub enum CallReply { + Ok { + /// A plain value, or a function the action returned (tucana 0.0.83). + value: NodeValueExecutionResult, + started_at: i64, + finished_at: i64, + }, + Err { + error: ErrorInfo, + started_at: i64, + finished_at: i64, + }, +} + +/// An action-initiated request in flight: abort handle for `CANCEL`, and the +/// deadline the sweeper enforces (no timer per request). +struct RunningRequest { + abort: tokio::task::AbortHandle, + deadline: Option, + error_kind: u8, +} + +/// Map split into independently locked shards: the reader task inserts, the +/// request tasks remove from other worker threads. +struct ShardedMap { + shards: Vec>>, +} + +impl ShardedMap { + fn new(shards: usize) -> Self { + Self { + shards: (0..shards).map(|_| Mutex::new(HashMap::new())).collect(), + } + } + + fn shard(&self, key: u32) -> std::sync::MutexGuard<'_, HashMap> { + self.shards[key as usize % self.shards.len()] + .lock() + .unwrap_or_else(|e| e.into_inner()) + } + + fn contains(&self, key: u32) -> bool { + self.shard(key).contains_key(&key) + } + + fn remove(&self, key: u32) -> Option { + self.shard(key).remove(&key) + } + + /// Removes and returns every entry for which `expired` is true. + fn drain_where(&self, mut expired: impl FnMut(&V) -> bool) -> Vec<(u32, V)> { + let mut out = Vec::new(); + for shard in &self.shards { + let mut map = shard.lock().unwrap_or_else(|e| e.into_inner()); + let keys: Vec = map + .iter() + .filter(|(_, v)| expired(v)) + .map(|(k, _)| *k) + .collect(); + for key in keys { + if let Some(v) = map.remove(&key) { + out.push((key, v)); + } + } + } + out + } +} + +/// One live action connection. +pub struct Connection { + pub action: String, + functions: HashMap, + tx: mpsc::UnboundedSender>, + pending: Mutex>>, + next_id: AtomicU32, + /// `HELLO.maxInFlight`: CALLs this action accepts concurrently. + call_slots: Option>, + /// Projects of the client certificate; `None` = every project of the action. + projects: Option>, + /// (index, count) for `split`-scaled connections. + shard: Option<(u32, u32)>, + /// The action's `STATUS`; CALLs are only routed while it is running. + status: AtomicU8, + draining: AtomicBool, + incoming: Mutex, + outgoing: Arc, + next_stream_id: AtomicU32, + stream_threshold: usize, + /// EXEC + SUBFLOW requests currently running for this connection. + requests_in_flight: AtomicU32, + max_in_flight: u32, + /// `STATUS not ready` was sent because the connection hit `max_in_flight`. + overloaded: AtomicBool, + request_tasks: ShardedMap, + rx_frames: AtomicU64, +} + +impl Connection { + /// Whether this connection may serve `project_id` (ticket and shard). + fn serves(&self, project_id: i64) -> bool { + if let Some(projects) = &self.projects + && !projects.contains(&project_id) + { + return false; + } + match self.shard { + Some((index, count)) => project_id.rem_euclid(count as i64) == index as i64, + None => true, + } + } + + fn routable(&self) -> bool { + self.status.load(Ordering::Relaxed) == status::RUNNING + && !self.draining.load(Ordering::Relaxed) + } + + /// Sends a `CALL` and waits for `RETURN` / `ERROR`. The wait is renewed + /// whenever `activity` fires (the action is driving a sub-flow of this call). + pub async fn call( + &self, + method: u16, + request: &ActionExecutionRequest, + timeout: Duration, + activity: Option<&Arc>, + trace: Option<&TraceContext>, + ) -> Result { + // Load protection: never more outstanding calls than the action accepts. + let _permit = match &self.call_slots { + Some(slots) => match slots.clone().try_acquire_owned() { + Ok(permit) => Some(permit), + Err(_) => { + match tokio::time::timeout(timeout, slots.clone().acquire_owned()).await { + Ok(Ok(permit)) => Some(permit), + _ => { + return Err(RuntimeError::new( + "T-APUS-000004", + "ApusActionOverloaded", + "action has no free call slot before the deadline", + )); + } + } + } + }, + None => None, + }; + + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let deadline_ms = timeout.as_millis().min(u32::MAX as u128) as u32; + let mut frame = FrameBuilder::traced(kind::CALL, id, 256, trace); + let mut streams = StreamOut::new(self.stream_threshold, &self.next_stream_id); + { + let body = frame.body(); + body.u16(method); + body.str(&request.execution_identifier); + body.i64(request.project_id); + body.u32(deadline_ms); + body.u8(request.parameters.len() as u8); + let mut out = Some(&mut streams); + for parameter in &request.parameters { + convert::write_param(body, parameter, &mut out); + } + } + + let (reply_tx, reply_rx) = oneshot::channel(); + self.lock_pending().insert(id, reply_tx); + // Removes the entry however this call ends, also when the execution + // around it is aborted (CANCEL): no pending entry outlives its call. + struct Pending<'a>(&'a Connection, u32, bool); + impl Drop for Pending<'_> { + fn drop(&mut self) { + if self.0.lock_pending().remove(&self.1).is_some() && self.2 { + let _ = self + .0 + .tx + .send(message::encode_cancel(self.1, cancel::CALLER)); + } + } + } + let mut pending = Pending(self, id, true); + if self.tx.send(frame.finish()).is_err() { + pending.2 = false; + return Err(transport_error("connection to action is closed")); + } + self.send_streams(streams.pending); + + match wait_for_reply(reply_rx, timeout, activity).await { + Some(Ok(reply)) => Ok(reply), + Some(Err(_)) => { + pending.2 = false; + Err(transport_error("connection to action closed during call")) + } + None => { + pending.2 = false; + self.lock_pending().remove(&id); + let _ = self.tx.send(message::encode_cancel(id, cancel::DEADLINE)); + Err(transport_error("action did not answer before the deadline")) + } + } + } + + fn send_streams(&self, pending: Vec<(u32, Vec)>) { + for (id, bytes) in pending { + self.outgoing.send(self.tx.clone(), id, bytes); + } + } + + fn lock_pending(&self) -> std::sync::MutexGuard<'_, HashMap>> { + self.pending.lock().unwrap_or_else(|err| err.into_inner()) + } + + fn lock_incoming(&self) -> std::sync::MutexGuard<'_, Incoming> { + self.incoming.lock().unwrap_or_else(|err| err.into_inner()) + } + + fn complete(&self, id: u32, reply: CallReply) { + if let Some(tx) = self.lock_pending().remove(&id) { + let _ = tx.send(reply); + } + } + + /// Registers the streams of a received value (in frame order, before their data). + /// On refusal every stream of the value is reset and the error returned. + fn expect_streams(&self, refs: &[(u32, i64)]) -> Result { + let mut incoming = self.lock_incoming(); + let mut receivers = Vec::with_capacity(refs.len()); + for (id, length) in refs { + match incoming.expect(*id, *length) { + Ok(rx) => receivers.push((*id, rx)), + Err(code) => { + for (registered, _) in &receivers { + incoming.fail(*registered, reset::CANCELLED); + } + drop(incoming); + for (other, _) in refs { + let _ = self.tx.send(message::encode_reset(*other, code)); + } + return Err(format!("stream {id} refused (code {code})")); + } + } + } + Ok(receivers) + } + + fn fail_all(&self) { + self.lock_pending().clear(); + self.lock_incoming().clear(); + self.outgoing.clear(); + for (_, request) in self.request_tasks.drain_where(|_| true) { + request.abort.abort(); + } + } +} + +async fn wait_for_reply( + reply_rx: oneshot::Receiver, + timeout: Duration, + activity: Option<&Arc>, +) -> Option> { + let Some(activity) = activity else { + return tokio::time::timeout(timeout, reply_rx).await.ok(); + }; + tokio::pin!(reply_rx); + loop { + tokio::select! { + reply = &mut reply_rx => return Some(reply), + _ = activity.notified() => continue, + _ = tokio::time::sleep(timeout) => return None, + } + } +} + +/// Waits for the streams of a value and fills their placeholders. +async fn complete_streams(receivers: StreamReceivers, values: &mut [Value]) -> Result<(), String> { + let mut contents = HashMap::with_capacity(receivers.len()); + for (id, rx) in receivers { + match rx.await { + Ok(Ok(bytes)) => { + contents.insert(id, bytes); + } + Ok(Err(code)) => return Err(format!("stream {id} was reset (code {code})")), + Err(_) => return Err(format!("stream {id} was dropped")), + } + } + for value in values { + convert::fill_streams(value, &mut contents)?; + } + Ok(()) +} + +fn transport_error(message: &str) -> RuntimeError { + RuntimeError::new("T-APUS-000001", "ApusTransport", message) +} + +/// The routing table and the projects this instance serves. +struct Routing { + routes: Routes, + /// `None`: no table configured, this instance serves every project. + served: Option, +} + +/// What this instance serves according to the routing table. +struct Served { + /// Projects listed for this instance. + listed: HashSet, + /// This instance is the default route: it also serves every project no + /// other instance lists. + default: bool, + /// Projects listed for other instances (only needed for the default route). + elsewhere: HashSet, +} + +impl Served { + fn new(routes: &Routes, instance_id: &str) -> Self { + let mut listed = HashSet::new(); + let mut elsewhere = HashSet::new(); + for route in &routes.routes { + let target = if route.instance_id == instance_id { + &mut listed + } else { + &mut elsewhere + }; + target.extend(route.projects.iter().copied()); + } + // The first instance with an empty list is the default route (as `Routes::instance_for`). + let default = routes + .routes + .iter() + .find(|r| r.projects.is_empty()) + .is_some_and(|r| r.instance_id == instance_id); + Self { + listed, + default, + elsewhere, + } + } + + fn contains(&self, project_id: i64) -> bool { + self.listed.contains(&project_id) || (self.default && !self.elsewhere.contains(&project_id)) + } + + fn is_empty(&self) -> bool { + self.listed.is_empty() && !self.default + } +} + +/// Registry of connected actions, keyed by action identifier. +pub struct ApusHub { + actions: RwLock>>>, + round_robin: AtomicUsize, + shards: Mutex>>, + routing: RwLock, + connections: AtomicUsize, +} + +impl ApusHub { + pub fn new(instance_id: &str, advertise: &str) -> Self { + Self { + actions: RwLock::new(HashMap::new()), + round_robin: AtomicUsize::new(0), + shards: Mutex::new(HashMap::new()), + routing: RwLock::new(Routing { + // Without a table this instance is the default route (empty project list). + routes: Routes { + version: 0, + routes: vec![Route { + instance_id: instance_id.to_string(), + endpoint: advertise.to_string(), + projects: Vec::new(), + }], + }, + served: None, + }), + connections: AtomicUsize::new(0), + } + } + + /// Picks a running connection of `service` that implements `function` and serves `project_id`. + pub fn pick( + &self, + service: &str, + function: &str, + project_id: i64, + ) -> Option<(Arc, u16)> { + let actions = self.actions.read().unwrap_or_else(|err| err.into_inner()); + let connections = actions.get(service)?; + if connections.is_empty() { + return None; + } + let start = self.round_robin.fetch_add(1, Ordering::Relaxed); + for offset in 0..connections.len() { + let connection = &connections[(start + offset) % connections.len()]; + if !connection.routable() || !connection.serves(project_id) { + continue; + } + if let Some(method) = connection.functions.get(function) { + return Some((connection.clone(), *method)); + } + } + None + } + + fn serves_project(&self, project_id: i64) -> bool { + let routing = self.routing.read().unwrap_or_else(|e| e.into_inner()); + routing + .served + .as_ref() + .is_none_or(|served| served.contains(project_id)) + } + + fn routes_frame(&self) -> Vec { + self.routing + .read() + .unwrap_or_else(|e| e.into_inner()) + .routes + .encode() + } + + /// Installs a new routing table and announces it to every connected action. + /// When this instance serves no project any more, actions are asked to leave. + pub fn set_routes(&self, routes: Routes, instance_id: &str) { + let served = Served::new(&routes, instance_id); + let empty = served.is_empty(); + let frame = routes.encode(); + log::info!( + "Apus routes v{}: instance {} serves {} project(s){}", + routes.version, + instance_id, + served.listed.len(), + if served.default { + " and is the default route" + } else { + "" + } + ); + *self.routing.write().unwrap_or_else(|e| e.into_inner()) = Routing { + routes, + served: Some(served), + }; + let actions = self.actions.read().unwrap_or_else(|err| err.into_inner()); + for connection in actions.values().flatten() { + let _ = connection.tx.send(frame.clone()); + if empty { + connection.draining.store(true, Ordering::Relaxed); + let _ = connection.tx.send(message::encode_goaway( + 0, + "this instance serves no project any more", + )); + } + } + } + + /// Graceful drain on shutdown: no new requests, in-flight ones finish. + pub fn goaway_all(&self, reason: &str) { + let actions = self.actions.read().unwrap_or_else(|err| err.into_inner()); + for connection in actions.values().flatten() { + connection.draining.store(true, Ordering::Relaxed); + let _ = connection.tx.send(message::encode_status(status::STOPPED)); + let _ = connection.tx.send(message::encode_goaway(0, reason)); + } + } + + fn register(&self, connection: Arc, max_per_action: usize) -> bool { + let mut actions = self.actions.write().unwrap_or_else(|err| err.into_inner()); + let list = actions.entry(connection.action.clone()).or_default(); + if list.len() >= max_per_action { + return false; + } + list.push(connection); + true + } + + fn unregister(&self, connection: &Arc) { + let mut actions = self.actions.write().unwrap_or_else(|err| err.into_inner()); + if let Some(list) = actions.get_mut(&connection.action) { + list.retain(|c| !Arc::ptr_eq(c, connection)); + } + } + + fn claim_shard(&self, action: &str, replicas: u32) -> Option { + let mut shards = self.shards.lock().unwrap_or_else(|e| e.into_inner()); + let claimed = shards + .entry(action.to_string()) + .or_insert_with(|| vec![false; replicas as usize]); + let index = claimed.iter().position(|taken| !taken)?; + claimed[index] = true; + Some(index as u32) + } + + fn release_shard(&self, action: &str, index: u32) { + let mut shards = self.shards.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(slot) = shards + .get_mut(action) + .and_then(|c| c.get_mut(index as usize)) + { + *slot = false; + } + } +} + +pub struct ServerSettings { + pub host: String, + pub port: u16, + pub instance_id: String, + /// EXEC + SUBFLOW requests accepted concurrently per connection. + pub max_in_flight: u32, + pub heartbeat: Duration, + pub max_frame: u32, + pub stream_window: u32, + pub stream_threshold: u32, + /// Mutual TLS: the client certificate identifies the action. `None`: plain TCP, + /// nobody is authenticated (development). + pub tls: Option, + pub max_connections: usize, + pub max_connections_per_action: usize, + pub handshake_timeout: Duration, +} + +pub async fn serve(settings: ServerSettings, hub: Arc, exec: Arc) { + match TcpListener::bind((settings.host.as_str(), settings.port)).await { + Ok(listener) => serve_on(listener, settings, hub, exec).await, + Err(err) => log::error!( + "Apus server failed to bind {}:{}: {}", + settings.host, + settings.port, + err + ), + } +} + +pub async fn serve_on( + listener: TcpListener, + settings: ServerSettings, + hub: Arc, + exec: Arc, +) { + log::info!( + "Apus v{} server on {} instance={} auth={}", + PROTOCOL_VERSION, + listener + .local_addr() + .map(|a| a.to_string()) + .unwrap_or_default(), + settings.instance_id, + if settings.tls.is_some() { + "client certificates" + } else { + "none (no TLS)" + }, + ); + let settings = Arc::new(settings); + loop { + match listener.accept().await { + Ok((stream, peer)) => { + if hub.connections.fetch_add(1, Ordering::Relaxed) >= settings.max_connections { + hub.connections.fetch_sub(1, Ordering::Relaxed); + log::warn!("Apus: connection limit reached, refusing {}", peer); + continue; + } + let hub = hub.clone(); + let exec = exec.clone(); + let settings = settings.clone(); + tokio::spawn(async move { + if let Err(err) = handle_connection(stream, hub.clone(), exec, settings).await { + log::warn!("Apus connection from {} ended: {}", peer, err); + } + hub.connections.fetch_sub(1, Ordering::Relaxed); + }); + } + Err(err) => log::warn!("Apus accept failed: {}", err), + } + } +} + +async fn reject_and_close( + writer: &mut BufWriter, + code: u16, + text: &str, +) -> std::io::Result<()> { + tokio::time::sleep(REJECT_DELAY).await; + writer + .write_all(&message::encode_reject(code, text)) + .await?; + writer.flush().await?; + Err(invalid(&format!("rejected HELLO: {text}"))) +} + +async fn handle_connection( + tcp: TcpStream, + hub: Arc, + exec: Arc, + settings: Arc, +) -> std::io::Result<()> { + tcp.set_nodelay(true)?; + // With TLS the verified client certificate says who the action is. + let (read_half, write_half, identity): ( + BoxRead, + BoxWrite, + Option>, + ) = match &settings.tls { + Some(acceptor) => { + let stream = tokio::time::timeout(settings.handshake_timeout, acceptor.accept(tcp)) + .await + .map_err(|_| invalid("TLS handshake timed out"))??; + let identity = match stream + .get_ref() + .1 + .peer_certificates() + .and_then(|certs| certs.first()) + { + Some(cert) => tls::peer_identity(cert).map_err(|err| err.to_string()), + None => Err("no client certificate".to_string()), + }; + let (r, w) = tokio::io::split(stream); + (Box::new(r), Box::new(w), Some(identity)) + } + None => { + let (r, w) = tcp.into_split(); + (Box::new(r), Box::new(w), None) + } + }; + let mut reader = BufReader::with_capacity(64 * 1024, read_half); + let mut writer = BufWriter::with_capacity(64 * 1024, write_half); + + // --- Handshake: HELLO (within the timeout, size-limited) -> WELCOME + ROUTES, or REJECT --- + let first = match tokio::time::timeout( + settings.handshake_timeout, + read_frame_limited(&mut reader, MAX_HELLO), + ) + .await + { + Ok(Ok(Some(frame))) => frame, + Ok(Ok(None)) => return Ok(()), + Ok(Err(err)) => return Err(err), + Err(_) => return Err(invalid("no HELLO within the handshake timeout")), + }; + if first.kind != kind::HELLO { + return reject_and_close( + &mut writer, + reject::PROTOCOL_ERROR, + "first frame must be HELLO", + ) + .await; + } + let hello = match Hello::decode(&first.body) { + Ok(hello) => hello, + Err(err) => { + return reject_and_close(&mut writer, reject::PROTOCOL_ERROR, &err.to_string()).await; + } + }; + if hello.version != PROTOCOL_VERSION { + return reject_and_close( + &mut writer, + reject::UNSUPPORTED_VERSION, + "unsupported protocol version", + ) + .await; + } + let identity = match identity { + Some(Ok(identity)) if identity.action == hello.action => Some(identity), + Some(Ok(_)) => { + return reject_and_close( + &mut writer, + reject::UNAUTHENTICATED, + "the client certificate is for another action", + ) + .await; + } + Some(Err(text)) => { + return reject_and_close(&mut writer, reject::UNAUTHENTICATED, &text).await; + } + None => None, + }; + let replicas = identity.as_ref().map_or(1, |identity| identity.replicas); + let shard = if hello.scaling == scaling::SPLIT && replicas > 1 { + match hub.claim_shard(&hello.action, replicas) { + Some(index) => Some((index, replicas)), + None => { + return reject_and_close( + &mut writer, + reject::SHARD_EXHAUSTED, + "all shards are claimed", + ) + .await; + } + } + } else { + None + }; + let action = hello.action.clone(); + let functions: HashMap = hello + .functions + .iter() + .enumerate() + .map(|(i, name)| (name.clone(), i as u16)) + .collect(); + + let (tx, mut rx) = mpsc::unbounded_channel::>(); + let stream_window = settings + .stream_window + .min(hello.stream_window.max(limits::STREAM_CHUNK as u32)); + let stream_threshold = if hello.features & feature::STREAMS != 0 { + settings.stream_threshold as usize + } else { + 0 + }; + let connection = Arc::new(Connection { + action: action.clone(), + functions, + tx: tx.clone(), + pending: Mutex::new(HashMap::new()), + next_id: AtomicU32::new(1), + call_slots: (hello.max_in_flight > 0) + .then(|| Arc::new(Semaphore::new(hello.max_in_flight as usize))), + projects: identity + .as_ref() + .and_then(|identity| identity.projects.clone()) + .map(|p| p.into_iter().collect()), + shard, + status: AtomicU8::new(status::NOT_READY), + draining: AtomicBool::new(false), + incoming: Mutex::new(Incoming::new(Side::Action)), + outgoing: Arc::new(Outgoing::new(stream_window)), + next_stream_id: AtomicU32::new(Side::Server.first_stream_id()), + stream_threshold, + requests_in_flight: AtomicU32::new(0), + max_in_flight: settings.max_in_flight, + overloaded: AtomicBool::new(false), + request_tasks: ShardedMap::new(64), + rx_frames: AtomicU64::new(0), + }); + if !hub.register(connection.clone(), settings.max_connections_per_action) { + if let Some((index, _)) = shard { + hub.release_shard(&action, index); + } + return reject_and_close( + &mut writer, + reject::TOO_MANY_CONNECTIONS, + "too many connections for this action", + ) + .await; + } + + let welcome = Welcome { + version: PROTOCOL_VERSION, + session_id: uuid::Uuid::new_v4().to_string(), + instance_id: settings.instance_id.clone(), + max_in_flight: settings.max_in_flight, + max_frame: settings.max_frame, + stream_window, + stream_threshold: stream_threshold as u32, + features: hello.features & feature::ALL, + heartbeat_ms: settings.heartbeat.as_millis() as u32, + }; + writer.write_all(&welcome.encode()).await?; + writer.write_all(&hub.routes_frame()).await?; + writer.flush().await?; + + // Writer task: drains queued frames and flushes once per batch. + let writer_task = tokio::spawn(async move { + while let Some(frame) = rx.recv().await { + if writer.write_all(&frame).await.is_err() { + return; + } + while let Ok(more) = rx.try_recv() { + if writer.write_all(&more).await.is_err() { + return; + } + } + if writer.flush().await.is_err() { + return; + } + } + }); + + log::info!( + "Apus action connected action={} functions={} shard={:?} projects={:?}", + action, + connection.functions.len(), + connection.shard, + connection.projects + ); + + let read_task = tokio::spawn(read_loop( + reader, + connection.clone(), + exec, + settings.clone(), + hub.clone(), + )); + let heartbeat_task = tokio::spawn(heartbeat( + connection.clone(), + settings.heartbeat, + read_task.abort_handle(), + )); + let sweeper_task = tokio::spawn(sweep(connection.clone())); + let expiry_task = identity.as_ref().map(|identity| { + tokio::spawn(expire( + connection.clone(), + identity.not_after, + read_task.abort_handle(), + )) + }); + let result = match read_task.await { + Ok(result) => result, + Err(err) if err.is_cancelled() => Err(invalid("peer missed its heartbeats")), + Err(err) => Err(invalid(&err.to_string())), + }; + if result.is_err() { + // Protocol error or dead peer: tell it why, best effort. + let _ = tx.send(message::encode_goaway( + reject::PROTOCOL_ERROR, + "protocol error", + )); + tokio::time::sleep(Duration::from_millis(10)).await; + } + + hub.unregister(&connection); + connection.fail_all(); + if let Some((index, _)) = connection.shard { + hub.release_shard(&action, index); + } + heartbeat_task.abort(); + sweeper_task.abort(); + if let Some(task) = expiry_task { + task.abort(); + } + writer_task.abort(); + log::info!("Apus action disconnected action={}", action); + result +} + +/// Ends a connection when its client certificate expires: TLS checks the certificate +/// only during the handshake, but connections live for days. `GOAWAY` first, so the +/// action renews its certificate (through Aquila) and reconnects; closed after a grace. +async fn expire(connection: Arc, not_after: i64, read_task: tokio::task::AbortHandle) { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_secs() as i64); + tokio::time::sleep(Duration::from_secs( + not_after.saturating_sub(now).max(0) as u64 + )) + .await; + log::info!( + "Apus client certificate expired action={}", + connection.action + ); + connection.draining.store(true, Ordering::Relaxed); + let _ = connection.tx.send(message::encode_goaway( + reject::UNAUTHENTICATED, + "client certificate expired", + )); + tokio::time::sleep(CERTIFICATE_GRACE).await; + read_task.abort(); +} + +/// Sends `PING` every interval; a peer that sent nothing for 3 intervals is dead. +async fn heartbeat( + connection: Arc, + interval: Duration, + read_task: tokio::task::AbortHandle, +) { + let mut ticker = tokio::time::interval(interval); + ticker.tick().await; + let mut last_seen = connection.rx_frames.load(Ordering::Relaxed); + let mut silent = 0; + let mut nonce = 0u64; + loop { + ticker.tick().await; + let seen = connection.rx_frames.load(Ordering::Relaxed); + if seen == last_seen { + silent += 1; + if silent >= 3 { + log::warn!( + "Apus action {} missed 3 heartbeats, closing", + connection.action + ); + connection + .status + .store(status::NOT_RESPONDING, Ordering::Relaxed); + read_task.abort(); + return; + } + } else { + silent = 0; + last_seen = seen; + } + nonce += 1; + if connection + .tx + .send(message::encode_ping(kind::PING, nonce)) + .is_err() + { + return; + } + } +} + +/// Once a second: ends requests past their deadline and resets idle incoming +/// streams (no timer per request or stream). +async fn sweep(connection: Arc) { + let mut ticker = tokio::time::interval(Duration::from_secs(1)); + loop { + ticker.tick().await; + let now = Instant::now(); + for (id, request) in connection + .request_tasks + .drain_where(|r| r.deadline.is_some_and(|deadline| deadline <= now)) + { + request.abort.abort(); + let error = ErrorInfo::new( + "APUS-DEADLINE", + "DEADLINE_EXCEEDED", + "request did not finish before its deadline", + ); + let _ = connection.tx.send(error.frame(request.error_kind, id)); + } + let expired = connection + .lock_incoming() + .expire(Duration::from_millis(limits::STREAM_IDLE_TIMEOUT_MS)); + for id in expired { + let _ = connection + .tx + .send(message::encode_reset(id, reset::TIMED_OUT)); + } + } +} + +fn protocol_error(text: &str) -> std::io::Error { + invalid(&format!("protocol error: {text}")) +} + +async fn read_loop( + mut reader: BufReader, + connection: Arc, + exec: Arc, + settings: Arc, + hub: Arc, +) -> std::io::Result<()> { + while let Some(frame) = read_frame_limited(&mut reader, settings.max_frame as usize).await? { + connection.rx_frames.fetch_add(1, Ordering::Relaxed); + if !kind::sent_by_action(frame.kind) { + if kind::is_known(frame.kind) { + return Err(protocol_error(&format!( + "an action may not send kind 0x{:02x}", + frame.kind + ))); + } + continue; // unknown kinds are ignored (forward compatibility) + } + let mut r = Reader::new(&frame.body); + match frame.kind { + kind::RETURN => { + let started_at = r.i64().map_err(decode_err)?; + let finished_at = r.i64().map_err(decode_err)?; + if frame.body.get(16) == Some(&apus_protocol::value::tag::FUNCTION) { + r.u8().map_err(decode_err)?; + let function = convert::read_function(&mut r).map_err(decode_err)?; + let value = NodeValueExecutionResult { + value: Some(node_value_execution_result::Value::SubFlowFunctionResult( + function, + )), + }; + connection.complete( + frame.id, + CallReply::Ok { + value, + started_at, + finished_at, + }, + ); + continue; + } + let mut streams = Vec::new(); + let value = convert::read_value(&mut r, &mut streams).map_err(decode_err)?; + if streams.is_empty() { + let value = raw_value_result(value); + connection.complete( + frame.id, + CallReply::Ok { + value, + started_at, + finished_at, + }, + ); + continue; + } + let receivers = connection.expect_streams(&streams); + let connection = connection.clone(); + tokio::spawn(async move { + let mut values = [value]; + let outcome = match receivers { + Ok(receivers) => complete_streams(receivers, &mut values).await, + Err(text) => Err(text), + }; + let reply = match outcome { + Ok(()) => CallReply::Ok { + value: raw_value_result(values.into_iter().next().unwrap()), + started_at, + finished_at, + }, + Err(text) => CallReply::Err { + error: ErrorInfo::new("T-APUS-000005", "STREAM", text), + started_at, + finished_at, + }, + }; + connection.complete(frame.id, reply); + }); + } + kind::ERROR => { + let started_at = r.i64().map_err(decode_err)?; + let finished_at = r.i64().map_err(decode_err)?; + let error = ErrorInfo::read(&mut r).map_err(decode_err)?; + connection.complete( + frame.id, + CallReply::Err { + error, + started_at, + finished_at, + }, + ); + } + kind::EXEC => { + let flow_id = r.i64().map_err(decode_err)?; + let project_id = r.i64().map_err(decode_err)?; + let execution_id = r.str().map_err(decode_err)?.to_string(); + let deadline_ms = r.u32().map_err(decode_err)?; + let mut streams = Vec::new(); + let payload = convert::read_value(&mut r, &mut streams).map_err(decode_err)?; + if let Some(error) = admission_error(&connection, &hub, frame.id, Some(project_id)) + { + reset_streams(&connection, &streams); + let _ = connection.tx.send(error.frame(kind::EXEC_ERROR, frame.id)); + continue; + } + let receivers = (!streams.is_empty()).then(|| connection.expect_streams(&streams)); + let deadline = (deadline_ms > 0) + .then(|| Instant::now() + Duration::from_millis(deadline_ms as u64)); + start_request(&connection, frame.id, kind::EXEC_ERROR, deadline, { + let connection = connection.clone(); + let exec = exec.clone(); + async move { + let mut values = [payload]; + if let Some(receivers) = receivers { + let receivers = receivers.map_err(|text| { + ExecFailure::new("T-APUS-000005", "STREAM", text) + })?; + complete_streams(receivers, &mut values) + .await + .map_err(|text| { + ExecFailure::new("T-APUS-000005", "STREAM", text) + })?; + } + let [payload] = values; + exec.execute( + &connection.action, + flow_id, + project_id, + &execution_id, + payload, + connection.projects.as_ref(), + ) + .await + } + }); + } + kind::SUBFLOW => { + let sub_flow_id = r.str().map_err(decode_err)?.to_string(); + let argc = r.u8().map_err(decode_err)? as usize; + let mut streams = Vec::new(); + let mut args = Vec::with_capacity(argc); + for _ in 0..argc { + args.push(convert::read_value(&mut r, &mut streams).map_err(decode_err)?); + } + if let Some(error) = admission_error(&connection, &hub, frame.id, None) { + reset_streams(&connection, &streams); + let _ = connection + .tx + .send(error.frame(kind::SUBFLOW_ERROR, frame.id)); + continue; + } + let receivers = (!streams.is_empty()).then(|| connection.expect_streams(&streams)); + start_request(&connection, frame.id, kind::SUBFLOW_ERROR, None, { + let exec = exec.clone(); + async move { + if let Some(receivers) = receivers { + let receivers = receivers.map_err(|text| { + ExecFailure::new("T-APUS-000005", "STREAM", text) + })?; + complete_streams(receivers, &mut args) + .await + .map_err(|text| { + ExecFailure::new("T-APUS-000005", "STREAM", text) + })?; + } + run_sub_flow(&exec, &sub_flow_id, args).await + } + }); + } + kind::CANCEL => { + if let Some(request) = connection.request_tasks.remove(frame.id) { + request.abort.abort(); + } + } + kind::STREAM_DATA => { + let outcome = connection.lock_incoming().data(frame.id, r.rest()); + let reply = match outcome { + DataOutcome::Credit(bytes) => message::encode_credit(frame.id, bytes), + DataOutcome::Reset(code) => message::encode_reset(frame.id, code), + }; + let _ = connection.tx.send(reply); + } + kind::STREAM_END => { + if !connection.lock_incoming().end(frame.id) { + let _ = connection + .tx + .send(message::encode_reset(frame.id, reset::UNKNOWN_STREAM)); + } + } + kind::STREAM_RESET => { + let code = r.u16().unwrap_or(reset::PROTOCOL_ERROR); + if Side::Server.owns(frame.id) { + connection.outgoing.reset(frame.id); + } else { + connection.lock_incoming().fail(frame.id, code); + } + } + kind::STREAM_CREDIT => { + let bytes = r.u32().map_err(decode_err)?; + if !Side::Server.owns(frame.id) { + return Err(protocol_error( + "credit for a stream the server did not open", + )); + } + connection.outgoing.credit(frame.id, bytes); + } + kind::PING => { + let mut pong = FrameBuilder::with_capacity(kind::PONG, frame.id, frame.body.len()); + pong.body().buf.extend_from_slice(&frame.body); + let _ = connection.tx.send(pong.finish()); + } + kind::PONG => {} + kind::GOAWAY => { + log::info!("Apus action {} is draining", connection.action); + connection.draining.store(true, Ordering::Relaxed); + } + kind::STATUS => { + let value = r.u8().map_err(decode_err)?; + if !matches!(value, status::NOT_READY | status::RUNNING | status::STOPPED) { + return Err(protocol_error("invalid STATUS")); + } + connection.status.store(value, Ordering::Relaxed); + } + _ => {} + } + } + Ok(()) +} + +/// Checks an action request before it starts: duplicate id, routing, ticket, +/// admission control. `None` = accepted (and counted as in flight). +fn admission_error( + connection: &Arc, + hub: &ApusHub, + id: u32, + project_id: Option, +) -> Option { + if connection.request_tasks.contains(id) { + return Some(ErrorInfo::new( + "APUS-DUPLICATE-ID", + "INVALID_ARGUMENT", + "request id already in flight", + )); + } + if let Some(project_id) = project_id { + if !hub.serves_project(project_id) { + return Some(ErrorInfo::new( + "APUS-WRONG-INSTANCE", + "UNAVAILABLE", + format!("project {project_id} is served by another instance"), + )); + } + if !connection.serves(project_id) { + return Some(ErrorInfo::new( + "APUS-FORBIDDEN", + "PERMISSION_DENIED", + format!("the ticket does not cover project {project_id}"), + )); + } + } + if connection + .requests_in_flight + .fetch_add(1, Ordering::Relaxed) + >= connection.max_in_flight + { + connection + .requests_in_flight + .fetch_sub(1, Ordering::Relaxed); + if !connection.overloaded.swap(true, Ordering::Relaxed) { + let _ = connection + .tx + .send(message::encode_status(status::NOT_READY)); + } + return Some(ErrorInfo::new( + "APUS-OVERLOADED", + "UNAVAILABLE", + "too many requests in flight", + )); + } + None +} + +fn reset_streams(connection: &Connection, streams: &[(u32, i64)]) { + for (id, _) in streams { + let _ = connection + .tx + .send(message::encode_reset(*id, reset::CANCELLED)); + } +} + +/// Runs one admitted action request (`EXEC` / `SUBFLOW`) with cancellation and +/// a deadline, and answers with the matching result frame. +fn start_request( + connection: &Arc, + id: u32, + error_kind: u8, + deadline: Option, + work: F, +) where + F: std::future::Future> + Send + 'static, +{ + let result_kind = error_kind - 1; // EXEC_RESULT / SUBFLOW_RESULT + + struct InFlight(Arc, u32); + impl Drop for InFlight { + fn drop(&mut self) { + let connection = &self.0; + let left = connection + .requests_in_flight + .fetch_sub(1, Ordering::Relaxed) + - 1; + connection.request_tasks.remove(self.1); + // Hysteresis: running again once more than half the slots are free. + if left.saturating_mul(2) < connection.max_in_flight + && connection.overloaded.swap(false, Ordering::Relaxed) + { + let _ = connection.tx.send(message::encode_status(status::RUNNING)); + } + } + } + + let guard = InFlight(connection.clone(), id); + let task_connection = connection.clone(); + // Hold the entry's shard while spawning so the task cannot remove its entry + // before it has been inserted. + let mut shard = connection.request_tasks.shard(id); + let task = tokio::spawn(async move { + let _guard = guard; + let connection = task_connection; + let mut streams = StreamOut::new(connection.stream_threshold, &connection.next_stream_id); + let frame = match work.await { + Ok(value) => { + let mut f = FrameBuilder::with_capacity(result_kind, id, 64); + convert::write_value(f.body(), &value, &mut Some(&mut streams)); + f.finish() + } + Err(failure) => failure.into_error().frame(error_kind, id), + }; + let _ = connection.tx.send(frame); + connection.send_streams(streams.pending); + }); + shard.insert( + id, + RunningRequest { + abort: task.abort_handle(), + deadline, + error_kind, + }, + ); +} + +pub(super) async fn run_sub_flow( + exec: &ExecContext, + sub_flow_id: &str, + args: Vec, +) -> Result { + match exec + .engine + .execute_sub_flow(sub_flow_id, args, Some(&exec.remote), exec.with_trace) + .await + { + Some(report) => match report.signal { + Signal::Success(value) | Signal::Return(value) => Ok(value), + Signal::Stop => Ok(Value { + kind: Some(tucana::shared::value::Kind::NullValue(0)), + }), + Signal::Failure(err) => Err(ExecFailure::new(&err.code, &err.category, err.message)), + }, + None => Err(ExecFailure::new( + "T-TAURUS-000002", + "SubFlowExecutionNotFound", + "No pending sub flow execution found for this execution identifier", + )), + } +} + +/// Loads a routing table file: `{"version": n, "instances": [{"id", "endpoint", "projects": [..]}]}`. +pub fn load_routes(path: &str) -> Result { + #[derive(serde::Deserialize)] + struct File { + version: u64, + instances: Vec, + } + #[derive(serde::Deserialize)] + struct Instance { + id: String, + endpoint: String, + #[serde(default)] + projects: Vec, + } + let data = std::fs::read_to_string(path).map_err(|e| format!("cannot read {path}: {e}"))?; + let file: File = + serde_json::from_str(&data).map_err(|e| format!("cannot parse {path}: {e}"))?; + Ok(Routes { + version: file.version, + routes: file + .instances + .into_iter() + .map(|i| Route { + instance_id: i.id, + endpoint: i.endpoint, + projects: i.projects, + }) + .collect(), + }) +} + +/// Loads the routing table and reloads it when the file changes (checked every second). +pub fn watch_routes(hub: Arc, path: String, instance_id: String) { + tokio::spawn(async move { + let mut last_modified = None; + let mut ticker = tokio::time::interval(Duration::from_secs(1)); + loop { + ticker.tick().await; + let modified = std::fs::metadata(&path).and_then(|m| m.modified()).ok(); + if modified.is_none() || modified == last_modified { + continue; + } + last_modified = modified; + match load_routes(&path) { + Ok(routes) => hub.set_routes(routes, &instance_id), + Err(err) => log::warn!("Apus: {}", err), + } + } + }); +} + +fn decode_err(err: apus_protocol::DecodeError) -> std::io::Error { + protocol_error(&err.to_string()) +} + +fn invalid(message: &str) -> std::io::Error { + std::io::Error::new(std::io::ErrorKind::InvalidData, message.to_string()) +} diff --git a/crates/taurus/src/app/apus/mod.rs b/crates/taurus/src/app/apus/mod.rs new file mode 100644 index 0000000..1d55707 --- /dev/null +++ b/crates/taurus/src/app/apus/mod.rs @@ -0,0 +1,14 @@ +//! Apus v1: direct connections between Taurus and actions. +//! See `Code0/apus/protocol/SPEC-v1.md`. + +pub mod convert; +pub mod exec; +pub mod hub; +pub mod remote; + +pub use exec::{ExecContext, ExecutionMode, FlowSource, load_flows_file}; +pub use hub::{ApusHub, ServerSettings, serve, watch_routes}; +pub use remote::RoutingRemote; + +#[cfg(test)] +mod tests; diff --git a/crates/taurus/src/app/apus/remote.rs b/crates/taurus/src/app/apus/remote.rs new file mode 100644 index 0000000..949f62e --- /dev/null +++ b/crates/taurus/src/app/apus/remote.rs @@ -0,0 +1,134 @@ +//! `RoutingRemote`: sends remote node calls over Apus when the target action +//! is connected directly, and falls back to the NATS/Aquila path otherwise. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use apus_protocol::TraceContext; +use taurus_core::runtime::remote::{RemoteExecution, RemoteRuntime}; +use taurus_core::time::now_unix_micros; +use taurus_core::types::errors::runtime_error::RuntimeError; +use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime; +use tonic::async_trait; +use tucana::shared::{NodeExecutionResult, node_execution_result}; + +use super::convert; +use super::hub::{ApusHub, CallReply}; + +#[derive(Clone)] +pub struct RoutingRemote { + apus: Option>, + nats: Option, + call_timeout: Duration, + /// Attach a W3C trace context to every `CALL` (normally: sampled executions only). + trace: bool, + spans: Arc, +} + +impl RoutingRemote { + pub fn new( + apus: Option>, + nats: Option, + call_timeout: Duration, + trace: bool, + ) -> Self { + Self { + apus, + nats, + call_timeout, + trace, + spans: Arc::new(AtomicU64::new(1)), + } + } + + fn trace_context(&self, execution_id: &str) -> Option { + if !self.trace { + return None; + } + let trace_id = uuid::Uuid::parse_str(execution_id) + .map(|id| *id.as_bytes()) + .unwrap_or_default(); + Some(TraceContext { + trace_id, + span_id: self.spans.fetch_add(1, Ordering::Relaxed).to_le_bytes(), + flags: 1, + }) + } +} + +#[async_trait] +impl RemoteRuntime for RoutingRemote { + async fn execute_remote( + &self, + execution: RemoteExecution, + ) -> Result { + let picked = self.apus.as_ref().and_then(|hub| { + hub.pick( + &execution.target_service, + &execution.request.function_identifier, + execution.request.project_id, + ) + }); + let Some((connection, method)) = picked else { + return match &self.nats { + Some(nats) => nats.execute_remote(execution).await, + None => Err(RuntimeError::new( + "T-APUS-000003", + "ApusNoRoute", + format!( + "action '{}' is not connected via Apus and NATS is disabled", + execution.target_service + ), + )), + }; + }; + + let trace = self.trace_context(&execution.request.execution_identifier); + let sent_at = now_unix_micros(); + let reply = connection + .call( + method, + &execution.request, + self.call_timeout, + execution.sub_flow_activity.as_ref(), + trace.as_ref(), + ) + .await?; + let received_at = now_unix_micros(); + + // The action's own timestamps, as in `ActionExecutionResponse.node_result`. + let (result, started_at, finished_at) = match reply { + CallReply::Ok { + value, + started_at, + finished_at, + } => ( + node_execution_result::Result::Success(value), + started_at, + finished_at, + ), + CallReply::Err { + error, + started_at, + finished_at, + } => ( + node_execution_result::Result::Error(convert::to_tucana_error(error)), + started_at, + finished_at, + ), + }; + + Ok(NodeExecutionResult { + started_at: if started_at > 0 { started_at } else { sent_at }, + finished_at: if finished_at > 0 { + finished_at + } else { + received_at + }, + parameter_results: Vec::new(), + id: None, + result: Some(result), + }) + } +} diff --git a/crates/taurus/src/app/apus/tests.rs b/crates/taurus/src/app/apus/tests.rs new file mode 100644 index 0000000..92a06cf --- /dev/null +++ b/crates/taurus/src/app/apus/tests.rs @@ -0,0 +1,941 @@ +//! Integration tests of the Apus v1 server: a raw-frame test action talks to a +//! real server (real engine, flows in memory) over loopback TCP or TLS. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use apus_protocol::frame::{Frame, FrameBuilder, kind}; +use apus_protocol::message::{Hello, Route, Routes, Welcome, encode_status, reset, status}; +use apus_protocol::value::tag; +use apus_protocol::{ApusValue, ErrorInfo, Reader, Writer, feature, read_frame, reject, tls}; +use taurus_core::runtime::engine::ExecutionEngine; +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, BufReader}; +use tokio::net::{TcpListener, TcpStream}; +use tucana::shared::number_value::Number; +use tucana::shared::value::Kind; +use tucana::shared::{ + InputType, ListValue, LiteralValue, NodeFunction, NodeParameter, NodeValue, NumberValue, + ReferenceValue, SubFlow, ValidationFlow, Value, node_value, reference_value, sub_flow, +}; + +use super::exec::{ExecContext, ExecutionMode, FlowSource}; +use super::hub::{ApusHub, ServerSettings, serve_on}; +use super::remote::RoutingRemote; + +const ACTION: &str = "test-action"; +const FUNCTIONS: [&str; 2] = ["test::echo", "test::map"]; + +fn int(i: i64) -> Value { + Value { + kind: Some(Kind::NumberValue(NumberValue { + number: Some(Number::Integer(i)), + })), + } +} + +fn string(s: &str) -> Value { + Value { + kind: Some(Kind::StringValue(s.into())), + } +} + +fn literal(db: i64, name: &str, value: Value) -> NodeParameter { + NodeParameter { + database_id: db, + runtime_parameter_id: name.into(), + value: Some(NodeValue { + value: Some(node_value::Value::LiteralValue(LiteralValue { + value: Some(value), + references: vec![], + })), + }), + cast: None, + } +} + +fn node(id: i64, function: &str, parameters: Vec, remote: bool) -> NodeFunction { + NodeFunction { + database_id: Some(id), + runtime_function_id: function.into(), + parameters, + next_node_id: None, + definition_source: remote.then(|| ACTION.to_string()), + } +} + +fn flow(flow_id: i64, project_id: i64, nodes: Vec) -> (i64, Arc) { + let starting_node_id = nodes[0].database_id.unwrap(); + ( + flow_id, + Arc::new(ValidationFlow { + flow_id, + project_id, + r#type: "REST".into(), + starting_node_id, + node_functions: nodes, + definition_source: Some(format!("action.{ACTION}")), + ..Default::default() + }), + ) +} + +/// 1, 2: echo a literal (projects 1 and 2). 3: echo a 100 KiB string. 4: map over [1, 2] with a sub-flow (+100). +/// 5: the action returns a function, which Taurus's `std::list::map` runs over [1, 2, 3]. +fn flows() -> HashMap> { + let sub_flow = NodeParameter { + database_id: 42, + runtime_parameter_id: "consumer".into(), + value: Some(NodeValue { + value: Some(node_value::Value::SubFlow(SubFlow { + execution_reference: Some(sub_flow::ExecutionReference::StartingNodeId(43)), + ..Default::default() + })), + }), + cast: None, + }; + let add = node( + 43, + "std::number::add", + vec![ + NodeParameter { + database_id: 431, + runtime_parameter_id: "first".into(), + value: Some(NodeValue { + value: Some(node_value::Value::ReferenceValue(ReferenceValue { + target: Some(reference_value::Target::InputType(InputType { + node_id: 41, + parameter_index: 1, + input_index: 0, + })), + paths: vec![], + })), + }), + cast: None, + }, + literal(432, "second", int(100)), + ], + false, + ); + HashMap::from([ + flow( + 1, + 1, + vec![node( + 11, + "test::echo", + vec![literal(111, "value", string("hello"))], + true, + )], + ), + flow( + 2, + 2, + vec![node( + 21, + "test::echo", + vec![literal(211, "value", string("other project"))], + true, + )], + ), + flow( + 3, + 1, + vec![node( + 31, + "test::echo", + vec![literal(311, "value", string(&"x".repeat(100 * 1024)))], + true, + )], + ), + flow( + 5, + 1, + vec![ + NodeFunction { + next_node_id: Some(52), + ..node( + 51, + "test::echo", + vec![literal(511, "value", string("square"))], + true, + ) + }, + node( + 52, + "std::list::map", + vec![ + literal( + 521, + "list", + Value { + kind: Some(Kind::ListValue(ListValue { + values: vec![int(1), int(2), int(3)], + })), + }, + ), + NodeParameter { + database_id: 522, + runtime_parameter_id: "transform".into(), + value: Some(NodeValue { + value: Some(node_value::Value::ReferenceValue(ReferenceValue { + target: Some(reference_value::Target::NodeId(51)), + paths: vec![], + })), + }), + cast: None, + }, + ], + false, + ), + ], + ), + flow( + 4, + 1, + vec![ + node( + 41, + "test::map", + vec![ + literal( + 411, + "items", + Value { + kind: Some(Kind::ListValue(ListValue { + values: vec![int(1), int(2)], + })), + }, + ), + sub_flow, + ], + true, + ), + add, + ], + ), + ]) +} + +struct Server { + port: u16, + hub: Arc, +} + +struct Options { + max_in_flight: u32, + heartbeat: Duration, + tls: Option, +} + +impl Default for Options { + fn default() -> Self { + Self { + max_in_flight: 64, + heartbeat: Duration::from_secs(30), + tls: None, + } + } +} + +async fn start(options: Options) -> Server { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let hub = Arc::new(ApusHub::new("taurus-a", &format!("127.0.0.1:{port}"))); + let exec = Arc::new(ExecContext { + client: None, + flows: FlowSource::Memory(flows()), + mode: ExecutionMode::Local, + engine: Arc::new(ExecutionEngine::new()), + remote: RoutingRemote::new(Some(hub.clone()), None, Duration::from_secs(5), false), + timeout: Duration::from_secs(5), + flow_type: "static".into(), + with_trace: false, + runtime_usage: std::sync::Mutex::new(None), + }); + let settings = ServerSettings { + host: "127.0.0.1".into(), + port, + instance_id: "taurus-a".into(), + max_in_flight: options.max_in_flight, + heartbeat: options.heartbeat, + max_frame: 1 << 20, + stream_window: 64 * 1024, + stream_threshold: 1024, + tls: options.tls, + max_connections: 100, + max_connections_per_action: 4, + handshake_timeout: Duration::from_millis(300), + }; + tokio::spawn(serve_on(listener, settings, hub.clone(), exec)); + Server { port, hub } +} + +/// A raw-frame test action. +struct Client { + reader: BufReader>, + writer: Box, +} + +fn hello() -> Vec { + Hello { + version: 1, + action: ACTION.into(), + scaling: 0, + max_in_flight: 0, + max_frame: 1 << 20, + stream_window: 64 * 1024, + features: feature::ALL, + functions: FUNCTIONS.map(String::from).to_vec(), + } + .encode() +} + +impl Client { + async fn raw(port: u16) -> Self { + let tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + tcp.set_nodelay(true).unwrap(); + let (r, w) = tcp.into_split(); + Self { + reader: BufReader::new(Box::new(r)), + writer: Box::new(w), + } + } + + async fn tls(port: u16, connector: tls::TlsConnector) -> Self { + let tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + let stream = connector + .connect(tls::server_name("localhost:0").unwrap(), tcp) + .await + .unwrap(); + let (r, w) = tokio::io::split(stream); + Self { + reader: BufReader::new(Box::new(r)), + writer: Box::new(w), + } + } + + /// Connects, authenticates and reports `running`; returns the WELCOME. + async fn connect(port: u16) -> (Self, Welcome) { + let mut client = Self::raw(port).await; + let welcome = client.handshake().await; + (client, welcome) + } + + async fn handshake(&mut self) -> Welcome { + self.send(hello()).await; + let welcome = self.recv().await.expect("WELCOME"); + assert_eq!(welcome.kind, kind::WELCOME); + let routes = self.recv().await.expect("ROUTES"); + assert_eq!(routes.kind, kind::ROUTES); + self.send(encode_status(status::RUNNING)).await; + Welcome::decode(&welcome.body).unwrap() + } + + async fn send(&mut self, frame: Vec) { + self.writer.write_all(&frame).await.unwrap(); + self.writer.flush().await.unwrap(); + } + + /// Next frame other than PING/STREAM_CREDIT, or None on close/timeout. + async fn recv(&mut self) -> Option { + loop { + let frame = tokio::time::timeout(Duration::from_secs(5), read_frame(&mut self.reader)) + .await + .ok()? + .ok()??; + if !matches!(frame.kind, kind::PING | kind::STREAM_CREDIT) { + return Some(frame); + } + } + } + + async fn exec(&mut self, id: u32, flow_id: i64, project_id: i64, deadline_ms: u32) { + let mut f = FrameBuilder::new(kind::EXEC, id); + let w = f.body(); + w.i64(flow_id); + w.i64(project_id); + w.str(&format!("00000000-0000-4000-8000-{id:012}")); + w.u32(deadline_ms); + w.null(); + self.send(f.finish()).await; + } + + async fn reply(&mut self, call_id: u32, value: &ApusValue) { + let mut f = FrameBuilder::new(kind::RETURN, call_id); + f.body().i64(1); + f.body().i64(2); + f.body().value(value); + self.send(f.finish()).await; + } + + /// Reads a CALL: (id, method, args). + async fn call(&mut self) -> (u32, u16, Vec) { + let frame = self.recv().await.expect("CALL"); + assert_eq!( + frame.kind, + kind::CALL, + "expected CALL, got 0x{:02x}", + frame.kind + ); + let mut r = Reader::new(&frame.body); + let method = r.u16().unwrap(); + let _execution = r.str().unwrap(); + let _project = r.i64().unwrap(); + let _deadline = r.u32().unwrap(); + let argc = r.u8().unwrap(); + ( + frame.id, + method, + (0..argc).map(|_| r.value().unwrap()).collect(), + ) + } + + async fn closed(&mut self) -> bool { + loop { + match tokio::time::timeout(Duration::from_secs(5), read_frame(&mut self.reader)).await { + Ok(Ok(None)) | Ok(Err(_)) => return true, + Ok(Ok(Some(_))) => continue, + Err(_) => return false, + } + } + } +} + +fn error_of(frame: &Frame) -> ErrorInfo { + ErrorInfo::read(&mut Reader::new(&frame.body)).unwrap() +} + +fn value_of(frame: &Frame) -> ApusValue { + Reader::new(&frame.body).value().unwrap() +} + +#[tokio::test] +async fn rejects_bad_handshakes() { + let server = start(Options::default()).await; + for (frame, code) in [( + apus_protocol::message::encode_ping(kind::PING, 1), + reject::PROTOCOL_ERROR, + )] { + let mut client = Client::raw(server.port).await; + client.send(frame).await; + let reject = client.recv().await.expect("REJECT"); + assert_eq!(reject.kind, kind::REJECT); + assert_eq!( + apus_protocol::message::decode_reject(&reject.body) + .unwrap() + .0, + code + ); + assert!(client.closed().await); + } +} + +#[tokio::test] +async fn closes_connections_without_hello() { + let server = start(Options::default()).await; + let mut client = Client::raw(server.port).await; + let started = std::time::Instant::now(); + assert!(client.closed().await); + assert!(started.elapsed() < Duration::from_secs(2)); +} + +#[tokio::test] +async fn executes_a_flow_that_calls_the_action() { + let server = start(Options::default()).await; + let (mut client, welcome) = Client::connect(server.port).await; + assert_eq!(welcome.instance_id, "taurus-a"); + client.exec(1, 1, 1, 5_000).await; + let (call, method, args) = client.call().await; + assert_eq!((method, args), (0, vec![ApusValue::string("hello")])); + client.reply(call, &ApusValue::string("echoed")).await; + let result = client.recv().await.unwrap(); + assert_eq!(result.kind, kind::EXEC_RESULT); + assert_eq!(value_of(&result), ApusValue::string("echoed")); +} + +#[tokio::test] +async fn a_returned_function_is_run_by_taurus() { + let server = start(Options::default()).await; + let (mut client, _) = Client::connect(server.port).await; + client.exec(1, 5, 1, 5_000).await; + let (call, _, _) = client.call().await; + client + .reply(call, &ApusValue::function("std::number::square", "")) + .await; + let result = client.recv().await.unwrap(); + assert_eq!(result.kind, kind::EXEC_RESULT, "{:?}", error_of(&result)); + assert_eq!( + value_of(&result), + ApusValue::List(vec![ + ApusValue::Int(1), + ApusValue::Int(4), + ApusValue::Int(9) + ]) + ); +} + +#[tokio::test] +async fn calls_only_reach_running_actions() { + let server = start(Options::default()).await; + let mut client = Client::raw(server.port).await; + client.send(hello()).await; + client.recv().await.unwrap(); // WELCOME + client.recv().await.unwrap(); // ROUTES, no STATUS running + client.exec(1, 1, 1, 5_000).await; + let result = client.recv().await.unwrap(); + assert_eq!(result.kind, kind::EXEC_ERROR); + assert_eq!(error_of(&result).code, "T-APUS-000003"); +} + +#[tokio::test] +async fn routes_projects_to_their_instance() { + let server = start(Options::default()).await; + let (mut client, _) = Client::connect(server.port).await; + server.hub.set_routes( + Routes { + version: 2, + routes: vec![ + Route { + instance_id: "taurus-a".into(), + endpoint: format!("127.0.0.1:{}", server.port), + projects: vec![1], + }, + Route { + instance_id: "taurus-b".into(), + endpoint: "127.0.0.1:1".into(), + projects: vec![2], + }, + ], + }, + "taurus-a", + ); + let routes = client.recv().await.unwrap(); + assert_eq!(routes.kind, kind::ROUTES); + let routes = Routes::decode(&routes.body).unwrap(); + assert_eq!(routes.instance_for(2).unwrap().instance_id, "taurus-b"); + client.exec(1, 2, 2, 5_000).await; + let result = client.recv().await.unwrap(); + assert_eq!(error_of(&result).code, "APUS-WRONG-INSTANCE"); + // A flow claimed for the wrong project is refused too. + client.exec(2, 2, 1, 5_000).await; + assert_eq!( + error_of(&client.recv().await.unwrap()).code, + "T-APUS-FLOW-000004" + ); +} + +#[tokio::test] +async fn the_default_route_serves_unlisted_projects() { + let server = start(Options::default()).await; + let (mut client, _) = Client::connect(server.port).await; + // taurus-a has an empty list (default route); taurus-b takes project 2. + server.hub.set_routes( + Routes { + version: 2, + routes: vec![ + Route { + instance_id: "taurus-a".into(), + endpoint: format!("127.0.0.1:{}", server.port), + projects: vec![], + }, + Route { + instance_id: "taurus-b".into(), + endpoint: "127.0.0.1:1".into(), + projects: vec![2], + }, + ], + }, + "taurus-a", + ); + assert_eq!( + client.recv().await.unwrap().kind, + kind::ROUTES, + "no GOAWAY: the default route still serves projects" + ); + client.exec(1, 1, 1, 5_000).await; + let (call, _, _) = client.call().await; + client.reply(call, &ApusValue::Null).await; + assert_eq!(client.recv().await.unwrap().kind, kind::EXEC_RESULT); + client.exec(2, 2, 2, 5_000).await; + assert_eq!( + error_of(&client.recv().await.unwrap()).code, + "APUS-WRONG-INSTANCE" + ); + // Dropped from the table: the instance asks its actions to leave. + server.hub.set_routes( + Routes { + version: 3, + routes: vec![Route { + instance_id: "taurus-b".into(), + endpoint: "127.0.0.1:1".into(), + projects: vec![], + }], + }, + "taurus-a", + ); + assert_eq!(client.recv().await.unwrap().kind, kind::ROUTES); + assert_eq!(client.recv().await.unwrap().kind, kind::GOAWAY); +} + +#[tokio::test] +async fn rejects_duplicate_ids_and_overload() { + let server = start(Options { + max_in_flight: 1, + ..Default::default() + }) + .await; + let (mut client, _) = Client::connect(server.port).await; + client.exec(7, 1, 1, 5_000).await; + let (call, _, _) = client.call().await; + client.exec(7, 1, 1, 5_000).await; + assert_eq!( + error_of(&client.recv().await.unwrap()).code, + "APUS-DUPLICATE-ID" + ); + client.exec(8, 1, 1, 5_000).await; + let status_frame = client.recv().await.unwrap(); + assert_eq!( + (status_frame.kind, status_frame.body[0]), + (kind::STATUS, status::NOT_READY) + ); + assert_eq!( + error_of(&client.recv().await.unwrap()).code, + "APUS-OVERLOADED" + ); + client.reply(call, &ApusValue::Null).await; + let mut kinds = vec![client.recv().await.unwrap(), client.recv().await.unwrap()]; + kinds.sort_by_key(|f| f.kind); + assert_eq!(kinds[0].kind, kind::STATUS); + assert_eq!(kinds[0].body[0], status::RUNNING); + assert_eq!(kinds[1].kind, kind::EXEC_RESULT); +} + +#[tokio::test] +async fn protocol_violations_close_only_that_connection() { + let server = start(Options::default()).await; + // An action may not send CALL. + let (mut client, _) = Client::connect(server.port).await; + client.send(FrameBuilder::new(kind::CALL, 1).finish()).await; + assert!(client.closed().await); + // A payload nested 10 000 levels deep. + let (mut client, _) = Client::connect(server.port).await; + let mut f = FrameBuilder::new(kind::EXEC, 1); + let w = f.body(); + w.i64(1); + w.i64(1); + w.str("e"); + w.u32(0); + for _ in 0..10_000 { + w.u8(tag::LIST); + w.u32(1); + } + w.null(); + client.send(f.finish()).await; + assert!(client.closed().await); + // The server still works for others. + let (mut client, _) = Client::connect(server.port).await; + client.exec(1, 1, 1, 5_000).await; + let (call, _, _) = client.call().await; + client.reply(call, &ApusValue::Null).await; + assert_eq!(client.recv().await.unwrap().kind, kind::EXEC_RESULT); +} + +#[tokio::test] +async fn streams_from_the_action_are_materialised() { + let server = start(Options::default()).await; + let (mut client, _) = Client::connect(server.port).await; + client.exec(1, 1, 1, 5_000).await; + let (call, _, _) = client.call().await; + let mut f = FrameBuilder::new(kind::RETURN, call); + f.body().i64(1); + f.body().i64(2); + f.body().stream(1, apus_protocol::value::TEXT_STREAM, 6); + client.send(f.finish()).await; + for chunk in ["str", "eam"] { + let mut data = FrameBuilder::new(kind::STREAM_DATA, 1); + data.body().buf.extend_from_slice(chunk.as_bytes()); + client.send(data.finish()).await; + } + client + .send(apus_protocol::frame::empty_frame(kind::STREAM_END, 1)) + .await; + let result = client.recv().await.unwrap(); + assert_eq!(value_of(&result), ApusValue::string("stream")); +} + +#[tokio::test] +async fn refuses_streams_with_the_wrong_parity() { + let server = start(Options::default()).await; + let (mut client, _) = Client::connect(server.port).await; + client.exec(1, 1, 1, 5_000).await; + let (call, _, _) = client.call().await; + let mut f = FrameBuilder::new(kind::RETURN, call); + f.body().i64(1); + f.body().i64(2); + f.body().stream(2, apus_protocol::value::TEXT_STREAM, 6); // even = server-owned id + client.send(f.finish()).await; + let reset_frame = client.recv().await.unwrap(); + assert_eq!( + ( + reset_frame.kind, + u16::from_le_bytes([reset_frame.body[0], reset_frame.body[1]]) + ), + (kind::STREAM_RESET, reset::PROTOCOL_ERROR) + ); + let result = client.recv().await.unwrap(); + assert_eq!(result.kind, kind::EXEC_ERROR); +} + +#[tokio::test] +async fn large_params_are_streamed_to_the_action() { + let server = start(Options::default()).await; + let (mut client, _) = Client::connect(server.port).await; + client.exec(1, 3, 1, 5_000).await; + let (call, _, args) = client.call().await; + let ApusValue::Stream { id, length, .. } = args[0] else { + panic!("expected a stream, got {:?}", args[0]) + }; + assert_eq!((id % 2, length), (0, 100 * 1024)); + let mut received = Vec::new(); + loop { + let frame = client.recv().await.unwrap(); + match frame.kind { + kind::STREAM_DATA => { + received.extend_from_slice(&frame.body); + client + .send(apus_protocol::message::encode_credit( + id, + frame.body.len() as u32, + )) + .await; + } + kind::STREAM_END => break, + other => panic!("unexpected frame 0x{other:02x}"), + } + } + assert_eq!(received, vec![b'x'; 100 * 1024]); + client + .reply(call, &ApusValue::Int(received.len() as i64)) + .await; + assert_eq!( + value_of(&client.recv().await.unwrap()), + ApusValue::Int(100 * 1024) + ); +} + +#[tokio::test] +async fn sub_flows_run_on_the_same_connection() { + let server = start(Options::default()).await; + let (mut client, _) = Client::connect(server.port).await; + client.exec(1, 4, 1, 5_000).await; + let (call, method, args) = client.call().await; + assert_eq!(method, 1); + let ApusValue::SubFlow { id: handle, .. } = &args[1] else { + panic!("expected a sub-flow handle") + }; + let mut results = Vec::new(); + for (request, item) in [(10u32, 1), (11, 2)] { + let mut f = FrameBuilder::new(kind::SUBFLOW, request); + f.body().str(handle); + f.body().u8(1); + f.body().int(item); + client.send(f.finish()).await; + let result = client.recv().await.unwrap(); + assert_eq!((result.kind, result.id), (kind::SUBFLOW_RESULT, request)); + results.push(value_of(&result)); + } + assert_eq!(results, vec![ApusValue::Int(101), ApusValue::Int(102)]); + client.reply(call, &ApusValue::List(results)).await; + assert_eq!(client.recv().await.unwrap().kind, kind::EXEC_RESULT); +} + +#[tokio::test] +async fn deadlines_end_requests() { + let server = start(Options::default()).await; + let (mut client, _) = Client::connect(server.port).await; + client.exec(1, 1, 1, 100).await; + client.call().await; // never answered + let started = std::time::Instant::now(); + let result = client.recv().await.unwrap(); + assert_eq!(error_of(&result).code, "APUS-DEADLINE"); + assert!(started.elapsed() < Duration::from_millis(1500)); +} + +#[tokio::test] +async fn cancel_stops_the_execution_and_its_call() { + let server = start(Options::default()).await; + let (mut client, _) = Client::connect(server.port).await; + client.exec(1, 1, 1, 5_000).await; + let (call, _, _) = client.call().await; + client + .send(apus_protocol::message::encode_cancel( + 1, + apus_protocol::cancel::CALLER, + )) + .await; + // Taurus cancels the outstanding CALL in turn and never answers the EXEC. + let cancel = client.recv().await.unwrap(); + assert_eq!((cancel.kind, cancel.id), (kind::CANCEL, call)); + client.reply(call, &ApusValue::Null).await; // late answer: ignored + assert!( + tokio::time::timeout(Duration::from_millis(300), client.recv()) + .await + .unwrap_or(None) + .is_none() + ); +} + +#[tokio::test] +async fn silent_actions_are_disconnected() { + let server = start(Options { + heartbeat: Duration::from_millis(100), + ..Default::default() + }) + .await; + let (mut client, welcome) = Client::connect(server.port).await; + assert_eq!(welcome.heartbeat_ms, 100); + let started = std::time::Instant::now(); + assert!(client.closed().await); + assert!(started.elapsed() < Duration::from_secs(2)); +} + +/// Aquila's role in the tests: a CA that issues server and action certificates. +struct TestCa { + cert: rcgen::Certificate, + key: rcgen::KeyPair, +} + +impl TestCa { + fn new() -> Self { + let key = rcgen::KeyPair::generate().unwrap(); + let mut params = rcgen::CertificateParams::new(Vec::::new()).unwrap(); + params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained); + let cert = params.self_signed(&key).unwrap(); + Self { cert, key } + } + + fn acceptor(&self) -> tls::TlsAcceptor { + let key = rcgen::KeyPair::generate().unwrap(); + let cert = rcgen::CertificateParams::new(vec!["localhost".to_string()]) + .unwrap() + .signed_by(&key, &self.cert, &self.key) + .unwrap(); + let ca = self.cert.pem(); + let config = tls::server_config( + cert.pem().as_bytes(), + key.serialize_pem().as_bytes(), + Some(ca.as_bytes()), + ) + .unwrap(); + tls::TlsAcceptor::from(config) + } + + /// An action certificate; `identity` = the URI SAN, `valid_for` = time until it expires. + fn connector(&self, identity: Option<&str>, valid_for: Option) -> tls::TlsConnector { + let key = rcgen::KeyPair::generate().unwrap(); + let mut params = rcgen::CertificateParams::new(vec!["action.local".to_string()]).unwrap(); + if let Some(uri) = identity { + params.subject_alt_names.push(rcgen::SanType::URI( + rcgen::Ia5String::try_from(uri).unwrap(), + )); + } + if let Some(valid_for) = valid_for { + params.not_after = time::OffsetDateTime::now_utc() + valid_for; + } + let cert = params.signed_by(&key, &self.cert, &self.key).unwrap(); + let ca = self.cert.pem(); + let config = tls::client_config( + ca.as_bytes(), + Some((cert.pem().as_bytes(), key.serialize_pem().as_bytes())), + ) + .unwrap(); + tls::TlsConnector::from(config) + } +} + +#[tokio::test] +async fn authenticates_actions_by_client_certificate() { + let ca = TestCa::new(); + let server = start(Options { + tls: Some(ca.acceptor()), + ..Default::default() + }) + .await; + + // The certificate names this action: calls work. + let identity = tls::identity_uri(ACTION, None, 1); + let mut client = Client::tls(server.port, ca.connector(Some(&identity), None)).await; + client.handshake().await; + client.exec(1, 1, 1, 5_000).await; + let (call, _, _) = client.call().await; + client.reply(call, &ApusValue::Bool(true)).await; + assert_eq!( + value_of(&client.recv().await.unwrap()), + ApusValue::Bool(true) + ); + + // A certificate for another action, or one without an identity: REJECT. + let other = tls::identity_uri("other-action", None, 1); + for connector in [ca.connector(Some(&other), None), ca.connector(None, None)] { + let mut client = Client::tls(server.port, connector).await; + client.send(hello()).await; + let frame = client.recv().await.expect("REJECT"); + assert_eq!(frame.kind, kind::REJECT); + assert_eq!( + apus_protocol::message::decode_reject(&frame.body) + .unwrap() + .0, + reject::UNAUTHENTICATED + ); + } + + // No client certificate, or plain TCP: the TLS handshake fails. + let without = + tls::TlsConnector::from(tls::client_config(ca.cert.pem().as_bytes(), None).unwrap()); + let mut client = Client::tls(server.port, without).await; + client.send(hello()).await; + assert!(client.closed().await); + let mut plain = Client::raw(server.port).await; + plain.send(hello()).await; + assert!(plain.closed().await); +} + +#[tokio::test] +async fn connections_end_when_the_certificate_expires() { + let ca = TestCa::new(); + let server = start(Options { + tls: Some(ca.acceptor()), + ..Default::default() + }) + .await; + let identity = tls::identity_uri(ACTION, Some(&[1]), 1); + let mut client = Client::tls( + server.port, + ca.connector(Some(&identity), Some(Duration::from_secs(2))), + ) + .await; + client.handshake().await; + let frame = client.recv().await.expect("GOAWAY"); + assert_eq!(frame.kind, kind::GOAWAY); + assert!( + server.hub.pick(ACTION, "test::echo", 1).is_none(), + "an expired connection gets no calls" + ); +} + +#[test] +fn loads_routing_files() { + let path = std::env::temp_dir().join(format!("apus-routes-{}.json", std::process::id())); + std::fs::write(&path, r#"{"version": 3, "instances": [{"id": "a", "endpoint": "h:1", "projects": [1, 2]}, {"id": "b", "endpoint": "h:2"}]}"#).unwrap(); + let routes = super::hub::load_routes(path.to_str().unwrap()).unwrap(); + std::fs::remove_file(path).ok(); + assert_eq!(routes.version, 3); + assert_eq!(routes.instance_for(2).unwrap().instance_id, "a"); + assert_eq!( + routes.instance_for(9).unwrap().instance_id, + "b", + "an instance without projects is the default route" + ); + let _ = Writer::default(); +} From 46cfdb34d26a74f912d5d99fe7f407b99c3aadae Mon Sep 17 00:00:00 2001 From: Raphael Date: Sun, 1 Nov 2026 11:34:54 +0100 Subject: [PATCH 3/4] feat: wire Apus into app/mod.rs and worker.rs app/mod.rs: declares pub mod apus; makes NATS optional (gated on the new nats_enabled config flag -- when disabled there's no execution queue, Apus only); builds an ApusHub when apus_enabled, constructs apus::RoutingRemote (hub + optional NATS remote + timeout + trace flag) and passes it to the worker in place of the bare NATS remote runtime; adds start_apus_server(...), which picks the FlowSource from either APUS_FLOWS_PATH or the NATS flow-store KV bucket, builds the rustls TlsAcceptor from apus_protocol::tls::server_config when all three TLS paths are configured (panics on startup if only some of the three are set), wires watch_routes, and spawns apus::serve; on shutdown calls hub.goaway_all(...). worker.rs: threads RoutingRemote through in place of the bare NATS remote runtime; after computing a flow run result, if the inbound execution message carries a NATS reply inbox, publishes the encoded result there, so an Apus instance that queued the execution over NATS gets the outcome back. Makes FlowRunResult and the functions exec.rs reuses (execute_flow, function_identifiers_by_node_id) pub(crate). --- crates/taurus/src/app/mod.rs | 210 ++++++++++++++++++++++++++++---- crates/taurus/src/app/worker.rs | 39 ++++-- 2 files changed, 214 insertions(+), 35 deletions(-) diff --git a/crates/taurus/src/app/mod.rs b/crates/taurus/src/app/mod.rs index 8d30e09..ddfa1df 100644 --- a/crates/taurus/src/app/mod.rs +++ b/crates/taurus/src/app/mod.rs @@ -7,6 +7,7 @@ //! the worker task to stop via a `Notify` and waits for it to exit on its own, //! so in-flight flow executions can finish before the process exits. +pub mod apus; mod definitions; mod worker; @@ -37,34 +38,83 @@ pub async fn run() { config.compiled_flow_cache_capacity, config.compiled_flow_cache_max_bytes, ); - let client = connect_nats(&config).await; + let client = if config.nats_enabled { + Some(connect_nats(&config).await) + } else { + log::info!("NATS disabled: no execution queue, Apus only"); + None + }; let mut health_task = spawn_health_task(&config); let (runtime_status_service, runtime_execution_service, mut runtime_status_heartbeat_task) = setup_dynamic_services_if_needed(&config).await; - let (runtime_usage, runtime_usage_task) = spawn_runtime_usage_if_needed(&config, &client); + let (runtime_usage, runtime_usage_task) = + spawn_runtime_usage_if_needed(&config, client.as_ref()); - let nats_remote = NATSRemoteRuntime::with_execution_result_timeout( - client.clone(), - Duration::from_secs(config.remote_runtime_timeout_secs), - ); - let worker_shutdown = Arc::new(Notify::new()); - let mut worker_task = worker::spawn_worker( - client, - engine, + let nats_remote = client.as_ref().map(|client| { + NATSRemoteRuntime::with_execution_result_timeout( + client.clone(), + Duration::from_secs(config.remote_runtime_timeout_secs), + ) + }); + let engine = Arc::new(engine); + let remote_timeout = Duration::from_secs(config.remote_runtime_timeout_secs); + let apus_hub = config.apus_enabled.then(|| { + Arc::new(apus::ApusHub::new( + &config.apus_instance_id, + &apus_advertise(&config), + )) + }); + let remote = apus::RoutingRemote::new( + apus_hub.clone(), nats_remote, - worker::ExecutionReporting { - runtime_execution_service, - runtime_usage, - result_sample_rate: config.execution_result_sample_rate, - }, - mode_label(&config).to_string(), - worker_shutdown.clone(), - config.max_concurrent_executions, - config.environment == Environment::Development, + remote_timeout, + config.apus_trace, ); + let mut apus_exec = None; + if let Some(hub) = &apus_hub { + apus_exec = Some( + start_apus_server( + &config, + hub.clone(), + client.as_ref(), + engine.clone(), + remote.clone(), + runtime_usage.clone(), + ) + .await, + ); + } + let worker_shutdown = Arc::new(Notify::new()); + let mut worker_task = match client.clone() { + Some(client) => worker::spawn_worker( + client, + engine, + remote, + worker::ExecutionReporting { + runtime_execution_service, + runtime_usage, + result_sample_rate: config.execution_result_sample_rate, + }, + mode_label(&config).to_string(), + worker_shutdown.clone(), + config.max_concurrent_executions, + config.environment == Environment::Development, + ), + // No queue worker without NATS; this task only ends on shutdown. + None => { + let shutdown = worker_shutdown.clone(); + tokio::spawn(async move { shutdown.notified().await }) + } + }; wait_for_shutdown(&mut worker_task, &mut health_task, &worker_shutdown).await; + if let Some(hub) = &apus_hub { + hub.goaway_all("taurus is shutting down"); + } + if let Some(exec) = &apus_exec { + exec.stop_runtime_usage(); + } // The worker owned every runtime usage recorder, so with it finished the // publisher flushes what's left and exits on its own. if let Some(handle) = runtime_usage_task @@ -93,6 +143,122 @@ pub async fn run() { telemetry.shutdown(); } +/// Accepts direct action connections (Apus data plane) next to the NATS worker. +async fn start_apus_server( + config: &Config, + hub: Arc, + client: Option<&async_nats::Client>, + engine: Arc, + remote: apus::RoutingRemote, + runtime_usage: Option, +) -> Arc { + let flows = if !config.apus_flows_path.is_empty() { + let flows = apus::load_flows_file(&config.apus_flows_path); + log::info!("Apus: {} flow(s) loaded into memory", flows.len()); + apus::FlowSource::Memory(flows) + } else { + let client = client.expect("Apus: NATS is disabled, set APUS_FLOWS_PATH"); + let jetstream = async_nats::jetstream::new(client.clone()); + let kv = match jetstream + .get_key_value(config.apus_flow_bucket.clone()) + .await + { + Ok(kv) => kv, + Err(err) => panic!( + "Apus: cannot open flow KV bucket '{}': {}", + config.apus_flow_bucket, err + ), + }; + apus::FlowSource::Kv { + kv, + cache: config + .apus_flow_cache + .then(|| Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()))), + } + }; + let mode = match (client, config.apus_execution.as_str()) { + (None, _) | (_, "local") => apus::ExecutionMode::Local, + _ => apus::ExecutionMode::Nats, + }; + log::info!( + "Apus enabled execution_mode={:?} nats={} flows_in_memory={} flow_cache={}", + mode, + client.is_some(), + !config.apus_flows_path.is_empty(), + config.apus_flow_cache + ); + let exec = Arc::new(apus::ExecContext { + client: client.cloned(), + flows, + mode, + engine, + remote, + timeout: Duration::from_secs(config.remote_runtime_timeout_secs), + flow_type: mode_label(config).to_string(), + with_trace: config.environment == Environment::Development, + runtime_usage: std::sync::Mutex::new(runtime_usage), + }); + exec.watch_flows(); + // With TLS, connections are mutual: the client certificate (issued by Aquila's CA) + // identifies the action. Without TLS (development) nobody is authenticated. + let tls_files = [ + &config.apus_tls_cert, + &config.apus_tls_key, + &config.apus_tls_client_ca, + ]; + let tls = if tls_files.iter().all(|path| path.is_empty()) { + log::warn!("Apus without TLS: actions are not authenticated (development only)"); + None + } else if tls_files.iter().any(|path| path.is_empty()) { + panic!( + "Apus: TLS needs APUS_TLS_CERT, APUS_TLS_KEY and APUS_TLS_CLIENT_CA (client certificates identify actions)" + ); + } else { + let read = |path: &str| { + std::fs::read(path).unwrap_or_else(|err| panic!("Apus: cannot read {path}: {err}")) + }; + let server = apus_protocol::tls::server_config( + &read(&config.apus_tls_cert), + &read(&config.apus_tls_key), + Some(&read(&config.apus_tls_client_ca)), + ) + .unwrap_or_else(|err| panic!("Apus: {err}")); + Some(apus_protocol::tls::TlsAcceptor::from(server)) + }; + if !config.apus_routing_path.is_empty() { + apus::watch_routes( + hub.clone(), + config.apus_routing_path.clone(), + config.apus_instance_id.clone(), + ); + } + let settings = apus::ServerSettings { + host: config.apus_host.clone(), + port: config.apus_port, + instance_id: config.apus_instance_id.clone(), + max_in_flight: config.apus_max_in_flight, + heartbeat: Duration::from_millis(config.apus_heartbeat_ms), + max_frame: config.apus_max_frame, + stream_window: apus_protocol::limits::DEFAULT_STREAM_WINDOW, + stream_threshold: config.apus_stream_threshold, + tls, + max_connections: config.apus_max_connections, + max_connections_per_action: config.apus_max_connections_per_action, + handshake_timeout: Duration::from_millis(apus_protocol::limits::HANDSHAKE_TIMEOUT_MS), + }; + tokio::spawn(apus::serve(settings, hub, exec.clone())); + exec +} + +/// The endpoint other parties reach this Apus server at (for `ROUTES`). +fn apus_advertise(config: &Config) -> String { + if config.apus_advertise.is_empty() { + format!("{}:{}", config.apus_host, config.apus_port) + } else { + config.apus_advertise.clone() + } +} + fn init_telemetry(config: &Config) -> telemetry::Telemetry { telemetry::Telemetry::initialize( &config.opentelemetry, @@ -278,11 +444,11 @@ async fn setup_dynamic_services_if_needed( /// Aquila drains the usage stream and forwards it to Sagittarius. fn spawn_runtime_usage_if_needed( config: &Config, - client: &async_nats::Client, + client: Option<&async_nats::Client>, ) -> (Option, Option>) { - if config.mode != DYNAMIC { + let Some(client) = client.filter(|_| config.mode == DYNAMIC) else { return (None, None); - } + }; let (recorder, handle) = runtime_usage::spawn( client.clone(), diff --git a/crates/taurus/src/app/worker.rs b/crates/taurus/src/app/worker.rs index ef1f275..0235e8c 100644 --- a/crates/taurus/src/app/worker.rs +++ b/crates/taurus/src/app/worker.rs @@ -37,6 +37,7 @@ //! [`crate::app::wait_for_shutdown`]) so the worker stops accepting new //! messages and waits for every in-flight execution to finish. +use crate::app::apus::RoutingRemote; use futures_lite::StreamExt; use prost::Message; use std::sync::Arc; @@ -45,7 +46,6 @@ use taurus_core::runtime::remote::RemoteRuntime; use taurus_core::time::now_unix_micros; use taurus_core::types::errors::runtime_error::RuntimeError; use taurus_core::types::signal::Signal; -use taurus_provider::providers::remote::nats_remote_runtime::NATSRemoteRuntime; use tokio::sync::{Notify, Semaphore}; use tokio::task::{JoinHandle, JoinSet}; use tucana::aquila::ActionSubFlowExecutionRequest; @@ -80,15 +80,14 @@ impl ExecutionReporting { pub fn spawn_worker( client: async_nats::Client, - engine: ExecutionEngine, - nats_remote: NATSRemoteRuntime, + engine: Arc, + nats_remote: RoutingRemote, reporting: ExecutionReporting, flow_type: String, shutdown: Arc, max_concurrent_executions: usize, with_trace: bool, ) -> JoinHandle<()> { - let engine = Arc::new(engine); let semaphore = Arc::new(Semaphore::new(max_concurrent_executions.max(1))); tokio::spawn(async move { @@ -149,10 +148,12 @@ pub fn spawn_worker( let nats_remote = nats_remote.clone(); let reporting = reporting.clone(); let flow_type = flow_type.clone(); + let client = client.clone(); in_flight.spawn(async move { let _permit = permit; process_execution_message( message, + &client, &engine, &nats_remote, reporting, @@ -223,8 +224,9 @@ pub fn spawn_worker( async fn process_execution_message( message: async_nats::Message, + client: &async_nats::Client, engine: &ExecutionEngine, - nats_remote: &NATSRemoteRuntime, + nats_remote: &RoutingRemote, mut reporting: ExecutionReporting, flow_type: &str, with_trace: bool, @@ -295,6 +297,15 @@ async fn process_execution_message( flow_id ); + // Apus: a Taurus instance that queued this execution on behalf of a + // directly connected action waits on the reply inbox for the outcome. + if let Some(reply) = message.reply.clone() { + let outcome = reported_execution_result(run_result.clone(), false); + if let Err(err) = client.publish(reply, outcome.encode_to_vec().into()).await { + log::error!("Failed to publish execution result reply: {:?}", err); + } + } + if let Some(runtime_usage) = reporting.runtime_usage.as_ref() { runtime_usage.record( run_result.flow_id, @@ -342,7 +353,7 @@ fn reported_execution_result(run_result: FlowRunResult, sampled: bool) -> Execut async fn process_sub_flow_execution_message( message: async_nats::Message, engine: &ExecutionEngine, - nats_remote: &NATSRemoteRuntime, + nats_remote: &RoutingRemote, client: &async_nats::Client, with_trace: bool, ) { @@ -503,17 +514,17 @@ fn build_sub_flow_decode_error_result() -> ExecutionResult { } #[derive(Clone)] -struct FlowRunResult { +pub(crate) struct FlowRunResult { execution_id: ExecutionId, - flow_id: i64, - started_at: i64, - finished_at: i64, + pub(crate) flow_id: i64, + pub(crate) started_at: i64, + pub(crate) finished_at: i64, input: Option, - signal: Signal, + pub(crate) signal: Signal, node_execution_results: Vec, } -async fn execute_flow( +pub(crate) async fn execute_flow( execution_id: ExecutionId, flow: ExecutionFlow, engine: &ExecutionEngine, @@ -555,7 +566,9 @@ async fn execute_flow( } } -fn function_identifiers_by_node_id(flow: &ExecutionFlow) -> std::collections::HashMap { +pub(crate) fn function_identifiers_by_node_id( + flow: &ExecutionFlow, +) -> std::collections::HashMap { flow.node_functions .iter() .filter_map(|function| { From 884c7cbfddce4291f52c60e0d14e5a7e0592b5f3 Mon Sep 17 00:00:00 2001 From: Raphael Date: Sun, 1 Nov 2026 11:35:02 +0100 Subject: [PATCH 4/4] config: add APUS_* env vars and NATS_ENABLED Adds Config fields (each via env_with_default) for the Apus data-plane server: APUS_ENABLED, APUS_HOST, APUS_PORT, APUS_MAX_IN_FLIGHT, APUS_HEARTBEAT_MS, APUS_TRACE, APUS_INSTANCE_ID, APUS_ADVERTISE, APUS_ROUTING_PATH, APUS_TLS_CERT/_KEY/_CLIENT_CA, APUS_STREAM_THRESHOLD, APUS_MAX_FRAME, APUS_MAX_CONNECTIONS, APUS_MAX_CONNECTIONS_PER_ACTION, APUS_EXECUTION, APUS_FLOW_BUCKET, APUS_FLOW_CACHE, APUS_FLOWS_PATH, and NATS_ENABLED (default true) to let NATS be disabled for an Apus-only deployment. --- crates/taurus/src/config/mod.rs | 61 +++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/crates/taurus/src/config/mod.rs b/crates/taurus/src/config/mod.rs index af426ed..68c64cd 100644 --- a/crates/taurus/src/config/mod.rs +++ b/crates/taurus/src/config/mod.rs @@ -91,6 +91,43 @@ pub struct Config { /// OpenTelemetry exporter configuration. pub opentelemetry: OpenTelemetry, + + /// Apus: accept direct action connections (`APUS_ENABLED`). + pub apus_enabled: bool, + pub apus_host: String, + pub apus_port: u16, + /// `EXEC` + `SUBFLOW` requests accepted concurrently per action connection. + pub apus_max_in_flight: u32, + pub apus_heartbeat_ms: u64, + /// Attach a trace context to every `CALL`. + pub apus_trace: bool, + /// This instance in the routing table, and the endpoint actions reach it at. + pub apus_instance_id: String, + pub apus_advertise: String, + /// Routing table file (`{"version", "instances": [{"id", "endpoint", "projects"}]}`), + /// reloaded on change. Empty: this instance serves every project. + pub apus_routing_path: String, + /// TLS (all three or none): certificate chain, key and the client CA (Aquila's) whose + /// certificates identify actions. Without TLS nobody is authenticated (development). + pub apus_tls_cert: String, + pub apus_tls_key: String, + pub apus_tls_client_ca: String, + /// Strings of at least this many bytes are sent as streams (0 = never). + pub apus_stream_threshold: u32, + pub apus_max_frame: u32, + pub apus_max_connections: usize, + pub apus_max_connections_per_action: usize, + /// `nats` (default): queue action-triggered executions on NATS. + /// `local`: execute them in this process. + pub apus_execution: String, + pub apus_flow_bucket: String, + /// Cache flows looked up for `EXEC` instead of reading the KV store each time. + pub apus_flow_cache: bool, + /// Load flows into memory from this JSON export (`tucana::shared::Flows`) instead of + /// the NATS KV store. Stands in for Aquila pushing flows to Taurus (control plane). + pub apus_flows_path: String, + /// `false`: no NATS at all (no connection, no `execution.*` worker, no KV store). + pub nats_enabled: bool, } /// Implementation for all relevant `Taurus` startup configurations @@ -162,6 +199,30 @@ impl Config { metrics_endpoint: optional_env("OPENTELEMETRY_METRICS_ENDPOINT"), traces_endpoint: optional_env("OPENTELEMETRY_TRACES_ENDPOINT"), }, + apus_enabled: env_with_default("APUS_ENABLED", false), + apus_host: env_with_default("APUS_HOST", String::from("127.0.0.1")), + apus_port: env_with_default("APUS_PORT", 7676_u16), + apus_max_in_flight: env_with_default("APUS_MAX_IN_FLIGHT", 4096_u32), + apus_heartbeat_ms: env_with_default("APUS_HEARTBEAT_MS", 10_000_u64), + apus_trace: env_with_default("APUS_TRACE", false), + apus_instance_id: env_with_default("APUS_INSTANCE_ID", String::from("taurus-local")), + apus_advertise: env_with_default("APUS_ADVERTISE", String::new()), + apus_routing_path: env_with_default("APUS_ROUTING_PATH", String::new()), + apus_tls_cert: env_with_default("APUS_TLS_CERT", String::new()), + apus_tls_key: env_with_default("APUS_TLS_KEY", String::new()), + apus_tls_client_ca: env_with_default("APUS_TLS_CLIENT_CA", String::new()), + apus_stream_threshold: env_with_default("APUS_STREAM_THRESHOLD", 64 * 1024_u32), + apus_max_frame: env_with_default("APUS_MAX_FRAME", 16 * 1024 * 1024_u32), + apus_max_connections: env_with_default("APUS_MAX_CONNECTIONS", 10_000_usize), + apus_max_connections_per_action: env_with_default( + "APUS_MAX_CONNECTIONS_PER_ACTION", + 64_usize, + ), + apus_execution: env_with_default("APUS_EXECUTION", String::from("nats")), + apus_flow_bucket: env_with_default("APUS_FLOW_BUCKET", String::from("flow_store")), + apus_flow_cache: env_with_default("APUS_FLOW_CACHE", false), + apus_flows_path: env_with_default("APUS_FLOWS_PATH", String::new()), + nats_enabled: env_with_default("NATS_ENABLED", true), } } }