fix(js): track one watchdog slot per connection

The per-command V8 watchdog kept a single global slot whose correctness relied
on the old process-wide lock (at most one command armed at a time). With
thread-per-connection several connections arm concurrently, so a second arm
overwrote the first's slot and left that command's isolate unbounded. Key the
armed slots by generation so each connection's command has its own; the worker
fires whichever have overrun and terminates each via its IsolateHandle.

Refs #430.

crates/obscura-js/src/cdp_watchdog.rs

@@ -1,16 +1,18 @@
 //! Shared per-command V8 watchdog for the CDP server.
 //!
-//! The CDP dispatcher holds a process-wide V8 lock around every V8-touching
-//! command, so at most one command's isolate is ever executing at a time. That
-//! lets a single long-lived watchdog thread bound the current command with a
-//! deadline, instead of spawning+joining a thread per command (which adds
-//! ~240us per command on the hot dispatch path). `arm` and `disarm` are just a
-//! mutex + condvar notify, in the low microseconds.
+//! One long-lived watchdog thread bounds every in-flight V8 command with a
+//! deadline, instead of spawning and joining a thread per command (which adds
+//! ~240us per command on the hot dispatch path). `arm` and `disarm` are a mutex
+//! plus a condvar notify, in the low microseconds.
 //!
-//! Correctness rests on the V8-lock serialization: between `arm` and `disarm`
-//! the caller holds the V8 lock, so no other command can arm, and the single
-//! slot is exclusively this command's.
+//! With the thread-per-connection server (issue #430) several connections can
+//! have a command armed at the same time (one isolate per connection, each on
+//! its own OS thread), so a single global slot would let one connection's arm
+//! overwrite another's and leave that command unbounded. The watchdog therefore
+//! tracks a set of armed slots keyed by a monotonic generation, fires whichever
+//! have overrun, and terminates each through its thread-safe `IsolateHandle`.
 
+use std::collections::HashMap;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Condvar, Mutex, OnceLock};
 use std::time::{Duration, Instant};
@@ -20,13 +22,12 @@ use crate::runtime::IsolateHandle;
 struct Slot {
     deadline: Instant,
     handle: IsolateHandle,
-    gen: u64,
     fired: Arc<AtomicBool>,
 }
 
 struct Shared {
-    // (current armed slot, monotonic generation counter)
-    state: Mutex<(Option<Slot>, u64)>,
+    // (armed slots keyed by generation, monotonic generation counter)
+    state: Mutex<(HashMap<u64, Slot>, u64)>,
     cv: Condvar,
 }
 
@@ -35,7 +36,7 @@ static SHARED: OnceLock<Arc<Shared>> = OnceLock::new();
 fn shared() -> &'static Arc<Shared> {
     SHARED.get_or_init(|| {
         let s = Arc::new(Shared {
-            state: Mutex::new((None, 0)),
+            state: Mutex::new((HashMap::new(), 0)),
             cv: Condvar::new(),
         });
         let worker = s.clone();
@@ -50,27 +51,31 @@ fn shared() -> &'static Arc<Shared> {
 fn watchdog_loop(s: Arc<Shared>) {
     let mut guard = s.state.lock().unwrap();
     loop {
-        let next_deadline = match &guard.0 {
-            None => None,
-            Some(slot) => {
-                let now = Instant::now();
-                if slot.deadline <= now {
-                    // Overran: terminate this isolate and clear the slot. The
-                    // dispatcher's disarm will observe `fired` and clear the V8
-                    // termination flag before the next command.
-                    slot.fired.store(true, Ordering::SeqCst);
-                    slot.handle.terminate_execution();
-                    guard.0 = None;
-                    None
-                } else {
-                    Some(slot.deadline - now)
-                }
+        let now = Instant::now();
+        // Terminate every slot that has overrun its deadline. The dispatcher's
+        // `disarm` will observe `fired` and clear the V8 termination flag before
+        // that isolate runs its next command.
+        let expired: Vec<u64> = guard
+            .0
+            .iter()
+            .filter(|(_, slot)| slot.deadline <= now)
+            .map(|(gen, _)| *gen)
+            .collect();
+        for gen in expired {
+            if let Some(slot) = guard.0.remove(&gen) {
+                slot.fired.store(true, Ordering::SeqCst);
+                slot.handle.terminate_execution();
             }
-        };
-        guard = match next_deadline {
-            // No armed command: block until arm() notifies. The worker holds the
-            // lock until it waits, so arm() (which needs the lock) cannot notify
-            // into the void; no lost wakeup.
+        }
+        // Sleep until the nearest remaining deadline, or until arm/disarm wakes
+        // us. The worker holds the lock until it waits, so a notify cannot be
+        // lost into the void.
+        let next = guard
+            .0
+            .values()
+            .map(|slot| slot.deadline.saturating_duration_since(now))
+            .min();
+        guard = match next {
             None => s.cv.wait(guard).unwrap(),
             Some(dur) => s.cv.wait_timeout(guard, dur).unwrap().0,
         };
@@ -84,19 +89,22 @@ pub struct Armed {
 }
 
 /// Arm the shared watchdog for the current command. If the isolate is still
-/// executing `budget` later, it is terminated. O(1), no thread spawn.
+/// executing `budget` later, it is terminated. O(1), no thread spawn. Safe to
+/// call concurrently from several connections: each command gets its own slot.
 pub fn arm(handle: IsolateHandle, budget: Duration) -> Armed {
     let s = shared();
     let mut guard = s.state.lock().unwrap();
     guard.1 += 1;
     let gen = guard.1;
     let fired = Arc::new(AtomicBool::new(false));
-    guard.0 = Some(Slot {
-        deadline: Instant::now() + budget,
-        handle,
+    guard.0.insert(
         gen,
-        fired: fired.clone(),
-    });
+        Slot {
+            deadline: Instant::now() + budget,
+            handle,
+            fired: fired.clone(),
+        },
+    );
     s.cv.notify_one();
     Armed { gen, fired }
 }
@@ -107,11 +115,8 @@ pub fn arm(handle: IsolateHandle, budget: Duration) -> Armed {
 pub fn disarm(armed: Armed) -> bool {
     let s = shared();
     let mut guard = s.state.lock().unwrap();
-    // Only clear the slot if it is still ours (a newer command would have a
-    // higher gen, though the V8 lock prevents that overlap in practice).
-    if guard.0.as_ref().map(|sl| sl.gen) == Some(armed.gen) {
-        guard.0 = None;
-        s.cv.notify_one();
-    }
+    guard.0.remove(&armed.gen);
+    // Wake the worker so it recomputes its sleep if we removed the nearest slot.
+    s.cv.notify_one();
     armed.fired.load(Ordering::SeqCst)
 }