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.

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<tokio::sync::mpsc::UnboundedSender<InterceptedRequest>>,
+    // 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<String, (Vec<u8>, 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,

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)),
     }
 }

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<Value, String> {
+    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)),
+    }
+}

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;