* 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<ServerMessage> 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.
This commit is contained in:
Timothy Olaleke
@@ -0,0 +1,2 @@
|
||||
target/
|
||||
.DS_Store
|
||||
@@ -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"
|
||||
|
||||
@@ -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 => {
|
||||
|
||||
@@ -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<mpsc::UnboundedReceiver<obscura_js::ops::InterceptedRequest>> = Some(irx);
|
||||
let mut intercepted_paused: HashMap<String, tokio::sync::oneshot::Sender<obscura_js::ops::InterceptResolution>> = 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<ServerMessage> =
|
||||
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<ServerMessage>,
|
||||
intercept_rx: &mut Option<mpsc::UnboundedReceiver<obscura_js::ops::InterceptedRequest>>,
|
||||
intercepted_paused: &mut HashMap<String, tokio::sync::oneshot::Sender<obscura_js::ops::InterceptResolution>>,
|
||||
deferred: &mut std::collections::VecDeque<ServerMessage>,
|
||||
) {
|
||||
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<obscura_browser::Page>;
|
||||
|
||||
// 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::<CdpRequest>(&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();
|
||||
|
||||
@@ -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<String> = None;
|
||||
let mut target_id: Option<String> = 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,<html><body><h1>x</h1></body></html>"},
|
||||
});
|
||||
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;
|
||||
}
|
||||
@@ -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<String> = 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::<Value>(&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;
|
||||
}
|
||||
@@ -4,3 +4,4 @@ extern crate html5ever;
|
||||
pub mod module_loader;
|
||||
pub mod runtime;
|
||||
pub mod ops;
|
||||
pub mod v8_lock;
|
||||
|
||||
@@ -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<Mutex<()>> = OnceLock::new();
|
||||
|
||||
/// Returns the process-wide V8 serialization lock.
|
||||
pub fn global() -> &'static Mutex<()> {
|
||||
V8_LOCK.get_or_init(|| Mutex::new(()))
|
||||
}
|
||||
Reference in New Issue
Block a user