From 85739d3f6ab76e16ce71477ed5844f6bcaee80e0 Mon Sep 17 00:00:00 2001 From: Timothy Olaleke Date: Sat, 9 May 2026 18:26:38 +0200 Subject: [PATCH] fix(#19): serialize V8 work across pages with process-wide async lock (#36) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(#19): serialize V8 work across pages with process-wide async lock Concurrent CDP clients (≥5) trip V8's `heap->isolate() == Isolate::TryGetCurrent()` invariant and abort the process via V8_Fatal — no Rust panic, just abort(3). Each Page owns its own JsRuntime/Isolate and they all run on a single OS thread via LocalSet+spawn_local; when two pages' V8-touching futures interleave across an .await, V8 trips the check. This adds a process-wide tokio::sync::Mutex (obscura_js::v8_lock) that is acquired: - At the top of obscura_cdp::dispatch::dispatch (covers every CDP handler that may call into a JsRuntime). - In server.rs's process_with_interception spawn_local that runs page.navigate_with_wait + execute_preload_script (this path bypasses dispatch). The guard keeps each V8-touching block contiguous on the thread: V8 fully exits one Isolate before another page is allowed in. Trade-off: non-V8 handlers (Browser.*, Storage.*, Emulation.*) also serialize, but those are cheap and the safety win dominates. The properly concurrent fix is to pin each JsRuntime to its own OS thread and message-pass; tracked as the larger #19 follow-up. Adds an opt-in smoke test (`--ignored`) that boots the CDP server and runs 5 parallel clients through createTarget+navigate. Note: the deterministic abort needs subresource-fetch yields (page.rs:285 join_all) which data: URLs skip — full repro requires JS-heavy real traffic (issue author used scrapegraphai.com). * fix(#19): suspend other Isolates and defer foreign CDP during interception nav Address maintainer feedback on PR #36 (comment 4328238087): the V8 fatal still reproduces with `Fetch.enable patterns:["*"]` + 5 concurrent `Page.navigate`s. The `obscura_js::v8_lock` mutex was insufficient because V8's actual invariant is "only one Isolate may be entered per OS thread", and the mutex can't enforce that across `.await` points inside V8 ops. Two distinct paths violated the invariant: 1. Foreign CDP messages dispatched while a nav task was running. `process_with_interception`'s `select!` routed every non-Fetch message through `process_cdp_message → dispatch → page::handle::*`. `get_session_page_mut` then `suspend_js`'es other pages (drops their `JsRuntime` → `Isolate::Exit`) while the spawned nav task is parked at `op_fetch_url`'s `resolve_rx.await` with its Isolate still entered. Drop trips `heap->isolate() == Isolate::TryGetCurrent()`. 2. Subsequent navs through `process_with_interception` did not suspend prior pages. After nav-1 completes (page-1's JsRuntime stays alive in `ctx.pages`), nav-2's `init_js` constructs Isolate-2 while Isolate-1 is still current. The next V8 scope unwind aborts via `Context::Exit`. Fix (3 changes in `crates/obscura-cdp/src/server.rs`): - `cdp_processor`: add a `VecDeque deferred` drained before each new `rx.recv()`, so deferred messages are processed with no nav task in flight. - `process_with_interception` `select!`: only handle `NewConnection` and `Fetch.{continue,fulfill,fail}Request` inline (safe — no V8 enter on this side); push everything else into `deferred`. - `process_with_interception` entry: explicitly `suspend_js()` every other page in `ctx.pages` before spawning the nav task, mirroring the invariant the regular dispatch path already enforces via `get_session_page_mut`. Verification: - New test `crates/obscura-cdp/tests/concurrent_navigations_with_fetch.rs` is the maintainer's exact repro (5 clients, `Fetch.enable patterns:["*"]`, concurrent `Page.navigate` to https://example.com, auto-reply `Fetch.continueRequest`). Aborts on PR #36 head, passes after fix (5/5 stable runs, ~1.2s each). - Existing data-URL smoke test (`concurrent_navigations`) still passes. - Full workspace test suite (90 tests) still passes. Run with: cargo test -p obscura-cdp --test concurrent_navigations_with_fetch \ -- --nocapture --ignored Honest caveat. This is still the lock/serialize approach — it converts the abort into latency. The properly concurrent fix (Option 2 in issue #19: pin each `JsRuntime` to its own OS thread and message-pass) remains the right long-term direction. * fix(cdp): bound deferred message queue to prevent OOM during stalled navigation PR #36 comment 4341743194: process_with_interception deferred foreign CDP messages into an unbounded VecDeque while a navigation task was in flight. A stalled navigation (e.g. waiting for Fetch interception resolution) could accumulate arbitrary queued messages and OOM the process. Add MAX_DEFERRED_MESSAGES = 256. When the cap is reached, return an explicit CDP error (-32000) to the client instead of silently dropping or growing without bound. All 90 workspace tests pass. --- .gitignore | 2 + crates/obscura-cdp/Cargo.toml | 4 + crates/obscura-cdp/src/dispatch.rs | 21 ++ crates/obscura-cdp/src/server.rs | 109 ++++++++- .../tests/concurrent_navigations.rs | 145 ++++++++++++ .../concurrent_navigations_with_fetch.rs | 215 ++++++++++++++++++ crates/obscura-js/src/lib.rs | 1 + crates/obscura-js/src/v8_lock.rs | 27 +++ 8 files changed, 521 insertions(+), 3 deletions(-) create mode 100644 .gitignore create mode 100644 crates/obscura-cdp/tests/concurrent_navigations.rs create mode 100644 crates/obscura-cdp/tests/concurrent_navigations_with_fetch.rs create mode 100644 crates/obscura-js/src/v8_lock.rs diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e08c794 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +target/ +.DS_Store diff --git a/crates/obscura-cdp/Cargo.toml b/crates/obscura-cdp/Cargo.toml index eec400d..ff6d1cb 100644 --- a/crates/obscura-cdp/Cargo.toml +++ b/crates/obscura-cdp/Cargo.toml @@ -18,3 +18,7 @@ thiserror = { workspace = true } anyhow = { workspace = true } futures-util = "0.3" obscura-net = { workspace = true } + +[dev-dependencies] +tokio-tungstenite = { workspace = true } +futures-util = "0.3" diff --git a/crates/obscura-cdp/src/dispatch.rs b/crates/obscura-cdp/src/dispatch.rs index 5ccd0d4..a8f7ab4 100644 --- a/crates/obscura-cdp/src/dispatch.rs +++ b/crates/obscura-cdp/src/dispatch.rs @@ -121,6 +121,27 @@ impl CdpContext { } pub async fn dispatch(req: &CdpRequest, ctx: &mut CdpContext) -> CdpResponse { + // Issue #19: V8 fatal abort under concurrent CDP work. + // + // Every CDP handler below may end up calling into a per-Page `JsRuntime` + // (each owning its own V8 Isolate). All of them run on a single OS + // thread (current_thread tokio + LocalSet). When two pages' V8-touching + // futures interleave across an `.await` (which `process_with_interception` + // can trigger by handling new CDP messages while a navigation task is in + // flight on `spawn_local`), V8 trips the + // `heap->isolate() == Isolate::TryGetCurrent()` invariant and aborts the + // process via `V8_Fatal` — no Rust panic, just `abort(3)`. + // + // Holding the process-wide V8 lock around the entire dispatch keeps each + // handler contiguous on the thread: V8 fully exits one Isolate before + // the next page is allowed in. This converts the abort into latency. + // It overshoots — non-V8 handlers (Browser.*, Storage.*, Emulation.*) + // also serialize — but those are cheap and the safety win dominates. + // + // The properly concurrent fix is to pin each `JsRuntime` to its own OS + // thread and message-pass; that's tracked as the larger #19 follow-up. + let _v8_guard = obscura_js::v8_lock::global().lock().await; + let (domain, method) = match req.method.split_once('.') { Some((d, m)) => (d, m), None => { diff --git a/crates/obscura-cdp/src/server.rs b/crates/obscura-cdp/src/server.rs index e4ef76d..0201964 100644 --- a/crates/obscura-cdp/src/server.rs +++ b/crates/obscura-cdp/src/server.rs @@ -9,6 +9,11 @@ use tokio_tungstenite::tungstenite::Message; use tracing::{error, info, warn}; use crate::dispatch::{self, CdpContext}; + +// PR #36 comment 4341743194: the deferral queue in `process_with_interception` +// must be bounded so a stalled navigation cannot OOM the process. When the cap +// is reached we return an explicit error response rather than silently dropping. +const MAX_DEFERRED_MESSAGES: usize = 256; use crate::types::CdpRequest; struct CdpMessage { @@ -89,7 +94,29 @@ async fn cdp_processor( let mut intercept_rx: Option> = Some(irx); let mut intercepted_paused: HashMap> = HashMap::new(); - while let Some(msg) = rx.recv().await { + // Issue #19 follow-up: messages deferred from inside + // `process_with_interception` because routing them through + // `process_cdp_message → dispatch` while a nav was in flight would have + // tripped V8's TryGetCurrent invariant. Drained at the top of each + // outer iteration so they get processed sequentially with no other nav + // in flight. + let mut deferred: std::collections::VecDeque = + std::collections::VecDeque::new(); + + loop { + // Drain any deferred messages from the previous interception window + // before pulling new ones off the wire. Each is processed with no + // nav-task spawn_local in flight, so dispatch's v8_lock can claim + // the only Isolate cleanly. + let msg = if let Some(d) = deferred.pop_front() { + d + } else { + match rx.recv().await { + Some(m) => m, + None => break, + } + }; + match msg { ServerMessage::NewConnection { reply_tx } => { let _ = reply_tx.send( @@ -105,6 +132,7 @@ async fn cdp_processor( process_with_interception( &cdp_msg.text, &mut ctx, &cdp_msg.reply_tx, &mut rx, &mut intercept_rx, &mut intercepted_paused, + &mut deferred, ).await; } else { if cdp_msg.text.contains("Fetch.") { @@ -169,6 +197,7 @@ async fn process_with_interception( rx: &mut mpsc::UnboundedReceiver, intercept_rx: &mut Option>, intercepted_paused: &mut HashMap>, + deferred: &mut std::collections::VecDeque, ) { let req: CdpRequest = match serde_json::from_str(text) { Ok(r) => r, @@ -203,6 +232,21 @@ async fn process_with_interception( } }; + // Issue #19 follow-up: V8 only allows ONE entered Isolate per OS thread. + // The regular dispatch path enforces this via `get_session_page_mut` + // (which `suspend_js`'es every other page before letting the target + // page run JS). The interception path here bypasses that — it removes + // the target page and spawns a nav task — so we have to enforce the + // same invariant explicitly. Otherwise nav-2's `init_js` constructs + // Isolate-2 while page-1's Isolate-1 is still alive in ctx.pages, and + // the next V8 scope unwind aborts the process via `Context::Exit`'s + // `heap->isolate() == Isolate::TryGetCurrent()` check. + for other in ctx.pages.iter_mut() { + if other.has_js() { + other.suspend_js(); + } + } + let url = req.params.get("url").and_then(|v| v.as_str()).unwrap_or(""); let wait_until = req.params.get("waitUntil") .and_then(|v| { @@ -238,18 +282,44 @@ async fn process_with_interception( let url_owned = url.to_string(); tokio::task::spawn_local(async move { + // Issue #19: serialize V8 work across pages. The interception path + // spawns navigation here while the parent task continues to pump + // CDP messages via `dispatch` (which also acquires this lock); both + // sides must coordinate or V8 aborts the process at concurrency >= 5. + let _v8_guard = obscura_js::v8_lock::global().lock().await; let result = page.navigate_with_wait(&url_owned, wait_until).await.map_err(|e| e.to_string()); for source in &preload_scripts { if let Err(e) = page.execute_preload_script(source) { tracing::debug!("Preload script error: {}", e); } } + drop(_v8_guard); let _ = nav_done_tx.send((page, result)).await; }); let navigate_result: Result<(), String>; let page_back: Option; + // Issue #19 follow-up (PR #36 maintainer's fetch-intercept repro): + // While the spawned nav task is executing V8 (potentially parked on + // `op_fetch_url`'s `resolve_rx.await` *with Isolate-N still entered*), + // we must NOT let the parent's `select!` route foreign Cdp messages + // through `process_cdp_message → dispatch → page handlers`, because + // those handlers call `get_session_page_mut` which `suspend_js`'es + // OTHER pages (drops their `JsRuntime`, which calls + // `JsRealmInner::destroy`). That trips V8's + // `heap->isolate() == Isolate::TryGetCurrent()` invariant and aborts + // the process via `V8_Fatal`. + // + // The `obscura_js::v8_lock` mutex doesn't save us here: it's a + // `tokio::sync::Mutex` that is released around `.await`s inside V8 + // ops, so it doesn't actually keep the V8 enter/exit pair contiguous + // on the thread. + // + // Park foreign Cdp messages into the outer deferred queue so the + // outer `cdp_processor` loop processes them after this nav fully + // completes (and its JsRuntime is no longer in flight on the + // LocalSet). loop { let has_irx = intercept_rx.is_some(); @@ -320,16 +390,46 @@ async fn process_with_interception( tracing::info!("INTERCEPTION select: received CDP message during navigation"); match msg { ServerMessage::NewConnection { reply_tx: new_tx } => { + // Safe: no V8 enter, just bookkeeping. let pid = ctx.create_page(); let sid = format!("{}-session", pid); ctx.sessions.insert(sid.clone(), pid.clone()); let _ = new_tx.send(json!({"__init": true, "pageId": pid, "sessionId": sid}).to_string()); } ServerMessage::Cdp(msg) => { - if msg.text.contains("Fetch.") { + if msg.text.contains("Fetch.continueRequest") + || msg.text.contains("Fetch.fulfillRequest") + || msg.text.contains("Fetch.failRequest") + { + // Safe: only flips a oneshot to resume the parked + // op inside the spawned nav task. No V8 enter on + // this side; the actual V8 work happens back on + // the nav task's thread. handle_fetch_resolution(&msg.text, ctx, &msg.reply_tx, intercepted_paused); } else { - process_cdp_message(&msg.text, ctx, &msg.reply_tx).await; + // UNSAFE during nav: would route through dispatch, + // which can `suspend_js` other pages and trip the + // V8 invariant. Defer until nav completes — + // pushed to the outer `cdp_processor` queue so + // it's processed sequentially with no nav task + // in flight. + if deferred.len() >= MAX_DEFERRED_MESSAGES { + tracing::warn!("INTERCEPTION: deferred queue full ({}), returning error to client", MAX_DEFERRED_MESSAGES); + if let Ok(req) = serde_json::from_str::(&msg.text) { + let resp = crate::types::CdpResponse::error( + req.id, + -32000, + "Server busy: navigation in progress, try again later".to_string(), + req.session_id, + ); + if let Ok(json) = serde_json::to_string(&resp) { + let _ = msg.reply_tx.send(json); + } + } + } else { + tracing::info!("INTERCEPTION: deferring CDP message until nav completes"); + deferred.push_back(ServerMessage::Cdp(msg)); + } } } } @@ -337,6 +437,9 @@ async fn process_with_interception( } } + // Deferred messages are handled by the outer `cdp_processor` loop + // (it drains `deferred` before pulling the next message off `rx`). + let mut page = page_back.expect("navigation task should return the page"); let network_events: Vec<_> = page.network_events.drain(..).collect(); diff --git a/crates/obscura-cdp/tests/concurrent_navigations.rs b/crates/obscura-cdp/tests/concurrent_navigations.rs new file mode 100644 index 0000000..309daf4 --- /dev/null +++ b/crates/obscura-cdp/tests/concurrent_navigations.rs @@ -0,0 +1,145 @@ +//! Issue #19 smoke test: 5 parallel CDP clients each performing +//! `Target.createTarget` + `Page.navigate` must not abort the process. +//! +//! NOTE on coverage. The deterministic abort in #19 requires the navigations +//! to interleave on the shared `LocalSet` thread, which only happens once +//! `navigate_single` actually yields — the heaviest yields come from +//! subresource fetches in `futures::future::join_all` (page.rs:285) when the +//! page has scripts/images to pull. `data:` URLs skip every fetch, so this +//! test exercises the chokepoint shape (5 clients hitting `dispatch` +//! concurrently) without driving the original abort. Treat it as a smoke +//! check: it ensures the V8-lock plumbing compiles and isn't hitting an +//! obvious deadlock, not as a strict regression for the abort. +//! +//! For an end-to-end repro of the abort, the issue author used +//! `scrapegraphai.com` driven from Node CDP at concurrency 5. Reproducing +//! that here would require standing up a local server with JS subresources; +//! out of scope for this PR. +//! +//! Run with `cargo test -p obscura-cdp --test concurrent_navigations +//! -- --nocapture --ignored` (the `ignored` gate keeps it out of the default +//! suite — it boots a real CDP server, which is heavier than a unit test). + +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio::net::TcpListener; +use tokio_tungstenite::{connect_async, tungstenite::Message}; + +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 +} + +async fn one_client(port: u16, id_base: u64) -> Result<(), String> { + let url = format!("ws://127.0.0.1:{}/devtools/browser", port); + let (mut ws, _) = connect_async(&url).await.map_err(|e| e.to_string())?; + + // Target.createTarget — get a sessionId via Target.attachToTarget. + let create = json!({ + "id": id_base, + "method": "Target.createTarget", + "params": {"url": "about:blank"}, + }); + ws.send(Message::Text(create.to_string().into())) + .await + .map_err(|e| e.to_string())?; + + let mut session_id: Option = None; + let mut target_id: Option = None; + while session_id.is_none() { + let msg = tokio::time::timeout(Duration::from_secs(5), 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(t) = v.get("result").and_then(|r| r.get("targetId")).and_then(|s| s.as_str()) { + target_id = Some(t.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(); + let _ = target_id; // kept for debugging — unused by the assertion path + + // Page.navigate to a data URL — exercises init_js + execute_scripts + // without touching the network, which is what the V8 race needs. + let nav = json!({ + "id": id_base + 1, + "method": "Page.navigate", + "sessionId": sid, + "params": {"url": "data:text/html,

x

"}, + }); + ws.send(Message::Text(nav.to_string().into())) + .await + .map_err(|e| e.to_string())?; + + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + if tokio::time::Instant::now() >= deadline { + return Err("timeout waiting for navigate response".to_string()); + } + let remaining = deadline - tokio::time::Instant::now(); + let msg = tokio::time::timeout(remaining, ws.next()) + .await + .map_err(|_| "timeout".to_string())? + .ok_or("ws closed mid-navigate")? + .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 + 1) { + return Ok(()); + } + } +} + +#[tokio::test(flavor = "current_thread")] +#[ignore = "boots a real CDP server; opt-in with --ignored"] +async fn concurrency_5_does_not_abort_v8() { + let port = pick_port().await; + let local = tokio::task::LocalSet::new(); + local + .run_until(async { + tokio::task::spawn_local(async move { + let _ = obscura_cdp::server::start(port).await; + }); + // Give the listener a beat. + tokio::time::sleep(Duration::from_millis(150)).await; + + let mut handles = Vec::new(); + for i in 0..5u64 { + let id_base = (i + 1) * 1000; + handles.push(tokio::task::spawn_local(async move { + one_client(port, id_base).await + })); + } + + let mut ok = 0usize; + for (i, h) in handles.into_iter().enumerate() { + match h.await { + Ok(Ok(())) => ok += 1, + Ok(Err(e)) => panic!("client {} failed: {}", i, e), + Err(e) => panic!("client {} join error: {}", i, e), + } + } + assert_eq!(ok, 5, "all 5 concurrent clients must complete navigate"); + }) + .await; +} diff --git a/crates/obscura-cdp/tests/concurrent_navigations_with_fetch.rs b/crates/obscura-cdp/tests/concurrent_navigations_with_fetch.rs new file mode 100644 index 0000000..1813eea --- /dev/null +++ b/crates/obscura-cdp/tests/concurrent_navigations_with_fetch.rs @@ -0,0 +1,215 @@ +//! Issue #19 hard repro: 5 parallel clients with Fetch interception enabled +//! must not abort V8. +//! +//! Per maintainer comment on PR #36 (https://github.com/h4ckf0r0day/obscura/pull/36#issuecomment-4328238087): +//! +//! "Tested locally and the V8 fatal still reproduces when Fetch.enable +//! interception is on with concurrent navigations: +//! - 5 clients enable Fetch with patterns: ["*"] +//! - All 5 issue Page.navigate concurrently +//! - Server aborts with: Check failed: heap->isolate() == Isolate::TryGetCurrent()" +//! +//! The smoke test (concurrent_navigations.rs) used data: URLs which skip +//! the subresource fetch path that drives the abort. This test: +//! 1. Calls Fetch.enable patterns:["*"] on each client +//! 2. Navigates to a URL with subresources (so op_fetch_url is reached +//! and parks on resolve_rx.await INSIDE the v8_lock guard) +//! 3. Auto-replies to Fetch.requestPaused with Fetch.continueRequest +//! (so the parked op resumes — that's where two Isolates can collide) +//! +//! Run with `cargo test -p obscura-cdp --test concurrent_navigations_with_fetch +//! -- --nocapture --ignored`. + +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio::net::TcpListener; +use tokio_tungstenite::{connect_async, tungstenite::Message}; + +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 +} + +/// One client: +/// 1. Target.createTarget → sessionId +/// 2. Fetch.enable patterns:["*"] +/// 3. Page.navigate to URL +/// 4. Loop: respond to Fetch.requestPaused with Fetch.continueRequest +/// until Page.navigate's response arrives. +async fn one_client_with_fetch(port: u16, id_base: u64, target_url: &str) -> Result<(), String> { + let url = format!("ws://127.0.0.1:{}/devtools/browser", port); + let (mut ws, _) = connect_async(&url).await.map_err(|e| e.to_string())?; + + // 1. Target.createTarget + let create = json!({ + "id": id_base, + "method": "Target.createTarget", + "params": {"url": "about:blank"}, + }); + ws.send(Message::Text(create.to_string().into())) + .await + .map_err(|e| e.to_string())?; + + let mut session_id: Option = None; + while session_id.is_none() { + let msg = tokio::time::timeout(Duration::from_secs(5), 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(); + + // 2. Fetch.enable patterns:["*"] + let fetch_enable = json!({ + "id": id_base + 1, + "method": "Fetch.enable", + "sessionId": sid, + "params": { + "patterns": [{"urlPattern": "*"}], + }, + }); + ws.send(Message::Text(fetch_enable.to_string().into())) + .await + .map_err(|e| e.to_string())?; + + // Wait for Fetch.enable response. + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + if tokio::time::Instant::now() >= deadline { + return Err("timeout waiting for Fetch.enable response".to_string()); + } + let remaining = deadline - tokio::time::Instant::now(); + let msg = tokio::time::timeout(remaining, ws.next()) + .await + .map_err(|_| "timeout".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 v.get("id").and_then(|x| x.as_u64()) == Some(id_base + 1) { + break; + } + } + + // 3. Page.navigate + let nav = json!({ + "id": id_base + 2, + "method": "Page.navigate", + "sessionId": sid, + "params": {"url": target_url}, + }); + ws.send(Message::Text(nav.to_string().into())) + .await + .map_err(|e| e.to_string())?; + + // 4. Drain events. Auto-respond to Fetch.requestPaused with continueRequest. + let deadline = tokio::time::Instant::now() + Duration::from_secs(60); + let mut auto_id = id_base + 1000; + loop { + if tokio::time::Instant::now() >= deadline { + return Err(format!("client {} timeout waiting for navigate", id_base)); + } + let remaining = deadline - tokio::time::Instant::now(); + let msg = tokio::time::timeout(remaining, ws.next()) + .await + .map_err(|_| "timeout".to_string())? + .ok_or("ws closed mid-navigate")? + .map_err(|e| e.to_string())?; + let text = match msg { + Message::Text(t) => t.to_string(), + _ => continue, + }; + let v: Value = match serde_json::from_str::(&text) { + Ok(v) => v, + Err(_) => continue, + }; + + // navigate response? + if v.get("id").and_then(|x| x.as_u64()) == Some(id_base + 2) { + return Ok(()); + } + + // Fetch.requestPaused event? + if v.get("method").and_then(|m| m.as_str()) == Some("Fetch.requestPaused") { + let req_id = v + .get("params") + .and_then(|p| p.get("requestId")) + .and_then(|s| s.as_str()) + .unwrap_or("") + .to_string(); + auto_id += 1; + let cont = json!({ + "id": auto_id, + "method": "Fetch.continueRequest", + "sessionId": sid, + "params": {"requestId": req_id}, + }); + ws.send(Message::Text(cont.to_string().into())) + .await + .map_err(|e| e.to_string())?; + } + } +} + +/// PR #36 maintainer's exact repro. +#[tokio::test(flavor = "current_thread")] +#[ignore = "boots a real CDP server + makes outbound HTTP; opt-in with --ignored"] +async fn fetch_intercept_concurrency_5_does_not_abort_v8() { + // Use example.com first — minimal subresources but real network. If this + // doesn't reproduce, escalate to a JS-heavy page in a follow-up. + let target_url = "https://example.com/"; + + let port = pick_port().await; + let local = tokio::task::LocalSet::new(); + local + .run_until(async { + tokio::task::spawn_local(async move { + let _ = obscura_cdp::server::start(port).await; + }); + tokio::time::sleep(Duration::from_millis(250)).await; + + let mut handles = Vec::new(); + for i in 0..5u64 { + let id_base = (i + 1) * 1000; + let url = target_url.to_string(); + handles.push(tokio::task::spawn_local(async move { + one_client_with_fetch(port, id_base, &url).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)), + } + } + if !errors.is_empty() { + panic!("errors: {:#?}", errors); + } + assert_eq!(ok, 5, "all 5 concurrent clients must complete navigate"); + }) + .await; +} diff --git a/crates/obscura-js/src/lib.rs b/crates/obscura-js/src/lib.rs index b538ebd..fe1d5cb 100644 --- a/crates/obscura-js/src/lib.rs +++ b/crates/obscura-js/src/lib.rs @@ -4,3 +4,4 @@ extern crate html5ever; pub mod module_loader; pub mod runtime; pub mod ops; +pub mod v8_lock; diff --git a/crates/obscura-js/src/v8_lock.rs b/crates/obscura-js/src/v8_lock.rs new file mode 100644 index 0000000..b598821 --- /dev/null +++ b/crates/obscura-js/src/v8_lock.rs @@ -0,0 +1,27 @@ +//! Process-wide async lock that serializes V8 work across all `JsRuntime`s. +//! +//! V8's invariant: on any given OS thread, only one `Isolate` may be entered +//! at a time, and HandleScope/ContextScope stacks must unwind on the Isolate +//! they belong to. Obscura's CDP server runs every `JsRuntime` (one per Page) +//! on a single OS thread via `tokio::task::LocalSet` + `spawn_local`. As soon +//! as two pages' V8-touching futures interleave across an `.await`, V8 trips +//! its `heap->isolate() == Isolate::TryGetCurrent()` check and aborts the +//! whole process (no Rust panic; `V8_Fatal` calls `abort(3)`). +//! +//! Acquiring this lock around any block that calls `JsRuntime::execute_script` +//! or `JsRuntime::run_event_loop` keeps that block contiguous on the thread: +//! V8 fully exits the prior Isolate before the next page is allowed in. This +//! converts the abort into latency. It is the issue-19 "Option 1" fix. +//! +//! The properly concurrent fix is to pin each `JsRuntime` to its own OS +//! thread (issue-19 "Option 2"); that's a larger refactor tracked separately. + +use std::sync::OnceLock; +use tokio::sync::Mutex; + +static V8_LOCK: OnceLock> = OnceLock::new(); + +/// Returns the process-wide V8 serialization lock. +pub fn global() -> &'static Mutex<()> { + V8_LOCK.get_or_init(|| Mutex::new(())) +}