From a0cbd602f291010ad0a54c0282c4e9735b4e5d24 Mon Sep 17 00:00:00 2001 From: ashaffah Date: Fri, 26 Jun 2026 12:44:12 +0700 Subject: [PATCH 1/5] fix(delay): abort in-flight delay tasks on stop Delay mode spawned a detached task per message; on stop they were not cancelled (lingering up to the delay), and the handles were never tracked. Hold them in AbortOnDrop guards (pruning finished ones so the list stays bounded) so pending delays are aborted when the node stops. --- src/runtime/nodes/delay.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/runtime/nodes/delay.rs b/src/runtime/nodes/delay.rs index 2cfd4a9..a09e8ef 100644 --- a/src/runtime/nodes/delay.rs +++ b/src/runtime/nodes/delay.rs @@ -7,7 +7,7 @@ use async_trait::async_trait; use serde::Deserialize; use serde_json::Value; -use crate::runtime::node::{Node, NodeCtx}; +use crate::runtime::node::{AbortOnDrop, Node, NodeCtx}; use crate::runtime::nodes::parse_config; fn default_mode() -> String { @@ -61,6 +61,8 @@ impl Node for DelayNode { // Default output (port 0) wires, for the spawned delay tasks. let outs = ctx.outs.first().cloned().unwrap_or_default(); let mut last: Option = None; + // In-flight delay tasks; aborted when the node stops (the Vec drops). + let mut pending: Vec = Vec::new(); while let Some(msg) = ctx.rx.recv().await { if self.mode == "rate" { @@ -69,13 +71,15 @@ impl Node for DelayNode { } } else { // Delay each message independently so the stream is not blocked. + pending.retain(|t| !t.0.is_finished()); // drop completed guards let outs = outs.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { tokio::time::sleep(interval).await; for out in &outs { let _ = out.send(msg.clone()).await; } }); + pending.push(AbortOnDrop(handle)); } } Ok(()) From bc1b3f0e9c9b536921ca636927df54a895c037e7 Mon Sep 17 00:00:00 2001 From: ashaffah Date: Fri, 26 Jun 2026 12:46:44 +0700 Subject: [PATCH 2/5] fix(auth): expire session tokens and prune on login Session tokens were kept in a set forever (removed only on logout), so repeated logins grew it unbounded. Store each token with a 7-day expiry, prune expired entries on login, and treat expired tokens as invalid. Add a session issue/validate/revoke round-trip test. --- src/auth.rs | 31 +++++++++++++++++++++++++------ 1 file changed, 25 insertions(+), 6 deletions(-) diff --git a/src/auth.rs b/src/auth.rs index b0d2165..225dbeb 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -5,7 +5,7 @@ //! The token is sent as `Authorization: Bearer `, or via the `?token=` //! query for SSE (EventSource cannot set headers). -use std::collections::HashSet; +use std::collections::HashMap; use std::sync::Mutex; use std::time::{Duration, Instant}; @@ -25,6 +25,8 @@ use crate::AppState; const MAX_FAILED_ATTEMPTS: u32 = 5; /// How long login stays locked after too many failures. const LOCKOUT: Duration = Duration::from_secs(30); +/// How long a session token stays valid after it is issued. +const SESSION_TTL: Duration = Duration::from_secs(7 * 24 * 60 * 60); /// Throttle state for the login endpoint (single admin account → process-global). #[derive(Default)] @@ -33,11 +35,11 @@ struct LoginGuard { locked_until: Option, } -/// Stores credentials + the set of active session tokens. +/// Stores credentials + active session tokens mapped to their expiry. pub struct Auth { user: String, pass: String, - sessions: Mutex>, + sessions: Mutex>, login_guard: Mutex, } @@ -46,7 +48,7 @@ impl Auth { Self { user, pass, - sessions: Mutex::new(HashSet::new()), + sessions: Mutex::new(HashMap::new()), login_guard: Mutex::new(LoginGuard::default()), } } @@ -87,12 +89,19 @@ impl Auth { fn issue(&self) -> String { *self.login_guard.lock().unwrap() = LoginGuard::default(); let token = Uuid::new_v4().to_string(); - self.sessions.lock().unwrap().insert(token.clone()); + let now = Instant::now(); + let mut sessions = self.sessions.lock().unwrap(); + // Drop expired tokens so the map stays bounded under repeated logins. + sessions.retain(|_, &mut expires| expires > now); + sessions.insert(token.clone(), now + SESSION_TTL); token } fn valid(&self, token: &str) -> bool { - self.sessions.lock().unwrap().contains(token) + match self.sessions.lock().unwrap().get(token) { + Some(&expires) => expires > Instant::now(), + None => false, + } } fn revoke(&self, token: &str) { @@ -191,6 +200,16 @@ mod tests { assert!(!ct_eq(b"secret", b"secre")); } + #[test] + fn session_issue_validate_revoke() { + let a = auth(); + let token = a.issue(); + assert!(a.valid(&token)); + assert!(!a.valid("not-a-token")); + a.revoke(&token); + assert!(!a.valid(&token)); + } + #[test] fn verify_accepts_only_exact_credentials() { let a = auth(); From dbf7effb8db4c869a5051b2adb64d077a1cb3211 Mon Sep 17 00:00:00 2001 From: ashaffah Date: Fri, 26 Jun 2026 12:48:34 +0700 Subject: [PATCH 3/5] fix(mqtt): build connection outside the registry lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_or_create held the global registry mutex while calling build(), which for tls/wss constructs a TLS config that can panic on unreadable platform certs — poisoning the mutex and breaking all future MQTT connects. Build outside the lock and double-check on insert (a concurrent task may have created the same connection meanwhile, in which case the redundant one is discarded). Confines a build panic to the one node. --- src/runtime/nodes/mqtt_shared.rs | 27 ++++++++++++++++++++++----- 1 file changed, 22 insertions(+), 5 deletions(-) diff --git a/src/runtime/nodes/mqtt_shared.rs b/src/runtime/nodes/mqtt_shared.rs index 355ea34..a5b0e3e 100644 --- a/src/runtime/nodes/mqtt_shared.rs +++ b/src/runtime/nodes/mqtt_shared.rs @@ -104,12 +104,17 @@ pub fn get_or_create( node_id: String, build: impl FnOnce() -> Result<(Connection, Announce, Announce), String>, ) -> Result, String> { - let mut reg = registry().lock().unwrap(); - if let Some(shared) = reg.get(&key) { - shared.refs.fetch_add(1, Ordering::SeqCst); - return Ok(shared.clone()); + // Fast path: reuse an existing connection. + { + let reg = registry().lock().unwrap(); + if let Some(shared) = reg.get(&key) { + shared.refs.fetch_add(1, Ordering::SeqCst); + return Ok(shared.clone()); + } } + // Build outside the registry lock: constructing a TLS config can panic on + // unreadable platform certs, and that must not poison the global mutex. let ((client_box, poller), birth, close) = build()?; let client: Arc = Arc::from(client_box); let subs: Arc>> = Arc::new(Mutex::new(Vec::new())); @@ -121,7 +126,6 @@ pub fn get_or_create( events, node_id, )); - let shared = Arc::new(Shared { key: key.clone(), client, @@ -130,6 +134,19 @@ pub fn get_or_create( driver: Mutex::new(Some(driver)), close, }); + + // Insert, unless another task created the same connection while we built. + let mut reg = registry().lock().unwrap(); + if let Some(existing) = reg.get(&key) { + existing.refs.fetch_add(1, Ordering::SeqCst); + let existing = existing.clone(); + drop(reg); + // Discard our redundant connection. + if let Some(handle) = shared.driver.lock().unwrap().take() { + handle.abort(); + } + return Ok(existing); + } reg.insert(key, shared.clone()); Ok(shared) } From 86b11593eae36b766042ee1efae51fdd07a35d31 Mon Sep 17 00:00:00 2001 From: ashaffah Date: Fri, 26 Jun 2026 14:49:33 +0700 Subject: [PATCH 4/5] fix(link): refcount link channels and drop unused ones The link registry kept a broadcast channel per name forever; over many deploys with distinct names it grew unbounded. Refcount each name and remove its channel when the last link-in/out using it stops (via an RAII guard). Add a test that the entry is removed on last release. --- src/runtime/nodes/link.rs | 79 +++++++++++++++++++++++++++++++-------- 1 file changed, 63 insertions(+), 16 deletions(-) diff --git a/src/runtime/nodes/link.rs b/src/runtime/nodes/link.rs index 5340b21..3ec82fc 100644 --- a/src/runtime/nodes/link.rs +++ b/src/runtime/nodes/link.rs @@ -17,20 +17,51 @@ use crate::model::Msg; use crate::runtime::node::{Node, NodeCtx}; use crate::runtime::nodes::parse_config; +/// A named link channel plus the number of nodes referencing it. +struct LinkChan { + tx: broadcast::Sender, + refs: usize, +} + /// Process-global registry of named link channels. -fn registry() -> &'static Mutex>> { - static LINKS: OnceLock>>> = OnceLock::new(); +fn registry() -> &'static Mutex> { + static LINKS: OnceLock>> = OnceLock::new(); LINKS.get_or_init(|| Mutex::new(HashMap::new())) } -/// The sender for a link name, creating it on first use. -fn channel_for(name: &str) -> broadcast::Sender { - registry() - .lock() - .unwrap() - .entry(name.to_string()) - .or_insert_with(|| broadcast::channel(64).0) - .clone() +/// Drops a link reference on drop, removing the channel when the last node +/// using the name stops — so the registry doesn't grow unbounded over deploys. +struct LinkGuard { + name: String, +} + +impl Drop for LinkGuard { + fn drop(&mut self) { + let mut reg = registry().lock().unwrap(); + if let Some(chan) = reg.get_mut(&self.name) { + chan.refs -= 1; + if chan.refs == 0 { + reg.remove(&self.name); + } + } + } +} + +/// Acquire the sender for a link name (creating it on first use) and a guard +/// that releases the reference when dropped. +fn acquire(name: &str) -> (broadcast::Sender, LinkGuard) { + let mut reg = registry().lock().unwrap(); + let chan = reg.entry(name.to_string()).or_insert_with(|| LinkChan { + tx: broadcast::channel(64).0, + refs: 0, + }); + chan.refs += 1; + ( + chan.tx.clone(), + LinkGuard { + name: name.to_string(), + }, + ) } #[derive(Debug, Default, Deserialize)] @@ -60,7 +91,8 @@ impl LinkOutNode { #[async_trait] impl Node for LinkInNode { async fn run(self: Box, ctx: NodeCtx) -> anyhow::Result<()> { - let mut rx = channel_for(&self.name).subscribe(); + let (tx, _guard) = acquire(&self.name); + let mut rx = tx.subscribe(); loop { match rx.recv().await { Ok(msg) => ctx.emit(msg).await, @@ -75,7 +107,7 @@ impl Node for LinkInNode { #[async_trait] impl Node for LinkOutNode { async fn run(self: Box, mut ctx: NodeCtx) -> anyhow::Result<()> { - let tx = channel_for(&self.name); + let (tx, _guard) = acquire(&self.name); while let Some(msg) = ctx.rx.recv().await { // Err = no link-in is listening on this name; ignore. let _ = tx.send(msg); @@ -89,10 +121,14 @@ mod tests { use super::*; use serde_json::json; + fn is_registered(name: &str) -> bool { + registry().lock().unwrap().contains_key(name) + } + #[test] fn same_name_shares_a_channel() { - let a = channel_for("link-test-share"); - let b = channel_for("link-test-share"); + let (a, _ga) = acquire("link-test-share"); + let (b, _gb) = acquire("link-test-share"); assert_eq!(a.receiver_count(), 0); let _rx = a.subscribe(); assert_eq!(b.receiver_count(), 1); // same underlying channel @@ -101,9 +137,20 @@ mod tests { #[tokio::test] async fn out_delivers_to_in() { // link-out → named channel → link-in subscriber. - let mut rx = channel_for("link-test-deliver").subscribe(); - let tx = channel_for("link-test-deliver"); + let (tx, _g) = acquire("link-test-deliver"); + let mut rx = tx.subscribe(); tx.send(Msg::new(json!("hi"))).unwrap(); assert_eq!(rx.recv().await.unwrap().payload, json!("hi")); } + + #[test] + fn channel_removed_when_last_ref_released() { + let (_tx1, g1) = acquire("link-test-refs"); + let (_tx2, g2) = acquire("link-test-refs"); + assert!(is_registered("link-test-refs")); + drop(g1); + assert!(is_registered("link-test-refs")); // still one holder + drop(g2); + assert!(!is_registered("link-test-refs")); // removed when unused + } } From 3b2963a546fe9176b73c72634b175781c582bacf Mon Sep 17 00:00:00 2001 From: ashaffah Date: Fri, 26 Jun 2026 14:54:25 +0700 Subject: [PATCH 5/5] fix(function): bound JS loop iterations (boa) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The js function language had no execution cap, so a `while (true) {}` script occupied a blocking thread forever (rhai is capped by operations, WASM by fuel — only js was unbounded). Set boa's loop-iteration limit so runaway loops error out. Add transform + runaway-loop tests. --- src/runtime/nodes/function.rs | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/runtime/nodes/function.rs b/src/runtime/nodes/function.rs index 6f18ef0..f0c266d 100644 --- a/src/runtime/nodes/function.rs +++ b/src/runtime/nodes/function.rs @@ -28,6 +28,8 @@ use crate::runtime::nodes::parse_config; /// Cap on rhai operations per message — bounds runaway / `while true` scripts. const MAX_RHAI_OPS: u64 = 10_000_000; +/// Cap on JS loop iterations per message — bounds runaway `while (true)` scripts. +const MAX_JS_LOOP_ITERATIONS: u64 = 10_000_000; /// Fuel budget for one WASM `transform` call — bounds infinite loops in the module. const WASM_FUEL: u64 = 100_000_000; /// Cap on the WASM output buffer the host reads back (length is guest-controlled). @@ -120,6 +122,9 @@ fn eval_js(script: &str, msg: Msg) -> Result { use boa_engine::{Context, Source, js_string, property::Attribute}; let mut ctx = Context::default(); + // Bound runaway loops (the JS analog of rhai's operation cap). + ctx.runtime_limits_mut() + .set_loop_iteration_limit(MAX_JS_LOOP_ITERATIONS); let payload_js = boa_engine::JsValue::from_json(&msg.payload, &mut ctx).map_err(|e| e.to_string())?; ctx.register_global_property(js_string!("payload"), payload_js, Attribute::all()) @@ -308,4 +313,25 @@ mod tests { "expected an operation-limit error, got: {err}" ); } + + #[test] + fn js_transforms_payload() { + let out = Program::Js { + script: Arc::from("payload.n = payload.n + 1;"), + } + .eval(Msg::new(json!({ "n": 1 }))) + .expect("eval"); + assert_eq!(out.payload, json!({ "n": 2 })); + } + + #[test] + fn js_runaway_loop_is_bounded() { + // Without the loop-iteration cap this would never return. + let err = Program::Js { + script: Arc::from("while (true) {}"), + } + .eval(Msg::new(json!(null))) + .unwrap_err(); + assert!(!err.is_empty(), "expected a loop-limit error"); + } }