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.
This commit is contained in:
Chris Tate
2026-03-14 16:29:47 -05:00
committed by GitHub
parent f962c7a4c4
commit a773863214
2 changed files with 25 additions and 8 deletions
+9 -1
View File
@@ -1903,13 +1903,19 @@ async fn handle_reload(state: &mut DaemonState) -> Result<Value, String> {
let mut rx = mgr.client.subscribe(); let mut rx = mgr.client.subscribe();
let _ = tokio::time::timeout(tokio::time::Duration::from_secs(10), async { let _ = tokio::time::timeout(tokio::time::Duration::from_secs(10), async {
while let Ok(event) = rx.recv().await { loop {
match rx.recv().await {
Ok(event) => {
if event.method == "Page.loadEventFired" if event.method == "Page.loadEventFired"
&& event.session_id.as_deref() == Some(&session_id) && event.session_id.as_deref() == Some(&session_id)
{ {
return; return;
} }
} }
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
Err(_) => break,
}
}
}) })
.await; .await;
@@ -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()), Ok(Err(_)) => return Err("Event stream closed".to_string()),
Err(_) => { Err(_) => {
return Err(format!( return Err(format!(
@@ -4203,6 +4210,7 @@ async fn handle_waitfordownload(cmd: &Value, state: &DaemonState) -> Result<Valu
return Ok(json!({ "path": path })); return Ok(json!({ "path": path }));
} }
} }
Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
Ok(Err(_)) => return Err("Event stream closed".to_string()), Ok(Err(_)) => return Err("Event stream closed".to_string()),
Err(_) => return Err("Timeout waiting for download".to_string()), Err(_) => return Err("Timeout waiting for download".to_string()),
} }
+11 -2
View File
@@ -469,11 +469,19 @@ impl BrowserManager {
let timeout = tokio::time::Duration::from_millis(self.default_timeout_ms); let timeout = tokio::time::Duration::from_millis(self.default_timeout_ms);
tokio::time::timeout(timeout, async { tokio::time::timeout(timeout, async {
while let Ok(event) = rx.recv().await { loop {
if event.method == event_name && event.session_id.as_deref() == Some(session_id) { match rx.recv().await {
Ok(event) => {
if event.method == event_name
&& event.session_id.as_deref() == Some(session_id)
{
return Ok(()); return Ok(());
} }
} }
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
}
}
Err("Event stream closed".to_string()) Err("Event stream closed".to_string())
}) })
.await .await
@@ -526,6 +534,7 @@ impl BrowserManager {
} }
} }
Ok(Ok(_)) => {} Ok(Ok(_)) => {}
Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => continue,
Ok(Err(_)) => break, Ok(Err(_)) => break,
Err(_) => { Err(_) => {
// Timeout on recv -- check if idle long enough // Timeout on recv -- check if idle long enough