feat(cdp): bound every CDP command with a shared V8 watchdog (#250)

A hung page (a runaway Runtime.evaluate, a synchronous DOM op) holds the
process-wide V8 lock and wedges every other session forever. The one-shot CLI has
a process-level hard deadline for this; the long-running server cannot force-exit,
so terminate just the offending isolate instead.

dispatch() arms a watchdog on the session page's isolate after taking the V8 lock
and disarms it after the handler; on overrun it terminates the isolate and clears
the termination flag so the session keeps working. OBSCURA_CDP_COMMAND_TIMEOUT_MS
tunes the bound (default 60s, 0 disables); navigation already self-bounds well
under it.

The watchdog is a single long-lived shared thread (cdp_watchdog), not a thread
per command: the V8 lock serializes commands so at most one is ever armed, and
arm/disarm are a mutex plus notify with no measurable per-command cost. Also
fixes a lost-wakeup in the per-call watchdog (spawn_watchdog) where disarming a
fast call before the watchdog thread started blocked for the full budget.

Verified: a hung Runtime.evaluate is terminated at the budget and the session
recovers; CDP integration 37/37, obstacle course 33/33; per-command latency
unchanged (watchdog on equals off within noise).

crates/obscura-js/src/cdp_watchdog.rs

@@ -0,0 +1,117 @@
+//! 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.
+//!
+//! 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.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Condvar, Mutex, OnceLock};
+use std::time::{Duration, Instant};
+
+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)>,
+    cv: Condvar,
+}
+
+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)),
+            cv: Condvar::new(),
+        });
+        let worker = s.clone();
+        std::thread::Builder::new()
+            .name("cdp-watchdog".into())
+            .spawn(move || watchdog_loop(worker))
+            .expect("spawn cdp watchdog");
+        s
+    })
+}
+
+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)
+                }
+            }
+        };
+        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.
+            None => s.cv.wait(guard).unwrap(),
+            Some(dur) => s.cv.wait_timeout(guard, dur).unwrap().0,
+        };
+    }
+}
+
+/// Handle to an armed command; pass to [`disarm`].
+pub struct Armed {
+    gen: u64,
+    fired: Arc<AtomicBool>,
+}
+
+/// Arm the shared watchdog for the current command. If the isolate is still
+/// executing `budget` later, it is terminated. O(1), no thread spawn.
+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,
+        gen,
+        fired: fired.clone(),
+    });
+    s.cv.notify_one();
+    Armed { gen, fired }
+}
+
+/// Disarm the command's watchdog. Returns true if it had already fired
+/// (terminated the isolate), in which case the caller must clear the V8
+/// termination flag before the next command runs.
+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();
+    }
+    armed.fired.load(Ordering::SeqCst)
+}

crates/obscura-js/src/lib.rs

@@ -1,6 +1,7 @@
 #[macro_use]
 extern crate html5ever;
 
+pub mod cdp_watchdog;
 pub mod module_loader;
 pub mod runtime;
 pub mod ops;

crates/obscura-js/src/runtime.rs

@@ -5,6 +5,10 @@ use std::rc::Rc;
 use deno_core::{JsRuntime, RuntimeOptions};
 use obscura_dom::DomTree;
 
+/// Re-exported so other crates (obscura-browser, obscura-cdp) can name the V8
+/// isolate handle without taking a direct dependency on deno_core.
+pub use deno_core::v8::IsolateHandle;
+
 use crate::module_loader::ObscuraModuleLoader;
 use crate::ops::{build_extension, ObscuraState};
 
@@ -25,6 +29,10 @@ pub struct ObscuraJsRuntime {
     state: Rc<RefCell<ObscuraState>>,
     object_store: HashMap<String, String>,
     object_counter: u64,
+    /// Thread-safe handle to this runtime's V8 isolate, captured at
+    /// construction. Lets a watchdog be armed from `&self` (the CDP dispatcher
+    /// only holds `&Page` on the hot path) and is stable for the isolate's life.
+    isolate_handle: IsolateHandle,
 }
 
 /// Handle to an armed V8 execution watchdog (see [`ObscuraJsRuntime::arm_watchdog`]).
@@ -36,6 +44,62 @@ pub struct WatchdogToken {
     fired: std::sync::Arc<std::sync::atomic::AtomicBool>,
 }
 
