From 1ca604735109588ddf1aa680b843caec8c7d62f2 Mon Sep 17 00:00:00 2001 From: SGavrl <277843151+SGavrl@users.noreply.github.com> Date: Fri, 10 Jul 2026 20:38:28 +0000 Subject: [PATCH] feat(cdp): stream large response bodies via takeResponseBodyAsStream + IO (#381) Network.getResponseBody returns the whole body in one base64 blob and the cache drops anything over 2 MiB, so large downloads (images, PDFs, archives) either failed or forced the long-running server to hold the full body in memory (issue #360). Add Fetch.takeResponseBodyAsStream + IO.read + IO.close. The body is moved out of the page cache into a stream handle and delivered in bounded chunks via IO.read, then freed on IO.close, so it is held once and released after the download. Binary bodies are byte-exact (stored base64). To stream a large body, raise OBSCURA_NETWORK_BODY_BUFFER_BYTES so it is cached. This is entirely opt-in: nothing runs unless a client calls the new methods, so the default fetch/render/scrape path is unchanged (verified: obstacle course 33/33, obscura-vs-Chrome head-to-head within noise of the pre-change baseline). Fixes #360. --- Cargo.lock | 1 + README.md | 5 ++- crates/obscura-browser/src/page.rs | 25 ++++++++++++ crates/obscura-cdp/Cargo.toml | 1 + crates/obscura-cdp/src/dispatch.rs | 8 ++++ crates/obscura-cdp/src/domains/fetch.rs | 30 ++++++++++++++ crates/obscura-cdp/src/domains/io.rs | 53 +++++++++++++++++++++++++ crates/obscura-cdp/src/domains/mod.rs | 1 + 8 files changed, 123 insertions(+), 1 deletion(-) create mode 100644 crates/obscura-cdp/src/domains/io.rs diff --git a/Cargo.lock b/Cargo.lock index 46c3b52..0fbe273 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1662,6 +1662,7 @@ name = "obscura-cdp" version = "0.1.0" dependencies = [ "anyhow", + "base64", "futures-util", "obscura-browser", "obscura-dom", diff --git a/README.md b/README.md index 2df30fd..b4c8a4b 100644 --- a/README.md +++ b/README.md @@ -363,10 +363,13 @@ Obscura implements the Chrome DevTools Protocol for Puppeteer/Playwright compati | **Runtime** | evaluate, callFunctionOn, getProperties, addBinding | | **DOM** | getDocument, querySelector, querySelectorAll, getOuterHTML, resolveNode | | **Network** | enable, setCookies, getCookies, setExtraHTTPHeaders, setUserAgentOverride | -| **Fetch** | enable, continueRequest, fulfillRequest, failRequest (live interception) | +| **Fetch** | enable, continueRequest, fulfillRequest, failRequest (live interception), takeResponseBodyAsStream | +| **IO** | read, close (stream a large response body in chunks) | | **Storage** | getCookies, setCookies, deleteCookies | | **Input** | dispatchMouseEvent, dispatchKeyEvent | | **LP** | getMarkdown (DOM-to-Markdown conversion) | + +To download a large resource without one giant `Network.getResponseBody` blob, call `Fetch.takeResponseBodyAsStream` then read it in chunks with `IO.read` / `IO.close`. Response bodies over the cache limit (`OBSCURA_NETWORK_BODY_BUFFER_BYTES`, default 2 MiB) are not retained, so raise that limit when you intend to stream large downloads. ## CLI Reference ### Tuning V8 diff --git a/crates/obscura-browser/src/page.rs b/crates/obscura-browser/src/page.rs index 671241d..7d8f5fe 100644 --- a/crates/obscura-browser/src/page.rs +++ b/crates/obscura-browser/src/page.rs @@ -1482,6 +1482,31 @@ impl Page { }) } + /// Take a stored response body as raw bytes for CDP streaming + /// (Fetch.takeResponseBodyAsStream). Removes it from the in-memory cache and + /// transfers ownership to the caller, so a large body is held once and freed + /// when the stream is closed rather than lingering in this long-running + /// process (issue #360). Binary bodies are stored base64 (byte-exact); text + /// bodies return their UTF-8 bytes. Returns None if the body was never + /// cached (e.g. it exceeded OBSCURA_NETWORK_BODY_BUFFER_BYTES and was + /// dropped) or the id is unknown. + pub fn take_response_body_raw(&mut self, request_id: &str) -> Option> { + let stored = if let Some(body) = self.response_bodies.remove(request_id) { + self.response_body_order.retain(|id| id != request_id); + body + } else { + self.js.as_ref()?.get_network_response_body(request_id).map(|b| StoredResponseBody { + body: b.body, + base64_encoded: b.base64_encoded, + })? + }; + if stored.base64_encoded { + BASE64.decode(stored.body.as_bytes()).ok() + } else { + Some(stored.body.into_bytes()) + } + } + /// Make the body stored under `from_id` also retrievable under `to_id`. /// The main navigation resource is stored under its internal request id, but /// the CDP layer reports it to clients with the navigation's loaderId as the diff --git a/crates/obscura-cdp/Cargo.toml b/crates/obscura-cdp/Cargo.toml index ff6d1cb..973dc00 100644 --- a/crates/obscura-cdp/Cargo.toml +++ b/crates/obscura-cdp/Cargo.toml @@ -17,6 +17,7 @@ tracing = { workspace = true } thiserror = { workspace = true } anyhow = { workspace = true } futures-util = "0.3" +base64 = "0.22" obscura-net = { workspace = true } [dev-dependencies] diff --git a/crates/obscura-cdp/src/dispatch.rs b/crates/obscura-cdp/src/dispatch.rs index bb3b4a1..d1218ea 100644 --- a/crates/obscura-cdp/src/dispatch.rs +++ b/crates/obscura-cdp/src/dispatch.rs @@ -45,6 +45,11 @@ pub struct CdpContext { pub next_isolated_context_id: i64, pub fetch_intercept: FetchInterceptState, pub intercept_tx: Option>, + // Open IO streams for Fetch.takeResponseBodyAsStream: handle -> (bytes, cursor). + // Each holds a response body taken out of the page cache so a large download + // is streamed chunk-by-chunk via IO.read and freed on IO.close (issue #360). + pub io_streams: HashMap, usize)>, + pub io_stream_counter: u64, } impl CdpContext { @@ -141,6 +146,8 @@ impl CdpContext { isolated_worlds: Vec::new(), valid_context_ids, next_isolated_context_id: 100, + io_streams: HashMap::new(), + io_stream_counter: 0, } } @@ -334,6 +341,7 @@ pub async fn dispatch(req: &CdpRequest, ctx: &mut CdpContext) -> CdpResponse { "Runtime" => domains::runtime::handle(method, &req.params, ctx, &req.session_id).await, "Network" => domains::network::handle(method, &req.params, ctx, &req.session_id).await, "Fetch" => domains::fetch::handle(method, &req.params, ctx, &req.session_id).await, + "IO" => domains::io::handle(method, &req.params, ctx).await, "Input" => domains::input::handle(method, &req.params, ctx, &req.session_id).await, "Storage" => domains::storage::handle(method, &req.params, ctx, &req.session_id).await, "LP" => domains::lp::handle(method, &req.params, ctx, &req.session_id).await, diff --git a/crates/obscura-cdp/src/domains/fetch.rs b/crates/obscura-cdp/src/domains/fetch.rs index 6a84fe3..457e6b9 100644 --- a/crates/obscura-cdp/src/domains/fetch.rs +++ b/crates/obscura-cdp/src/domains/fetch.rs @@ -181,6 +181,36 @@ pub async fn handle( "getResponseBody" => { Ok(json!({ "body": "", "base64Encoded": false })) } + "takeResponseBodyAsStream" => { + // Hand the client a streaming handle for a large response body so it + // can pull it in chunks via IO.read and free it with IO.close, + // instead of receiving one giant base64 blob (issue #360). The body + // is moved out of the page cache into the stream, so it is held once + // and released on close. Requires the body to have been cached + // (raise OBSCURA_NETWORK_BODY_BUFFER_BYTES for large downloads). + let request_id = params + .get("requestId") + .and_then(|v| v.as_str()) + .ok_or("Fetch.takeResponseBodyAsStream requires requestId")?; + + let bytes = { + let page = ctx.get_session_page_mut(session_id).ok_or("No page")?; + page.take_response_body_raw(request_id) + } + .or_else(|| { + ctx.pages + .iter_mut() + .find_map(|p| p.take_response_body_raw(request_id)) + }) + .ok_or_else(|| { + format!("Fetch.takeResponseBodyAsStream: no cached body for {request_id}") + })?; + + let handle = format!("stream-{}", ctx.io_stream_counter); + ctx.io_stream_counter += 1; + ctx.io_streams.insert(handle.clone(), (bytes, 0)); + Ok(json!({ "stream": handle })) + } _ => Err(format!("Unknown Fetch method: {}", method)), } } diff --git a/crates/obscura-cdp/src/domains/io.rs b/crates/obscura-cdp/src/domains/io.rs new file mode 100644 index 0000000..9115e9b --- /dev/null +++ b/crates/obscura-cdp/src/domains/io.rs @@ -0,0 +1,53 @@ +use base64::Engine as _; +use serde_json::{json, Value}; + +use crate::dispatch::CdpContext; + +// Default chunk size when the client does not pass `size`. Chrome uses a similar +// order of magnitude; keeping chunks bounded is the point of streaming (issue +// #360), so we never return the whole body in one IO.read. +const DEFAULT_CHUNK: usize = 1 << 20; // 1 MiB + +/// CDP IO domain. Streams a response body handed out by +/// Fetch.takeResponseBodyAsStream: IO.read returns the next base64 chunk and +/// IO.close frees the buffer. Nothing here runs unless a client opened a stream. +pub async fn handle(method: &str, params: &Value, ctx: &mut CdpContext) -> Result { + match method { + "read" => { + let handle = params + .get("handle") + .and_then(|v| v.as_str()) + .ok_or("IO.read requires handle")?; + let size = params + .get("size") + .and_then(|v| v.as_u64()) + .map(|s| s as usize) + .unwrap_or(DEFAULT_CHUNK) + .max(1); + + let (bytes, cursor) = ctx + .io_streams + .get_mut(handle) + .ok_or_else(|| format!("IO.read: unknown handle {handle}"))?; + + let start = (*cursor).min(bytes.len()); + let end = start.saturating_add(size).min(bytes.len()); + let chunk = &bytes[start..end]; + let data = base64::engine::general_purpose::STANDARD.encode(chunk); + *cursor = end; + let eof = end >= bytes.len(); + + Ok(json!({ "data": data, "eof": eof, "base64Encoded": true })) + } + "close" => { + let handle = params + .get("handle") + .and_then(|v| v.as_str()) + .ok_or("IO.close requires handle")?; + // Dropping the entry frees the buffered body. + ctx.io_streams.remove(handle); + Ok(json!({})) + } + _ => Err(format!("Unknown IO method: {}", method)), + } +} diff --git a/crates/obscura-cdp/src/domains/mod.rs b/crates/obscura-cdp/src/domains/mod.rs index ca65521..41534f7 100644 --- a/crates/obscura-cdp/src/domains/mod.rs +++ b/crates/obscura-cdp/src/domains/mod.rs @@ -6,6 +6,7 @@ pub mod domsnapshot; pub mod runtime; pub mod network; pub mod fetch; +pub mod io; pub mod input; pub mod storage; pub mod accessibility;