From 5e91178e03298dec8e1418ed1e2ea3cc3f605971 Mon Sep 17 00:00:00 2001 From: drakezhang Date: Wed, 22 Jul 2026 12:16:56 +0800 Subject: [PATCH] Add heartbeat event and daemon browser-liveness reaper Introduce a `system.heartbeat` protocol event and use it as a liveness signal on the daemon side. - Track `last_seen` per connected browser, bumped on every inbound WS frame (tool response, event, ping/pong, or heartbeat). - Add a reaper task that drops any browser silent past a 60s window and purges its sessions, mirroring the WS disconnect cleanup. This releases half-open connections left behind when the OS resumes from sleep and kills the service worker without delivering a close frame, so the daemon no longer pins a phantom "connected" browser and can idle-exit normally. - Wire the reaper into both the foreground daemon and the test/embed entry point, aborting it on shutdown. Co-authored-by: Cursor --- crates/bsk-cli/src/daemon/browsers.rs | 83 ++++++++++++++++++- crates/bsk-cli/src/daemon/mod.rs | 2 + crates/bsk-cli/src/daemon/start.rs | 49 +++++++++++ crates/bsk-cli/src/daemon/state.rs | 5 ++ crates/bsk-cli/src/daemon/ws.rs | 14 +++- crates/bsk-cli/tests/browser_list_ordering.rs | 1 + crates/bsk-cli/tests/handshake_compat.rs | 1 + crates/bsk-protocol/src/frame.rs | 8 ++ 8 files changed, 161 insertions(+), 2 deletions(-) diff --git a/crates/bsk-cli/src/daemon/browsers.rs b/crates/bsk-cli/src/daemon/browsers.rs index c25f3de..6a6b296 100644 --- a/crates/bsk-cli/src/daemon/browsers.rs +++ b/crates/bsk-cli/src/daemon/browsers.rs @@ -22,10 +22,29 @@ use bsk_protocol::system::BrowserStatusEntry; /// seconds to wake, discover the WS port, and finish `system.handshake`. /// `session.start` polls the registry for up to this long before /// returning `no_browser_connected`. -pub const EXTENSION_CONNECT_WAIT: Duration = Duration::from_secs(5); +/// +/// Sized to comfortably cover one MV3 keepalive-alarm period (30s, the +/// smallest Chrome allows) plus handshake slack: after a long idle the +/// service worker is only revived on its next alarm tick, so a shorter +/// window would make the first command after idle fail with +/// `no_browser_connected` even though the extension is about to +/// reconnect. Callers that want a snapshot (`bsk status`/`browsers`) +/// pass their own, shorter wait instead of this constant. +pub const EXTENSION_CONNECT_WAIT: Duration = Duration::from_secs(35); const EXTENSION_CONNECT_POLL: Duration = Duration::from_millis(50); +/// A connected browser is considered dead if no frame (including the +/// ~20s `system.heartbeat`) has arrived within this window. Set to three +/// missed heartbeats so a briefly-busy service worker (GC, heavy CDP) is +/// not reaped on a single late beat. Used by the liveness reaper to drop +/// half-open connections that never delivered a socket close — e.g. after +/// the OS resumed from sleep and killed the worker. +pub const BROWSER_LIVENESS_TIMEOUT: Duration = Duration::from_secs(60); + +/// How often the liveness reaper scans the registry for stale browsers. +pub const BROWSER_LIVENESS_TICK: Duration = Duration::from_secs(15); + /// Process-wide monotonic counter for [`BrowserClient::generation`]. Used /// by the reconnect-race guard: when an old WS task tears down it only /// removes itself from the registry if the registry's current @@ -125,9 +144,29 @@ pub struct BrowserClient { /// `true` when the peer's `protocol_version` differs from ours /// (same major, minor drift). Set during WS handshake (M10.4). pub version_skew: bool, + /// Monotonic timestamp of the most recent inbound frame from this + /// browser (any response/event, including `system.heartbeat`). + /// Bumped by [`BrowserClient::touch`] from the WS read loop and read + /// by the liveness reaper to detect a silently-dead connection. + pub last_seen: Mutex, } impl BrowserClient { + /// Record that a frame just arrived from this browser. + pub fn touch(&self) { + if let Ok(mut last) = self.last_seen.lock() { + *last = Instant::now(); + } + } + + /// How long since the last inbound frame from this browser. + pub fn idle_for(&self) -> Duration { + match self.last_seen.lock() { + Ok(last) => last.elapsed(), + Err(poisoned) => poisoned.into_inner().elapsed(), + } + } + pub fn status_entry(&self, session_count: u32) -> BrowserStatusEntry { BrowserStatusEntry { instance_id: self.id.0.clone(), @@ -206,6 +245,20 @@ impl BrowserRegistry { guard.values().cloned().collect() } + /// Return every currently-registered browser whose last inbound + /// frame is older than `threshold`. The caller is responsible for + /// dropping them (via [`Self::remove_if_generation_matches`]) and + /// purging their sessions — done outside the lock so session + /// teardown never blocks the registry. + pub fn stale_browsers(&self, threshold: Duration) -> Vec> { + let guard = self.inner.lock().expect("browser registry poisoned"); + guard + .values() + .filter(|c| c.idle_for() >= threshold) + .cloned() + .collect() + } + pub fn len(&self) -> usize { self.inner.lock().expect("browser registry poisoned").len() } @@ -348,6 +401,7 @@ mod tests { generation: next_browser_generation(), connected_at_ms: 0, version_skew: false, + last_seen: Mutex::new(Instant::now()), }) } @@ -543,6 +597,33 @@ mod tests { assert_eq!(reg.len(), 1); } + #[test] + fn touch_resets_idle_for() { + let client = fake_client("a", ""); + // Force the last_seen well into the past, then touch. + *client.last_seen.lock().unwrap() = Instant::now() - Duration::from_secs(120); + assert!(client.idle_for() >= Duration::from_secs(120)); + client.touch(); + assert!(client.idle_for() < Duration::from_secs(1)); + } + + #[test] + fn stale_browsers_reports_only_silent_connections() { + let reg = BrowserRegistry::new(); + let fresh = fake_client("fresh", ""); + let stale = fake_client("stale", ""); + *stale.last_seen.lock().unwrap() = Instant::now() - Duration::from_secs(90); + reg.insert(fresh); + reg.insert(stale); + + let stale_ids: Vec = reg + .stale_browsers(Duration::from_secs(60)) + .into_iter() + .map(|c| c.id.0.clone()) + .collect(); + assert_eq!(stale_ids, vec!["stale".to_string()]); + } + #[tokio::test] async fn wait_for_any_connected_skips_when_already_populated() { let reg = BrowserRegistry::new(); diff --git a/crates/bsk-cli/src/daemon/mod.rs b/crates/bsk-cli/src/daemon/mod.rs index 99ee7e6..c639f73 100644 --- a/crates/bsk-cli/src/daemon/mod.rs +++ b/crates/bsk-cli/src/daemon/mod.rs @@ -45,6 +45,7 @@ pub async fn run( None => None, }; let session_idle_task = start::spawn_session_idle_reaper(Arc::clone(&state)); + let browser_liveness_task = start::spawn_browser_liveness_reaper(Arc::clone(&state)); // Ensure WS/IPC accept loops have polled `shutdown.notified()` before any // caller can invoke `DaemonHandle::shutdown()` — `Notify::notify_waiters()` // drops wakeups when nothing is registered yet, which otherwise hangs @@ -55,5 +56,6 @@ pub async fn run( ws_handle, ipc_handle, session_idle_task, + browser_liveness_task, )) } diff --git a/crates/bsk-cli/src/daemon/start.rs b/crates/bsk-cli/src/daemon/start.rs index 1d72e63..0e48fc6 100644 --- a/crates/bsk-cli/src/daemon/start.rs +++ b/crates/bsk-cli/src/daemon/start.rs @@ -200,6 +200,7 @@ pub fn run_foreground(cfg: DaemonConfig) -> Result<()> { let state = Arc::new(DaemonState::new(cfg.clone())); let session_idle_task = spawn_session_idle_reaper(Arc::clone(&state)); + let browser_liveness_task = spawn_browser_liveness_reaper(Arc::clone(&state)); let ws_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), cfg.ws_port); let ws_handle = ws::WsServer::new(Arc::clone(&state)) .bind(ws_addr) @@ -361,6 +362,8 @@ pub fn run_foreground(cfg: DaemonConfig) -> Result<()> { let _ = ipc_task.await; session_idle_task.abort(); let _ = session_idle_task.await; + browser_liveness_task.abort(); + let _ = browser_liveness_task.await; ws_handle.shutdown.notify_waiters(); let _ = ws_handle.task.await; @@ -373,6 +376,52 @@ pub fn run_foreground(cfg: DaemonConfig) -> Result<()> { Ok(()) } +/// Spawn the browser-liveness reaper shared by the production foreground +/// daemon and the test/embed entry point. +/// +/// A normal disconnect closes the WebSocket, which the per-connection +/// read loop observes and cleans up. But if the OS resumed from sleep +/// and killed the MV3 service worker, the socket can be left half-open +/// with no close frame ever delivered. Without the ~20s `system.heartbeat` +/// arriving, such a connection would otherwise sit in the registry +/// forever — pinning a phantom "connected" browser and preventing the +/// daemon from ever idle-exiting. This reaper drops any browser that has +/// gone silent past [`BROWSER_LIVENESS_TIMEOUT`] and purges its sessions, +/// mirroring the WS disconnect cleanup. +pub(crate) fn spawn_browser_liveness_reaper( + state: Arc, +) -> tokio::task::JoinHandle<()> { + use crate::daemon::browsers::{BROWSER_LIVENESS_TICK, BROWSER_LIVENESS_TIMEOUT}; + tokio::spawn(async move { + let mut ticker = tokio::time::interval(BROWSER_LIVENESS_TICK); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // The first tick is immediate; skip it so a freshly connected + // browser always gets at least one full window before a scan. + ticker.tick().await; + loop { + ticker.tick().await; + for client in state.browsers.stale_browsers(BROWSER_LIVENESS_TIMEOUT) { + if state + .browsers + .remove_if_generation_matches(&client.id, client.generation) + .is_some() + { + warn!( + id = %client.id, + idle_secs = client.idle_for().as_secs(), + "reaping unresponsive browser (no heartbeat within liveness window)" + ); + for s in state.sessions.purge_browser(&client.id) { + state.tool_queues.remove(&s.id); + state.session_interrupts.drop_session(&s.id); + debug!(session = %s.id, "purged session on browser liveness timeout"); + } + } + } + } + }) +} + /// Spawn the cooperative session-idle reaper shared by the production /// foreground daemon and the test/embed daemon entry point. pub(crate) fn spawn_session_idle_reaper(state: Arc) -> tokio::task::JoinHandle<()> { diff --git a/crates/bsk-cli/src/daemon/state.rs b/crates/bsk-cli/src/daemon/state.rs index 203b3ac..d6632da 100644 --- a/crates/bsk-cli/src/daemon/state.rs +++ b/crates/bsk-cli/src/daemon/state.rs @@ -82,6 +82,7 @@ pub struct DaemonHandle { ws: WsHandle, ipc: Option, session_idle_task: JoinHandle<()>, + browser_liveness_task: JoinHandle<()>, } impl DaemonHandle { @@ -90,12 +91,14 @@ impl DaemonHandle { ws: WsHandle, ipc: Option, session_idle_task: JoinHandle<()>, + browser_liveness_task: JoinHandle<()>, ) -> Self { Self { state, ws, ipc, session_idle_task, + browser_liveness_task, } } @@ -116,6 +119,8 @@ impl DaemonHandle { pub async fn shutdown(self) { self.session_idle_task.abort(); let _ = await_join(self.session_idle_task).await; + self.browser_liveness_task.abort(); + let _ = await_join(self.browser_liveness_task).await; self.ws.shutdown.notify_waiters(); let _ = await_join(self.ws.task).await; if let Some(ipc) = self.ipc { diff --git a/crates/bsk-cli/src/daemon/ws.rs b/crates/bsk-cli/src/daemon/ws.rs index 4073771..0932fa5 100644 --- a/crates/bsk-cli/src/daemon/ws.rs +++ b/crates/bsk-cli/src/daemon/ws.rs @@ -304,6 +304,7 @@ async fn drive_connection( generation, connected_at_ms, version_skew, + last_seen: std::sync::Mutex::new(std::time::Instant::now()), }); state.browsers.insert(Arc::clone(&client)); info!( @@ -352,15 +353,21 @@ async fn drive_connection( msg = reader.next() => { match msg { Some(Ok(Message::Text(t))) => { + // Any inbound frame — tool response, event, or the + // ~20s heartbeat — counts as proof of life. + pump_browser.touch(); handle_inbound_text(&pump_state, &pump_browser, &t).await; } Some(Ok(Message::Binary(_))) => { warn!("ignoring binary message"); } Some(Ok(Message::Ping(p))) => { + pump_browser.touch(); let _ = writer.send(Message::Pong(p)).await; } - Some(Ok(Message::Pong(_))) => {} + Some(Ok(Message::Pong(_))) => { + pump_browser.touch(); + } Some(Ok(Message::Frame(_))) => {} Some(Ok(Message::Close(_))) | None => break, Some(Err(err)) => { @@ -416,6 +423,11 @@ async fn handle_inbound_text(state: &Arc, client: &Arc match ev.event { + bsk_protocol::EventKind::SystemHeartbeat => { + // Liveness only — `touch()` already ran for this frame in + // the read loop, so there is nothing else to do. + debug!(id = %client.id, "heartbeat"); + } bsk_protocol::EventKind::SessionWindowClosed => { handle_session_window_closed(state, &client.id, &ev.payload); } diff --git a/crates/bsk-cli/tests/browser_list_ordering.rs b/crates/bsk-cli/tests/browser_list_ordering.rs index f19f320..b62a597 100644 --- a/crates/bsk-cli/tests/browser_list_ordering.rs +++ b/crates/bsk-cli/tests/browser_list_ordering.rs @@ -51,6 +51,7 @@ fn fake_client(id: &str, label: &str, connected_at_ms: i64) -> Arc