diff --git a/crates/obscura-cdp/tests/concurrent_connections_heavy_page.rs b/crates/obscura-cdp/tests/concurrent_connections_heavy_page.rs new file mode 100644 index 0000000..e2e51a3 --- /dev/null +++ b/crates/obscura-cdp/tests/concurrent_connections_heavy_page.rs @@ -0,0 +1,245 @@ +//! Issue #430 regression: concurrent CDP connections driving a subresource-heavy +//! page must not abort the process. +//! +//! Why this and not `concurrent_navigations` (data: URLs): the #430 abort needs +//! a page whose event loop is still busy when the navigation settle loop's +//! per-tick `tokio::time::timeout` around `run_event_loop` fires. A cancelled +//! `run_event_loop` future left an isolate entered on the shared LocalSet +//! thread; a second connection's `execute_script` then tripped V8's +//! `heap->isolate() == Isolate::TryGetCurrent()` check and `abort(3)`'d. `data:` +//! URLs settle instantly (no subresources, no busy loop), so they never drive +//! it. This test serves a local page with SLOW subresources plus a `setInterval` +//! so the settle loop keeps pumping, then drives it from several independent +//! connections at once. +//! +//! On the single-LocalSet server this aborted deterministically. The +//! thread-per-connection server confines each connection's isolates to their own +//! OS thread, so the abort cannot happen (V8's check is per-thread) and all +//! clients complete. +//! +//! Run with `cargo test -p obscura-cdp --test concurrent_connections_heavy_page +//! -- --nocapture`. + +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio_tungstenite::{connect_async, tungstenite::Message}; + +const CLIENTS: u64 = 4; + +async fn pick_port() -> u16 { + let l = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = l.local_addr().unwrap().port(); + drop(l); + port +} + +/// Minimal HTTP fixture: a root page with two subresources that respond after a +/// delay (keeping the navigation settle loop pumping) and an inline +/// `setInterval` that keeps the event loop non-idle. Serves every connection it +/// accepts until the test ends. +async fn serve_heavy_fixture(listener: TcpListener) { + let body = "\ + \ +

heavy

