From a773863214e1e5effc704c225648a6db46aff133 Mon Sep 17 00:00:00 2001
From: Chris Tate <chris@ctate.dev>
Date: Sat, 14 Mar 2026 16:29:47 -0500
Subject: [PATCH] fix: handle broadcast channel lag instead of treating it as
 stream closure (#797)

The CDP event broadcast channel (capacity 256) can overflow on slow CI
runners when Chrome emits many events during navigation. Previously,
RecvError::Lagged was treated the same as RecvError::Closed, causing
spurious "Event stream closed" errors even though Chrome was still
running. Now all 5 event-receiving loops correctly continue on Lagged
instead of breaking.
---
 cli/src/native/actions.rs | 18 +++++++++++++-----
 cli/src/native/browser.rs | 15 ++++++++++++---
 2 files changed, 25 insertions(+), 8 deletions(-)

diff --git a/cli/src/native/actions.rs b/cli/src/native/actions.rs
index 5ba4e19..2887853 100644
--- a/cli/src/native/actions.rs
+++ b/cli/src/native/actions.rs
@@ -1903,11 +1903,17 @@ async fn handle_reload(state: &mut DaemonState) -> Result<Value, String> {
 
     let mut rx = mgr.client.subscribe();
     let _ = tokio::time::timeout(tokio::time::Duration::from_secs(10), async {
-        while let Ok(event) = rx.recv().await {
-            if event.method == "Page.loadEventFired"
-                && event.session_id.as_deref() == Some(&session_id)
-            {
-                return;
+        loop {
+            match rx.recv().await {
+                Ok(event) => {
+                    if event.method == "Page.loadEventFired"
+                        && event.session_id.as_deref() == Some(&session_id)
+                    {
+                        return;
+                    }
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                Err(_) => break,
             }
         }
     })
@@ -4165,6 +4171,7 @@ async fn handle_responsebody(cmd: &Value, state: &DaemonState) -> Result<Value,
                     }
                 }
             }
+            Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
             Ok(Err(_)) => return Err("Event stream closed".to_string()),
             Err(_) => {
                 return Err(format!(
@@ -4203,6 +4210,7 @@ async fn handle_waitfordownload(cmd: &Value, state: &DaemonState) -> Result<Valu
                     return Ok(json!({ "path": path }));
                 }
             }
+            Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
             Ok(Err(_)) => return Err("Event stream closed".to_string()),
             Err(_) => return Err("Timeout waiting for download".to_string()),
         }
diff --git a/cli/src/native/browser.rs b/cli/src/native/browser.rs
index 7a84177..c41bc37 100644
--- a/cli/src/native/browser.rs
+++ b/cli/src/native/browser.rs
@@ -469,9 +469,17 @@ impl BrowserManager {
         let timeout = tokio::time::Duration::from_millis(self.default_timeout_ms);
 
         tokio::time::timeout(timeout, async {
-            while let Ok(event) = rx.recv().await {
-                if event.method == event_name && event.session_id.as_deref() == Some(session_id) {
-                    return Ok(());
+            loop {
+                match rx.recv().await {
+                    Ok(event) => {
+                        if event.method == event_name
+                            && event.session_id.as_deref() == Some(session_id)
+                        {
+                            return Ok(());
+                        }
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
+                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                 }
             }
             Err("Event stream closed".to_string())
@@ -526,6 +534,7 @@ impl BrowserManager {
                         }
                     }
                     Ok(Ok(_)) => {}
+                    Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
                     Ok(Err(_)) => break,
                     Err(_) => {
                         // Timeout on recv -- check if idle long enough