Fix task cancellation, session token expiry, and resource management

src/auth.rs

@@ -5,7 +5,7 @@
 //! The token is sent as `Authorization: Bearer <token>`, 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<Instant>,
 }
 
-/// 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<HashSet<String>>,
+    sessions: Mutex<HashMap<String, Instant>>,
     login_guard: Mutex<LoginGuard>,
 }
 
@@ -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();

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<Instant> = None;
+        // In-flight delay tasks; aborted when the node stops (the Vec drops).
+        let mut pending: Vec<AbortOnDrop> = 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(())

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<Msg, String> {
     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");
+    }
 }

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<Msg>,
+    refs: usize,
+}
+
 /// Process-global registry of named link channels.
-fn registry() -> &'static Mutex<HashMap<String, broadcast::Sender<Msg>>> {
-    static LINKS: OnceLock<Mutex<HashMap<String, broadcast::Sender<Msg>>>> = OnceLock::new();
+fn registry() -> &'static Mutex<HashMap<String, LinkChan>> {
+    static LINKS: OnceLock<Mutex<HashMap<String, LinkChan>>> = 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<Msg> {
-    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<Msg>, 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<Self>, 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<Self>, 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
+    }
 }

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<Arc<Shared>, 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<dyn MqttClient> = Arc::from(client_box);
     let subs: Arc<Mutex<Vec<Subscriber>>> = 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)
 }