\ + \ + "; + loop { + let (mut sock, _) = match listener.accept().await { + Ok(s) => s, + Err(_) => return, + }; + tokio::task::spawn_local(async move { + let mut buf = [0u8; 2048]; + let n = sock.read(&mut buf).await.unwrap_or(0); + if n == 0 { + return; + } + let req = String::from_utf8_lossy(&buf[..n]); + let path = req + .split_whitespace() + .nth(1) + .unwrap_or("/") + .to_string(); + + // Slow subresources: respond after a delay so `active_requests` stays + // > 0 through several 10ms settle ticks, which is what forces the + // per-tick timeout to cancel `run_event_loop` mid-poll on the old + // single-thread server. + let (ctype, payload): (&str, Vec) = if path == "/slow.js" { + tokio::time::sleep(Duration::from_millis(120)).await; + ("application/javascript", b"void 0;".to_vec()) + } else if path == "/slow.png" { + tokio::time::sleep(Duration::from_millis(120)).await; + // 1x1 transparent PNG. + ( + "image/png", + vec![ + 0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0x00, 0x00, 0x00, 0x0d, + 0x49, 0x48, 0x44, 0x52, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, + 0x08, 0x06, 0x00, 0x00, 0x00, 0x1f, 0x15, 0xc4, 0x89, 0x00, 0x00, 0x00, + 0x0a, 0x49, 0x44, 0x41, 0x54, 0x78, 0x9c, 0x63, 0x00, 0x01, 0x00, 0x00, + 0x05, 0x00, 0x01, 0x0d, 0x0a, 0x2d, 0xb4, 0x00, 0x00, 0x00, 0x00, 0x49, + 0x45, 0x4e, 0x44, 0xae, 0x42, 0x60, 0x82, + ], + ) + } else { + ("text/html", body.as_bytes().to_vec()) + }; + + let header = format!( + "HTTP/1.1 200 OK\r\nContent-Type: {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + ctype, + payload.len() + ); + let _ = sock.write_all(header.as_bytes()).await; + let _ = sock.write_all(&payload).await; + let _ = sock.flush().await; + }); + } +} + +/// One client: open its own CDP connection, create a target at the heavy page, +/// navigate, then repeatedly `Runtime.evaluate` an awaited promise. Returns Err +/// on protocol failure; a V8 abort would take the whole process down instead. +async fn one_client(ws_port: u16, page_url: String, id_base: u64) -> Result<(), String> { + let url = format!("ws://127.0.0.1:{}/devtools/browser", ws_port); + let (mut ws, _) = connect_async(&url).await.map_err(|e| e.to_string())?; + + let create = json!({ + "id": id_base, + "method": "Target.createTarget", + "params": {"url": page_url}, + }); + ws.send(Message::Text(create.to_string().into())) + .await + .map_err(|e| e.to_string())?; + + let mut session_id: Option = None; + let deadline = tokio::time::Instant::now() + Duration::from_secs(20); + while session_id.is_none() { + if tokio::time::Instant::now() >= deadline { + return Err("timeout waiting for sessionId".to_string()); + } + let msg = tokio::time::timeout(Duration::from_secs(10), ws.next()) + .await + .map_err(|_| "timeout waiting for createTarget".to_string())? + .ok_or("ws closed")? + .map_err(|e| e.to_string())?; + let text = match msg { + Message::Text(t) => t.to_string(), + _ => continue, + }; + let v: Value = serde_json::from_str(&text).map_err(|e| e.to_string())?; + if let Some(s) = v + .get("params") + .and_then(|p| p.get("sessionId")) + .and_then(|s| s.as_str()) + { + session_id = Some(s.to_string()); + } + } + let sid = session_id.unwrap(); + + // A few awaited evaluations: each drives resolve_promises_until while the + // page's setInterval keeps the loop busy — the settle/eval cancellation path. + for k in 0..3u64 { + let eval = json!({ + "id": id_base + 100 + k, + "method": "Runtime.evaluate", + "sessionId": sid, + "params": { + "expression": "new Promise(r => setTimeout(() => r(1 + 1), 40))", + "awaitPromise": true, + "returnByValue": true, + }, + }); + ws.send(Message::Text(eval.to_string().into())) + .await + .map_err(|e| e.to_string())?; + + let d = tokio::time::Instant::now() + Duration::from_secs(15); + loop { + if tokio::time::Instant::now() >= d { + return Err(format!("timeout waiting for evaluate {}", k)); + } + let remaining = d - tokio::time::Instant::now(); + let msg = tokio::time::timeout(remaining, ws.next()) + .await + .map_err(|_| "timeout".to_string())? + .ok_or("ws closed mid-evaluate")? + .map_err(|e| e.to_string())?; + let text = match msg { + Message::Text(t) => t.to_string(), + _ => continue, + }; + let v: Value = serde_json::from_str(&text).map_err(|e| e.to_string())?; + if v.get("id").and_then(|x| x.as_u64()) == Some(id_base + 100 + k) { + break; + } + } + } + + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn concurrent_connections_heavy_page_do_not_abort_v8() { + let ws_port = pick_port().await; + // Bind the fixture listener up front so we can hand its port to the clients. + let fixture = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let fixture_port = fixture.local_addr().unwrap().port(); + let page_url = format!("http://127.0.0.1:{}/", fixture_port); + + let local = tokio::task::LocalSet::new(); + local + .run_until(async { + // Fixture HTTP server. + tokio::task::spawn_local(serve_heavy_fixture(fixture)); + + // CDP server. allow_private_network so it may fetch 127.0.0.1. + tokio::task::spawn_local(async move { + let _ = obscura_cdp::server::start_with_full_serve_options( + ws_port, + "127.0.0.1", + None, + false, + None, + false, + None, + true, + ) + .await; + }); + tokio::time::sleep(Duration::from_millis(200)).await; + + let mut handles = Vec::new(); + for i in 0..CLIENTS { + let url = page_url.clone(); + let id_base = (i + 1) * 1000; + handles.push(tokio::task::spawn_local(async move { + one_client(ws_port, url, id_base).await + })); + } + + let mut ok = 0usize; + let mut errors = Vec::new(); + for (i, h) in handles.into_iter().enumerate() { + match h.await { + Ok(Ok(())) => ok += 1, + Ok(Err(e)) => errors.push(format!("client {}: {}", i, e)), + Err(e) => errors.push(format!("client {} join: {}", i, e)), + } + } + assert!( + errors.is_empty(), + "clients failed (a V8 abort would instead kill the process): {:#?}", + errors + ); + assert_eq!(ok as u64, CLIENTS, "every concurrent client must complete"); + }) + .await; +}