+/// Arm a V8 termination watchdog directly from an isolate handle, with no
+/// runtime borrow. The CDP dispatcher uses this to bound every command so a
+/// hung page cannot hold the process-wide V8 lock forever. Pair with
+/// [`WatchdogToken::stop`]; if `stop` returns true, clear the termination flag
+/// via [`ObscuraJsRuntime::cancel_termination`] before reusing the isolate.
+pub fn spawn_watchdog(handle: IsolateHandle, budget: std::time::Duration) -> WatchdogToken {
+    let pair = std::sync::Arc::new((std::sync::Mutex::new(false), std::sync::Condvar::new()));
+    let fired = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let pair_c = pair.clone();
+    let fired_c = fired.clone();
+    let join = std::thread::spawn(move || {
+        let (lock, cvar) = &*pair_c;
+        let mut cancelled = lock.lock().unwrap();
+        let deadline = std::time::Instant::now() + budget;
+        loop {
+            // Check first: stop() may have set this (and notified into the void)
+            // before this thread even started, which happens constantly for fast
+            // CDP commands where stop() is called right after spawn. Without this
+            // top check the lost notify means we wait the full budget before
+            // noticing, and stop()'s join() blocks for that whole time.
+            if *cancelled {
+                return;
+            }
+            let remaining = deadline.saturating_duration_since(std::time::Instant::now());
+            if remaining.is_zero() {
+                fired_c.store(true, std::sync::atomic::Ordering::SeqCst);
+                handle.terminate_execution();
+                return;
+            }
+            let (guard, _) = cvar.wait_timeout(cancelled, remaining).unwrap();
+            cancelled = guard;
+            if *cancelled {
+                return;
+            }
+        }
+    });
+    WatchdogToken { pair, join: Some(join), fired }
+}
+
+impl WatchdogToken {
+    /// Stop the watchdog. Returns true if it had already fired (terminated the
+    /// isolate). The caller must then clear the termination flag via
+    /// [`ObscuraJsRuntime::cancel_termination`] before the next eval.
+    pub fn stop(mut self) -> bool {
+        {
+            let (lock, cvar) = &*self.pair;
+            *lock.lock().unwrap() = true;
+            cvar.notify_one();
+        }
+        if let Some(j) = self.join.take() {
+            let _ = j.join();
+        }
+        self.fired.load(std::sync::atomic::Ordering::SeqCst)
+    }
+}
+
 impl ObscuraJsRuntime {
     pub fn new() -> Self {
         Self::with_base_url("about:blank")
@@ -70,11 +134,14 @@ impl ObscuraJsRuntime {
             )
             .expect("init should not fail");
 
+        let isolate_handle = runtime.v8_isolate().thread_safe_handle();
+
         ObscuraJsRuntime {
             runtime,
             state,
             object_store: HashMap::new(),
             object_counter: 0,
+            isolate_handle,
         }
     }
 
@@ -687,45 +754,14 @@ impl ObscuraJsRuntime {
     /// `budget` elapses, forcing V8 to throw an uncatchable error and hand
     /// control back. Always balance with [`Self::disarm_watchdog`].
     pub fn arm_watchdog(&mut self, budget: std::time::Duration) -> WatchdogToken {
-        let handle = self.runtime.v8_isolate().thread_safe_handle();
-        let pair = std::sync::Arc::new((std::sync::Mutex::new(false), std::sync::Condvar::new()));
-        let fired = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
-        let pair_c = pair.clone();
-        let fired_c = fired.clone();
-        let join = std::thread::spawn(move || {
-            let (lock, cvar) = &*pair_c;
-            let mut cancelled = lock.lock().unwrap();
-            let deadline = std::time::Instant::now() + budget;
-            loop {
-                let remaining = deadline.saturating_duration_since(std::time::Instant::now());
-                if remaining.is_zero() {
-                    fired_c.store(true, std::sync::atomic::Ordering::SeqCst);
-                    handle.terminate_execution();
-                    return;
-                }
-                let (guard, _) = cvar.wait_timeout(cancelled, remaining).unwrap();
-                cancelled = guard;
-                if *cancelled {
-                    return;
-                }
-            }
-        });
-        WatchdogToken { pair, join: Some(join), fired }
+        spawn_watchdog(self.runtime.v8_isolate().thread_safe_handle(), budget)
     }
 
     /// Stop a watchdog armed by [`Self::arm_watchdog`]. If it had already fired
     /// (terminated the isolate), clear V8's termination flag so the isolate is
     /// usable again, and return `true`.
-    pub fn disarm_watchdog(&mut self, mut token: WatchdogToken) -> bool {
-        {
-            let (lock, cvar) = &*token.pair;
-            *lock.lock().unwrap() = true;
-            cvar.notify_one();
-        }
-        if let Some(j) = token.join.take() {
-            let _ = j.join();
-        }
-        let fired = token.fired.load(std::sync::atomic::Ordering::SeqCst);
+    pub fn disarm_watchdog(&mut self, token: WatchdogToken) -> bool {
+        let fired = token.stop();
         if fired {
             self.runtime.v8_isolate().cancel_terminate_execution();
             tracing::warn!("V8 watchdog fired: terminated a synchronous overrun");
@@ -733,6 +769,20 @@ impl ObscuraJsRuntime {
         fired
     }
 
+    /// This runtime's V8 isolate handle (captured at construction, stable for
+    /// the isolate's life). Lets the CDP dispatcher arm a per-command watchdog
+    /// from `&self`.
+    pub fn isolate_handle(&self) -> IsolateHandle {
+        self.isolate_handle.clone()
+    }
+
+    /// Clear V8's termination flag after a watchdog armed externally (via the
+    /// isolate handle) fired, so the isolate is usable for the next command.
+    /// No-op when the isolate is not terminating.
+    pub fn cancel_termination(&mut self) {
+        self.runtime.v8_isolate().cancel_terminate_execution();
+    }
+
     /// Drive the event loop for at most `budget_ms`, bounded against BOTH async
     /// idle (tokio timeout) and synchronous hangs (V8 watchdog). A microtask
     /// storm that pins the thread is terminated ~500ms past the budget; a