mirror of
https://github.com/Tencent/BrowserSkill.git
synced 2026-09-10 13:01:23 +08:00
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 <cursoragent@cursor.com>
This commit is contained in:
@@ -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<Instant>,
|
||||
}
|
||||
|
||||
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<std::sync::Arc<BrowserClient>> {
|
||||
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<String> = 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();
|
||||
|
||||
@@ -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,
|
||||
))
|
||||
}
|
||||
|
||||
@@ -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<DaemonState>,
|
||||
) -> 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<DaemonState>) -> tokio::task::JoinHandle<()> {
|
||||
|
||||
@@ -82,6 +82,7 @@ pub struct DaemonHandle {
|
||||
ws: WsHandle,
|
||||
ipc: Option<IpcHandle>,
|
||||
session_idle_task: JoinHandle<()>,
|
||||
browser_liveness_task: JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl DaemonHandle {
|
||||
@@ -90,12 +91,14 @@ impl DaemonHandle {
|
||||
ws: WsHandle,
|
||||
ipc: Option<IpcHandle>,
|
||||
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 {
|
||||
|
||||
@@ -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<DaemonState>, client: &Arc<BrowserClien
|
||||
}
|
||||
}
|
||||
Frame::Event(ev) => 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);
|
||||
}
|
||||
|
||||
@@ -51,6 +51,7 @@ fn fake_client(id: &str, label: &str, connected_at_ms: i64) -> Arc<BrowserClient
|
||||
generation: next_browser_generation(),
|
||||
connected_at_ms,
|
||||
version_skew: false,
|
||||
last_seen: Mutex::new(std::time::Instant::now()),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -242,6 +242,7 @@ async fn status_surfaces_version_skew_for_skewed_browser() {
|
||||
generation: bsk::daemon::browsers::next_browser_generation(),
|
||||
connected_at_ms: 0,
|
||||
version_skew: true,
|
||||
last_seen: Mutex::new(std::time::Instant::now()),
|
||||
});
|
||||
state.browsers.insert(client);
|
||||
|
||||
|
||||
@@ -40,6 +40,14 @@ pub struct EventFrame {
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum EventKind {
|
||||
/// Application-level keepalive emitted by the extension roughly
|
||||
/// every 20s while the WS link is up. Two purposes: (1) the
|
||||
/// send/receive activity resets the MV3 service-worker idle timer
|
||||
/// (Chrome 116+), keeping the worker — and therefore the socket —
|
||||
/// alive during use; (2) the daemon treats it as a liveness signal
|
||||
/// so a silently-dead browser can be reaped. Carries no payload.
|
||||
#[serde(rename = "system.heartbeat")]
|
||||
SystemHeartbeat,
|
||||
#[serde(rename = "session.activity")]
|
||||
SessionActivity,
|
||||
#[serde(rename = "session.window_closed")]
|
||||
|
||||
Reference in New Issue
Block a user