From 928089d392a90dc2345e3d4bea310b1a5cdd66a7 Mon Sep 17 00:00:00 2001 From: Sergi Gonzalez <31130069+TheHypnoo@users.noreply.github.com> Date: Sun, 27 Sep 2026 18:29:16 +0200 Subject: [PATCH 1/3] test(fetch): cover Request, Response and fetch bodies built from async iterables --- .../fetch/request/async-iterable-body-send.ts | 32 +++++++++++++ .../fetch/request/async-iterable-body.ts | 48 +++++++++++++++++++ .../fetch/request/mutated-init-local.ts | 31 ++++++++++++ 3 files changed, 111 insertions(+) create mode 100644 test-parity/node-suite/fetch/request/async-iterable-body-send.ts create mode 100644 test-parity/node-suite/fetch/request/async-iterable-body.ts create mode 100644 test-parity/node-suite/fetch/request/mutated-init-local.ts diff --git a/test-parity/node-suite/fetch/request/async-iterable-body-send.ts b/test-parity/node-suite/fetch/request/async-iterable-body-send.ts new file mode 100644 index 00000000000..96b4e0afba3 --- /dev/null +++ b/test-parity/node-suite/fetch/request/async-iterable-body-send.ts @@ -0,0 +1,32 @@ +// fetch(url, { body: asyncIterable, duplex: "half" }) must send the iterable's +// chunks as the request body. +import http from "node:http"; + +const enc = new TextEncoder(); +async function* gen() { + yield enc.encode("hello "); + yield "async "; + yield enc.encode("world"); +} + +const server = http.createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => res.end(`${req.method} ${Buffer.concat(chunks).toString()}`)); +}); + +server.listen(0, "127.0.0.1", async () => { + const { port } = server.address() as { port: number }; + try { + const res = await fetch(`http://127.0.0.1:${port}/`, { + method: "POST", + // Cast only the body: a cast around the whole init literal is a + // separate fetch-init lowering gap. + body: gen() as any, + duplex: "half", + }); + console.log(res.status, await res.text()); + } finally { + server.close(); + } +}); diff --git a/test-parity/node-suite/fetch/request/async-iterable-body.ts b/test-parity/node-suite/fetch/request/async-iterable-body.ts new file mode 100644 index 00000000000..e03478de2a3 --- /dev/null +++ b/test-parity/node-suite/fetch/request/async-iterable-body.ts @@ -0,0 +1,48 @@ +// Request/Response bodies built from async iterables (async generators), +// the shape Astro's node adapter uses for POST bodies. +const enc = new TextEncoder(); + +async function* bytes(...parts: string[]) { + for (const p of parts) yield enc.encode(p); +} + +async function* strings(...parts: string[]) { + for (const p of parts) yield p; +} + +async function* empty() {} + +async function* throwing() { + yield enc.encode("partial"); + throw new Error("boom"); +} + +function req(body: unknown): Request { + return new Request("http://x/", { method: "POST", body, duplex: "half" } as any); +} + +async function attempt(label: string, fn: () => Promise) { + try { + console.log(label, JSON.stringify(await fn())); + } catch (e: any) { + console.log(label, "threw", e?.name, e?.message); + } +} + +await attempt("text:", () => req(bytes('{"a":', "1}")).text()); +await attempt("json:", () => req(bytes('{"a":', "1}")).json()); +await attempt("arrayBuffer:", async () => { + const buf = await req(bytes("ab", "cd")).arrayBuffer(); + return [buf.byteLength, ...new Uint8Array(buf)]; +}); +await attempt("bodyUsed:", async () => { + const r = req(bytes("x")); + const before = r.bodyUsed; + await r.text(); + return [before, r.bodyUsed]; +}); +await attempt("string chunks:", () => req(strings("a", "b")).text()); +await attempt("empty:", () => req(empty()).text()); +await attempt("throwing:", () => req(throwing()).text()); +await attempt("response:", () => new Response(bytes("res", "ponse") as any).text()); +await attempt("response json:", () => new Response(bytes('[1,', "2]") as any).json()); diff --git a/test-parity/node-suite/fetch/request/mutated-init-local.ts b/test-parity/node-suite/fetch/request/mutated-init-local.ts new file mode 100644 index 00000000000..a9145f414fd --- /dev/null +++ b/test-parity/node-suite/fetch/request/mutated-init-local.ts @@ -0,0 +1,31 @@ +// A `const` RequestInit local mutated after its literal initializer (the shape +// Astro's node adapter uses: `Object.assign(options, { body, duplex })`) must be +// read as it is at the `new Request` call, not as its initializer literal. +const enc = new TextEncoder(); +async function* gen() { + yield enc.encode('{"n":'); + yield enc.encode("1}"); +} + +let headerCalls = 0; +function makeHeaders() { + headerCalls++; + return { "x-a": "1" }; +} + +const assigned = { method: "POST", headers: makeHeaders() }; +Object.assign(assigned, { body: '{"n":1}', duplex: "half" }); +const r1 = new Request("http://x/", assigned); +console.log("assign:", r1.method, r1.headers.get("x-a"), JSON.stringify(await r1.text())); + +const set: Record = { method: "PUT" }; +set.body = "plain"; +const r2 = new Request("http://x/", set); +console.log("set:", r2.method, JSON.stringify(await r2.text())); + +const iter = { method: "POST", headers: makeHeaders() }; +Object.assign(iter, { body: gen(), duplex: "half" }); +const r3 = new Request("http://x/", iter); +console.log("async iterable:", JSON.stringify(await r3.json())); + +console.log("header initializers ran:", headerCalls); From 21c5a373f02e46a351dc0a2d14c6338721ea05d7 Mon Sep 17 00:00:00 2001 From: Sergi Gonzalez <31130069+TheHypnoo@users.noreply.github.com> Date: Sun, 27 Sep 2026 18:29:16 +0200 Subject: [PATCH 2/3] fix(fetch): read Request, Response and fetch bodies from async iterables new Request(url, { body: asyncGenerator(), duplex: "half" }) and new Response(asyncGenerator()) stringified the body to "[object AsyncGenerator]", and fetch() sent an empty body. Astro's node adapter passes every POST body this way, so request.json() threw. The BodyInit classifier now stashes an async iterable in a GC-scanned pending slot; take_pending_fetch_body_stream_id converts it with js_readable_stream_from_iterable, so every consumer gets it. Converting runs the iterator's JS, so the consumers copy or root their raw arguments first, and ReadableStream.from keeps the iterator and chunks in runtime handles. The fetch() send path now goes through the classifier. A Request body whose iterable throws rejects text()/json()/... with the original error, kept as a GC-visited value on the record. String chunks are UTF-8 encoded in the Fetch body drain. --- .../src/expr/logical_collections.rs | 7 +- .../perry-stdlib/src/fetch/body_metadata.rs | 24 +--- .../src/fetch/bun_server_bridge.rs | 1 + crates/perry-stdlib/src/fetch/dispatch.rs | 7 ++ crates/perry-stdlib/src/fetch/gc.rs | 6 + crates/perry-stdlib/src/fetch/lifecycle.rs | 1 + crates/perry-stdlib/src/fetch/mod.rs | 109 +++++++++++++----- crates/perry-stdlib/src/fetch/request_copy.rs | 21 +++- crates/perry-stdlib/src/fetch/request_ctor.rs | 41 ++++--- .../perry-stdlib/src/fetch/response_ctor.rs | 5 + crates/perry-stdlib/src/streams.rs | 66 +++++++---- crates/perry-stdlib/src/streams/subclass.rs | 20 +++- test-parity/node_suite_baseline.json | 4 +- 13 files changed, 219 insertions(+), 93 deletions(-) diff --git a/crates/perry-codegen/src/expr/logical_collections.rs b/crates/perry-codegen/src/expr/logical_collections.rs index 9aacf847bfa..e49b793797a 100644 --- a/crates/perry-codegen/src/expr/logical_collections.rs +++ b/crates/perry-codegen/src/expr/logical_collections.rs @@ -548,6 +548,8 @@ pub(crate) fn lower(ctx: &mut FnCtx<'_>, expr: &Expr) -> Result { signal, redirect, } => { + // Drop BodyInit state a throwing operand could leave behind. + ctx.block().call(DOUBLE, "js_response_body_init_reset", &[]); // Lower `init.signal` / `init.redirect` (when present) with the // other operands so they can be stashed for // `js_fetch_with_options` right before the call below. @@ -655,7 +657,10 @@ pub(crate) fn lower(ctx: &mut FnCtx<'_>, expr: &Expr) -> Result { // place it can be correct (slice 1b's `BufferSlice` finding). let url_handle = blk.call(I64, "js_fetch_input_ptr", &[(DOUBLE, &vals[0])]); let method_handle = unbox_to_i64(blk, &vals[1]); - let body_handle = unbox_to_i64(blk, &vals[2]); + // The shared BodyInit classifier: a stream or async-iterable + // body is handed to `js_fetch_with_options` out of band. + let body_handle = + blk.call(I64, "js_response_body_init_ptr", &[(DOUBLE, &vals[2])]); // Stash the AbortSignal so `js_fetch_with_options` can cancel // the request when it aborts (`controller.abort()` / // `AbortSignal.timeout`). diff --git a/crates/perry-stdlib/src/fetch/body_metadata.rs b/crates/perry-stdlib/src/fetch/body_metadata.rs index 1612e536fbb..8e904ad3775 100644 --- a/crates/perry-stdlib/src/fetch/body_metadata.rs +++ b/crates/perry-stdlib/src/fetch/body_metadata.rs @@ -572,13 +572,7 @@ pub unsafe extern "C" fn js_request_blob(handle: f64) -> *mut perry_runtime::Pro let blob_id = alloc_blob(BlobData::blob(body, content_type)); perry_runtime::js_promise_resolve(promise, handle_to_f64(blob_id)); } - Err(err_msg) if err_msg == BODY_ALREADY_USED_MESSAGE => { - reject_fetch_type_error(promise, BODY_ALREADY_USED_MESSAGE); - } - Err(err_msg) => { - let err_nan = f64::from_bits(fetch_error_bits(err_msg)); - perry_runtime::js_promise_reject(promise, err_nan); - } + Err(err) => reject_request_body(promise, err), } promise } @@ -589,13 +583,7 @@ pub unsafe extern "C" fn js_request_bytes(handle: f64) -> *mut perry_runtime::Pr let promise = perry_runtime::js_promise_new(); match consume_request_body(handle) { Ok(body) => resolve_bytes_promise(promise, body), - Err(err_msg) if err_msg == BODY_ALREADY_USED_MESSAGE => { - reject_fetch_type_error(promise, BODY_ALREADY_USED_MESSAGE); - } - Err(err_msg) => { - let err_nan = f64::from_bits(fetch_error_bits(err_msg)); - perry_runtime::js_promise_reject(promise, err_nan); - } + Err(err) => reject_request_body(promise, err), } promise } @@ -613,13 +601,7 @@ pub unsafe extern "C" fn js_request_form_data(handle: f64) -> *mut perry_runtime } Err(message) => reject_fetch_type_error(promise, message), }, - Err(err_msg) if err_msg == BODY_ALREADY_USED_MESSAGE => { - reject_fetch_type_error(promise, BODY_ALREADY_USED_MESSAGE); - } - Err(err_msg) => { - let err_nan = f64::from_bits(fetch_error_bits(err_msg)); - perry_runtime::js_promise_reject(promise, err_nan); - } + Err(err) => reject_request_body(promise, err), } promise } diff --git a/crates/perry-stdlib/src/fetch/bun_server_bridge.rs b/crates/perry-stdlib/src/fetch/bun_server_bridge.rs index 385901645a0..914e46c56c3 100644 --- a/crates/perry-stdlib/src/fetch/bun_server_bridge.rs +++ b/crates/perry-stdlib/src/fetch/bun_server_bridge.rs @@ -65,6 +65,7 @@ pub unsafe extern "C" fn js_bun_http_request_from_json(snapshot_ptr: *const Stri duplex: "half".to_string(), signal, cached_headers_id: None, + body_error: None, }; gc::ensure_gc_registered(); REQUEST_REGISTRY.lock().unwrap().insert(id, record); diff --git a/crates/perry-stdlib/src/fetch/dispatch.rs b/crates/perry-stdlib/src/fetch/dispatch.rs index ab85e6373a2..3b04b057a78 100644 --- a/crates/perry-stdlib/src/fetch/dispatch.rs +++ b/crates/perry-stdlib/src/fetch/dispatch.rs @@ -120,6 +120,13 @@ pub extern "C" fn js_response_body_init_ptr(value: f64) -> i64 { // as a body) has no byte payload and is not a real heap object — don't // deref it via ToString (that was the crash path); fall through instead. if !perry_runtime::value::addr_class::is_handle_band(addr) { + // An async iterable (e.g. an async generator) is a streamed body, + // not a ToString one. The consumer converts it; see + // `PENDING_FETCH_BODY_ITERABLE`. + if unsafe { crate::streams::has_async_iterator(value) } { + set_pending_fetch_body_iterable(value); + return 0; + } return js_jsvalue_to_string(value) as i64; } } diff --git a/crates/perry-stdlib/src/fetch/gc.rs b/crates/perry-stdlib/src/fetch/gc.rs index 8ab112ed9c4..e8f8fb51735 100644 --- a/crates/perry-stdlib/src/fetch/gc.rs +++ b/crates/perry-stdlib/src/fetch/gc.rs @@ -132,6 +132,8 @@ extern "C" fn scan_fetch_roots_ffi( ctx: *mut c_void, ) { let mut visitor = FfiFetchRootVisitor { visit, ctx }; + // Owned by no handle yet, so a root on every mark kind. + visit_pending_fetch_body_iterable(&mut visitor); if unsafe { perry_ffi_gc_root_visitor_is_full_mark(ctx) } { // Full marking reaches registry edges only through live handles // (`lifecycle::observe`), except for young ids, which are roots. @@ -144,6 +146,7 @@ extern "C" fn scan_fetch_roots_ffi( #[cfg(test)] pub(super) fn scan_fetch_roots(mark: &mut dyn FnMut(f64)) { let mut visitor = perry_runtime::gc::RuntimeRootVisitor::for_copy(mark); + visit_pending_fetch_body_iterable(&mut visitor); scan_fetch_roots_with(&mut visitor); } @@ -167,6 +170,9 @@ pub(super) fn scan_fetch_roots_with(visitor: &mut V) { if let Ok(mut requests) = REQUEST_REGISTRY.lock() { for (_, request) in requests.iter_mut().filter(|(id, _)| lifecycle::owns(**id)) { visitor.visit_nanbox_f64_slot(&mut request.signal); + if let Some(error) = &mut request.body_error { + visitor.visit_nanbox_f64_slot(error); + } } } } diff --git a/crates/perry-stdlib/src/fetch/lifecycle.rs b/crates/perry-stdlib/src/fetch/lifecycle.rs index e70668fe478..b402758ba47 100644 --- a/crates/perry-stdlib/src/fetch/lifecycle.rs +++ b/crates/perry-stdlib/src/fetch/lifecycle.rs @@ -306,6 +306,7 @@ extern "C" fn observe(bits: u64, mark: Mark, ctx: *mut c_void) -> bool { } if let Some(request) = REQUEST_REGISTRY.lock().unwrap().get(&id) { edges.push(request.signal.to_bits()); + edges.extend(request.body_error.map(f64::to_bits)); if let Some(headers) = request.cached_headers_id { edges.push(handle_to_f64(headers).to_bits()); } diff --git a/crates/perry-stdlib/src/fetch/mod.rs b/crates/perry-stdlib/src/fetch/mod.rs index 4ac31bc000a..00ed366f2a9 100644 --- a/crates/perry-stdlib/src/fetch/mod.rs +++ b/crates/perry-stdlib/src/fetch/mod.rs @@ -412,6 +412,11 @@ fn request_headers_snapshot(request: &RequestRecord) -> HeadersStore { thread_local! { static PENDING_FETCH_BODY_STREAM_ID: Cell = const { Cell::new(0) }; + // An async-iterable BodyInit (NaN-boxed bits, 0 = none). The classifier + // only stashes it: converting it runs the iterator's JS, so that happens + // in `take_pending_fetch_body_stream_id`, whose callers must not hold + // unrooted heap pointers across the call. Visited by the fetch GC scanner. + static PENDING_FETCH_BODY_ITERABLE: Cell = const { Cell::new(0) }; // Codegen coerces BodyInit to a StringHeader before js_response_new, so // preserve whether the original value was a string long enough for the // Response constructor to install Fetch's default Content-Type. @@ -427,15 +432,30 @@ pub(super) fn set_pending_fetch_body_content_type(content_type: Option<&'static pub(super) fn reset_pending_fetch_body_init() { PENDING_FETCH_BODY_STREAM_ID.with(|pending| pending.set(0)); + PENDING_FETCH_BODY_ITERABLE.with(|pending| pending.set(0)); PENDING_FETCH_BODY_CONTENT_TYPE.with(|pending| pending.set(None)); } +fn set_pending_fetch_body_iterable(value: f64) { + PENDING_FETCH_BODY_ITERABLE.with(|pending| pending.set(value.to_bits())); +} + +fn visit_pending_fetch_body_iterable(visitor: &mut V) { + PENDING_FETCH_BODY_ITERABLE.with(|pending| { + let mut bits = pending.get(); + if bits != 0 { + visitor.visit_nanbox_u64_slot(&mut bits); + pending.set(bits); + } + }); +} + fn take_pending_fetch_body_content_type() -> Option<&'static str> { PENDING_FETCH_BODY_CONTENT_TYPE.with(|pending| pending.replace(None)) } fn take_pending_fetch_body_stream_id() -> Option { - PENDING_FETCH_BODY_STREAM_ID.with(|pending| { + let stream = PENDING_FETCH_BODY_STREAM_ID.with(|pending| { let id = pending.get(); pending.set(0); if id != 0 && crate::streams::js_stream_handle_kind(id) == 1 { @@ -443,9 +463,22 @@ fn take_pending_fetch_body_stream_id() -> Option { } else { None } + }); + let iterable = PENDING_FETCH_BODY_ITERABLE.with(|pending| pending.replace(0)); + stream.or_else(|| { + (iterable != 0).then(|| unsafe { + crate::streams::js_readable_stream_from_iterable(f64::from_bits(iterable)) as usize + }) }) } +/// Drain a body stream, returning the error it ended with (if any) alongside +/// the bytes it produced before that. +fn drain_body_stream(stream_id: usize) -> (Vec, Option) { + let bytes = crate::streams::drain_readable_into_bytes(stream_id); + (bytes, crate::streams::readable_stream_error(stream_id)) +} + /// Extract the registry id from a Web Fetch handle f64 value. /// /// Web Fetch handles (Request / Response / Headers / Blob) are returned by @@ -731,6 +764,9 @@ pub unsafe extern "C" fn js_fetch_with_options( // `Request` object and call `fetch(request, init)`; its handle id lands in // the `url_ptr` slot. Recover url/method/body/headers from the Request // registry so the request is dispatched (`init` members override). + let url = string_from_header(url_ptr); + let method = string_from_header(method_ptr); + let headers_json = string_from_header(headers_json_ptr); let form_data_body = body_metadata::serialize_form_data(body_ptr as usize); let form_data_content_type = form_data_body .as_ref() @@ -738,14 +774,18 @@ pub unsafe extern "C" fn js_fetch_with_options( let body_bytes = form_data_body .map(|(body, _)| body) .or_else(|| fetch_request_body_bytes(body_ptr)); + // Last: may run an async-iterable body's JS (the strings above are owned). + let body_bytes = take_pending_fetch_body_stream_id() + .map(crate::streams::drain_readable_into_bytes) + .or(body_bytes); let mut inputs = match request_handle::resolve_fetch_inputs( - string_from_header(url_ptr), - string_from_header(method_ptr), + url, + method, // Read the body as raw bytes (binary bodies probe the buffer/typed-array // registry first) so a Buffer/Uint8Array body isn't corrupted by a lossy // StringHeader read (#5757). body_bytes, - string_from_header(headers_json_ptr), + headers_json, url_ptr as usize, pending_redirect, ) { @@ -1088,6 +1128,9 @@ struct RequestRecord { /// repeat reads return the same handle (preserves `req.headers === /// req.headers`). Mirrors `FetchResponse::cached_headers_id` (#1649). cached_headers_id: Option, + /// The error the body's stream ended with; consuming the body rejects + /// with it. A GC root, visited with `signal`. + body_error: Option, } static HEADERS_REGISTRY: std::sync::LazyLock>> = @@ -1539,6 +1582,7 @@ pub extern "C" fn js_request_clone(handle: f64) -> f64 { duplex: req.duplex.clone(), signal: req.signal, cached_headers_id: None, + body_error: req.body_error, }) }) }; @@ -1556,19 +1600,42 @@ pub extern "C" fn js_request_clone(handle: f64) -> f64 { /// Read and consume a request's stored body. Bodiless requests are reusable and /// resolve to an empty body, matching Node's Fetch Body behavior. -fn consume_request_body(handle: f64) -> Result, &'static str> { +fn consume_request_body(handle: f64) -> Result, RequestBodyError> { let id = handle_id(handle); let mut guard = REQUEST_REGISTRY.lock().unwrap(); - let req = guard.get_mut(&id).ok_or("Invalid request handle")?; + let req = guard + .get_mut(&id) + .ok_or(RequestBodyError::Message("Invalid request handle"))?; let body = match &req.body { Some(body) => body.clone(), None => return Ok(Vec::new()), }; if req.body_used { - return Err(BODY_ALREADY_USED_MESSAGE); + return Err(RequestBodyError::Message(BODY_ALREADY_USED_MESSAGE)); } req.body_used = true; - Ok(body) + match req.body_error { + Some(error) => Err(RequestBodyError::Thrown(error)), + None => Ok(body), + } +} + +enum RequestBodyError { + Message(&'static str), + /// The body's stream ended with this error; reject with it unchanged. + Thrown(f64), +} + +unsafe fn reject_request_body(promise: *mut perry_runtime::Promise, err: RequestBodyError) { + match err { + RequestBodyError::Message(msg) if msg == BODY_ALREADY_USED_MESSAGE => { + reject_fetch_type_error(promise, BODY_ALREADY_USED_MESSAGE); + } + RequestBodyError::Message(msg) => { + perry_runtime::js_promise_reject(promise, f64::from_bits(fetch_error_bits(msg))); + } + RequestBodyError::Thrown(error) => perry_runtime::js_promise_reject(promise, error), + } } /// request.text() -> Promise. Mirrors `js_fetch_response_text`: the @@ -1586,13 +1653,7 @@ pub unsafe extern "C" fn js_request_text(handle: f64) -> *mut perry_runtime::Pro let result_nan = f64::from_bits(JSValue::string_ptr(result_str).bits()); perry_runtime::js_promise_resolve(promise, result_nan); } - Err(err_msg) if err_msg == BODY_ALREADY_USED_MESSAGE => { - reject_fetch_type_error(promise, BODY_ALREADY_USED_MESSAGE); - } - Err(err_msg) => { - let err_nan = f64::from_bits(fetch_error_bits(err_msg)); - perry_runtime::js_promise_reject(promise, err_nan); - } + Err(err) => reject_request_body(promise, err), } promise } @@ -1605,13 +1666,8 @@ pub unsafe extern "C" fn js_request_json(handle: f64) -> *mut perry_runtime::Pro let promise = perry_runtime::js_promise_new_cross_thread(); let body = match consume_request_body(handle) { Ok(b) => b, - Err(err_msg) if err_msg == BODY_ALREADY_USED_MESSAGE => { - reject_fetch_type_error(promise, BODY_ALREADY_USED_MESSAGE); - return promise; - } - Err(err_msg) => { - let err_nan = f64::from_bits(fetch_error_bits(err_msg)); - perry_runtime::js_promise_reject(promise, err_nan); + Err(err) => { + reject_request_body(promise, err); return promise; } }; @@ -1634,13 +1690,8 @@ pub unsafe extern "C" fn js_request_array_buffer(handle: f64) -> *mut perry_runt let promise = perry_runtime::js_promise_new_cross_thread(); let body = match consume_request_body(handle) { Ok(b) => b, - Err(err_msg) if err_msg == BODY_ALREADY_USED_MESSAGE => { - reject_fetch_type_error(promise, BODY_ALREADY_USED_MESSAGE); - return promise; - } - Err(err_msg) => { - let err_nan = f64::from_bits(fetch_error_bits(err_msg)); - perry_runtime::js_promise_reject(promise, err_nan); + Err(err) => { + reject_request_body(promise, err); return promise; } }; diff --git a/crates/perry-stdlib/src/fetch/request_copy.rs b/crates/perry-stdlib/src/fetch/request_copy.rs index 8e50ac593ea..c0ffb721166 100644 --- a/crates/perry-stdlib/src/fetch/request_copy.rs +++ b/crates/perry-stdlib/src/fetch/request_copy.rs @@ -24,6 +24,12 @@ pub unsafe extern "C" fn js_request_new_from_input(input: f64, init: f64) -> f64 return js_request_new_from_init(url, init_root.get_nanbox_f64()); }; let signal_root = scope.root_nanbox_f64(request.signal); + let body_error_root = scope.root_nanbox_f64( + request + .body_error + .unwrap_or_else(|| f64::from_bits(TAG_UNDEFINED)), + ); + let mut body_error = request.body_error.is_some(); request.headers = request_headers_snapshot(&request); request.cached_headers_id = None; request.body_used = false; @@ -122,6 +128,7 @@ pub unsafe extern "C" fn js_request_new_from_input(input: f64, init: f64) -> f64 throw_fetch_type_error("Request with GET/HEAD method cannot have body."); } if has_override_body { + body_error = false; let form_data = body_metadata::serialize_form_data(handle_id(body_root.get_nanbox_f64())); let content_type = if let Some((bytes, content_type)) = form_data { request.body = Some(bytes); @@ -140,9 +147,16 @@ pub unsafe extern "C" fn js_request_new_from_input(input: f64, init: f64) -> f64 } else { dispatch::body_bytes_from_header(ptr) }; - request.body = stream - .map(crate::streams::drain_readable_into_bytes) - .or(bytes); + request.body = match stream.map(drain_body_stream) { + Some((drained, error)) => { + if let Some(error) = error { + body_error = true; + body_error_root.set_nanbox_f64(error); + } + Some(drained) + } + None => bytes, + }; content_type }; if let Some(content_type) = content_type { @@ -169,6 +183,7 @@ pub unsafe extern "C" fn js_request_new_from_input(input: f64, init: f64) -> f64 } } request.signal = signal_root.get_nanbox_f64(); + request.body_error = body_error.then(|| body_error_root.get_nanbox_f64()); let id = alloc_fetch_handle_id(); gc::ensure_gc_registered(); REQUEST_REGISTRY.lock().unwrap().insert(id, request); diff --git a/crates/perry-stdlib/src/fetch/request_ctor.rs b/crates/perry-stdlib/src/fetch/request_ctor.rs index 17fc41d30e6..7a4fc0eb11a 100644 --- a/crates/perry-stdlib/src/fetch/request_ctor.rs +++ b/crates/perry-stdlib/src/fetch/request_ctor.rs @@ -38,6 +38,19 @@ pub unsafe extern "C" fn js_request_new( throw_fetch_type_error(&format!("'{raw_method}' HTTP method is unsupported.")); } let method = normalize_method(&raw_method); + // Copy every string argument now: taking or draining the body stream below + // can run JS, which may move heap strings. Root `signal` for the same reason. + let referrer = string_from_header(referrer_ptr).unwrap_or_else(|| "about:client".to_string()); + let referrer_policy = string_from_header(referrer_policy_ptr).unwrap_or_default(); + let mode = string_from_header(mode_ptr).unwrap_or_else(|| "cors".to_string()); + let credentials = + string_from_header(credentials_ptr).unwrap_or_else(|| "same-origin".to_string()); + let cache = string_from_header(cache_ptr).unwrap_or_else(|| "default".to_string()); + let redirect = string_from_header(redirect_ptr).unwrap_or_else(|| "follow".to_string()); + let integrity = string_from_header(integrity_ptr).unwrap_or_default(); + let duplex = string_from_header(duplex_ptr).unwrap_or_else(|| "half".to_string()); + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let signal_root = scope.root_nanbox_f64(signal); // A Buffer / Uint8Array / typed-array / ArrayBuffer body reaches us as a // BufferHeader/TypedArrayHeader pointer (codegen ran the value through // `js_get_string_pointer_unified`), NOT a StringHeader — the same for both @@ -85,9 +98,11 @@ pub unsafe extern "C" fn js_request_new( { throw_fetch_type_error("Request with GET/HEAD method cannot have body."); } - let body = pending_stream_id - .map(crate::streams::drain_readable_into_bytes) - .or(non_stream_body); + let (body, body_error) = match pending_stream_id.map(drain_body_stream) { + Some((bytes, error)) => (Some(bytes), error), + None => (non_stream_body, None), + }; + let body_error = body_error.map(|error| scope.root_nanbox_f64(error)); let headers_id_in = handle_id(headers_handle); let mut headers = if headers_id_in != 0 { HEADERS_REGISTRY @@ -110,7 +125,7 @@ pub unsafe extern "C" fn js_request_new( // lock: the scanner takes that same lock during a collection on this // thread, and a collection triggered by the allocation under the guard // would deadlock. - let signal = body_metadata::signal_or_default(signal); + let signal = body_metadata::signal_or_default(signal_root.get_nanbox_f64()); let id = alloc_fetch_handle_id(); let record = RequestRecord { url, @@ -119,18 +134,18 @@ pub unsafe extern "C" fn js_request_new( body_used: false, headers, destination: String::new(), - referrer: string_from_header(referrer_ptr).unwrap_or_else(|| "about:client".to_string()), - referrer_policy: string_from_header(referrer_policy_ptr).unwrap_or_default(), - mode: string_from_header(mode_ptr).unwrap_or_else(|| "cors".to_string()), - credentials: string_from_header(credentials_ptr) - .unwrap_or_else(|| "same-origin".to_string()), - cache: string_from_header(cache_ptr).unwrap_or_else(|| "default".to_string()), - redirect: string_from_header(redirect_ptr).unwrap_or_else(|| "follow".to_string()), - integrity: string_from_header(integrity_ptr).unwrap_or_default(), + referrer, + referrer_policy, + mode, + credentials, + cache, + redirect, + integrity, keepalive: body_metadata::bool_from_js(keepalive), - duplex: string_from_header(duplex_ptr).unwrap_or_else(|| "half".to_string()), + duplex, signal, cached_headers_id: None, + body_error: body_error.map(|error| error.get_nanbox_f64()), }; super::gc::ensure_gc_registered(); REQUEST_REGISTRY.lock().unwrap().insert(id, record); diff --git a/crates/perry-stdlib/src/fetch/response_ctor.rs b/crates/perry-stdlib/src/fetch/response_ctor.rs index a0eb98d3d45..c8da948b62a 100644 --- a/crates/perry-stdlib/src/fetch/response_ctor.rs +++ b/crates/perry-stdlib/src/fetch/response_ctor.rs @@ -90,6 +90,10 @@ pub unsafe extern "C" fn js_response_new( headers_handle: f64, ) -> f64 { let _fetch_roots = lifecycle::pin_handles(&[status, headers_handle]); + // Taking the stream can run an async-iterable body's JS: root the one + // string argument still read afterwards. + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let status_text = (!status_text_ptr.is_null()).then(|| scope.root_string_ptr(status_text_ptr)); let body_stream_id = take_pending_fetch_body_stream_id(); // Consume before validation so a throwing constructor cannot leak body // metadata into the next Response construction on this thread. @@ -98,6 +102,7 @@ pub unsafe extern "C" fn js_response_new( let body_opt = dispatch::body_bytes_from_header(body_ptr); let body_present = body_opt.is_some() || body_stream_id.is_some(); let body = body_opt.unwrap_or_default(); + let status_text_ptr = status_text.map_or(std::ptr::null(), |h| h.get_raw_const_ptr()); let (status_u16, status_text) = response_init(status, status_text_ptr, body_present); let headers_id = handle_id(headers_handle); let registered = (headers_id != 0) diff --git a/crates/perry-stdlib/src/streams.rs b/crates/perry-stdlib/src/streams.rs index 4dcd44f92b6..9be233d55bc 100644 --- a/crates/perry-stdlib/src/streams.rs +++ b/crates/perry-stdlib/src/streams.rs @@ -580,6 +580,17 @@ pub(crate) const STREAM_HANDLE_ID_END: usize = perry_runtime::value::addr_class: static READABLE_STREAMS: std::sync::LazyLock>> = std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); +/// The error a readable stream ended with, if any. A stream built from an +/// iterable that threw keeps it in `pending_error_after_chunks` until a reader +/// has popped the queued chunks, so check both. +pub(crate) fn readable_stream_error(stream_id: usize) -> Option { + let g = READABLE_STREAMS.lock().unwrap(); + let s = g.get(&stream_id)?; + match s.state { + ReadableState::Errored => Some(f64::from_bits(s.error_value)), + _ => s.pending_error_after_chunks.map(f64::from_bits), + } +} static WRITABLE_STREAMS: std::sync::LazyLock>> = std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); static TRANSFORM_STREAMS: std::sync::LazyLock>> = @@ -1705,16 +1716,23 @@ unsafe fn is_callable_value(value: f64) -> bool { raw >= 0x10000 && perry_runtime::closure::is_closure_ptr(raw as usize) } -unsafe fn call_symbol_async_iterator(value: f64) -> Option { +unsafe fn async_iterator_method(value: f64) -> Option { let sym = perry_runtime::symbol::well_known_symbol("asyncIterator"); if sym.is_null() { return None; } let sym_value = f64::from_bits(JSValue::pointer(sym as *const u8).bits()); let method = perry_runtime::symbol::js_object_get_symbol_property(value, sym_value); - if !is_callable_value(method) { - return None; - } + is_callable_value(method).then_some(method) +} + +/// Whether `value` has a callable `Symbol.asyncIterator` (without calling it). +pub(crate) unsafe fn has_async_iterator(value: f64) -> bool { + async_iterator_method(value).is_some() +} + +unsafe fn call_symbol_async_iterator(value: f64) -> Option { + let method = async_iterator_method(value)?; let iterator = with_implicit_this(value, || js_native_call_value(method, std::ptr::null(), 0)); if iterator.to_bits() == TAG_UNDEFINED { None @@ -1796,29 +1814,31 @@ unsafe fn chunks_from_async_iterable(value: f64) -> Option { } else { return None; }; - let done_key = js_string_from_bytes(b"done".as_ptr(), 4); - let value_key = js_string_from_bytes(b"value".as_ptr(), 5); + // Each `next()` runs JS, which can move heap values: keep everything we + // hold across it in runtime handles. + let scope = perry_runtime::gc::RuntimeHandleScope::new(); + let iterator = scope.root_nanbox_f64(iterator); + let done_key = scope.root_string_ptr(js_string_from_bytes(b"done".as_ptr(), 4)); + let value_key = scope.root_string_ptr(js_string_from_bytes(b"value".as_ptr(), 5)); let mut chunks = Vec::new(); + let finish = + |chunks: &[perry_runtime::gc::RuntimeHandle<'_>], error: Option| ReadableFromSource { + chunks: perry_runtime::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(chunks) + .into_iter() + .map(f64::to_bits) + .collect(), + error, + }; for _ in 0..100_000 { - let step = match try_call_iterator_next(iterator) { + let step = match try_call_iterator_next(iterator.get_nanbox_f64()) { Ok(Some(step)) => step, Ok(None) => break, - Err(reason) => { - return Some(ReadableFromSource { - chunks, - error: Some(reason), - }); - } + Err(reason) => return Some(finish(&chunks, Some(reason))), }; let step_result = match await_maybe_promise(step) { SettledValue::Fulfilled(result) => result, - SettledValue::Rejected(reason) => { - return Some(ReadableFromSource { - chunks, - error: Some(reason), - }); - } + SettledValue::Rejected(reason) => return Some(finish(&chunks, Some(reason))), SettledValue::Pending => break, }; let result_ptr = js_nanbox_get_pointer(step_result); @@ -1826,16 +1846,16 @@ unsafe fn chunks_from_async_iterable(value: f64) -> Option { break; } let result_obj = result_ptr as *const ObjectHeader; - let done_val = js_object_get_field_by_name(result_obj, done_key); + let done_val = js_object_get_field_by_name(result_obj, done_key.get_raw_const_ptr()); let done = f64::from_bits(done_val.bits()); if perry_runtime::value::js_is_truthy(done) != 0 { break; } - let item = js_object_get_field_by_name(result_obj, value_key); - chunks.push(item.bits()); + let item = js_object_get_field_by_name(result_obj, value_key.get_raw_const_ptr()); + chunks.push(scope.root_nanbox_u64(item.bits())); } - Some(ReadableFromSource::closed(chunks)) + Some(finish(&chunks, None)) } /// `ReadableStream.from(iterable)` (Node 20+, #1645) — build a Web diff --git a/crates/perry-stdlib/src/streams/subclass.rs b/crates/perry-stdlib/src/streams/subclass.rs index f19c4057517..32eaebdf09d 100644 --- a/crates/perry-stdlib/src/streams/subclass.rs +++ b/crates/perry-stdlib/src/streams/subclass.rs @@ -389,6 +389,24 @@ pub unsafe extern "C" fn js_transform_stream_subclass_init( f64::from_bits(TAG_UNDEFINED) } +/// A Fetch body chunk's bytes. Unlike `read_bytes_from_chunk`, a string chunk +/// (valid in an async-iterable body) is UTF-8 encoded rather than dropped. +unsafe fn body_chunk_bytes(chunk_bits: u64) -> Option> { + if JSValue::from_bits(chunk_bits).is_any_string() { + let mut scratch = [0u8; perry_runtime::value::SHORT_STRING_MAX_LEN]; + let (ptr, len) = perry_runtime::string::str_bytes_from_jsvalue( + f64::from_bits(chunk_bits), + &mut scratch, + )?; + return Some(if ptr.is_null() { + Vec::new() + } else { + std::slice::from_raw_parts(ptr, len as usize).to_vec() + }); + } + read_bytes_from_chunk(chunk_bits) +} + /// Drain a ReadableStream's body to bytes, DRIVING its `pull` source. Used by /// `new Response(stream)` / `new Request(url, { body: stream })` to materialize /// the body at construction time. @@ -435,7 +453,7 @@ pub fn drain_readable_into_bytes(stream_id: usize) -> Vec { let mut got_chunk = false; for chunk in chunks { unsafe { - if let Some(bytes) = read_bytes_from_chunk(chunk) { + if let Some(bytes) = body_chunk_bytes(chunk) { out.extend_from_slice(&bytes); got_chunk = true; } diff --git a/test-parity/node_suite_baseline.json b/test-parity/node_suite_baseline.json index d9cc100c26b..e9be511dfba 100644 --- a/test-parity/node_suite_baseline.json +++ b/test-parity/node_suite_baseline.json @@ -68,8 +68,8 @@ "total": 69 }, "fetch": { - "pass": 13, - "total": 13 + "pass": 16, + "total": 16 }, "fs": { "pass": 168, From cc526133ccc92bb53f334ee3be0e719d07d823f9 Mon Sep 17 00:00:00 2001 From: Sergi Gonzalez <31130069+TheHypnoo@users.noreply.github.com> Date: Sun, 27 Sep 2026 18:29:34 +0200 Subject: [PATCH 3/3] docs(changelog): add fragment for async-iterable fetch bodies --- .../11552-request-async-iterable-body.md | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) create mode 100644 changelog.d/11552-request-async-iterable-body.md diff --git a/changelog.d/11552-request-async-iterable-body.md b/changelog.d/11552-request-async-iterable-body.md new file mode 100644 index 00000000000..6f57f43d769 --- /dev/null +++ b/changelog.d/11552-request-async-iterable-body.md @@ -0,0 +1,18 @@ +Fixed Web Fetch bodies built from async iterables. `new Request(url, { body: +asyncGenerator(), duplex: "half" })` and `new Response(asyncGenerator())` +stringified the body to `"[object AsyncGenerator]"`, and `fetch(url, { body: +asyncGenerator() })` sent an empty body. Astro's `@astrojs/node` adapter passes +every POST body this way, so `request.json()` threw `SyntaxError`. + +The BodyInit classifier (`js_response_body_init_ptr`) now stashes an async +iterable in a GC-scanned pending slot, and `take_pending_fetch_body_stream_id` +converts it with `js_readable_stream_from_iterable`, so `Request`, `Request` +copies, `Response` and `fetch()` all pick it up. The conversion runs the +iterator's JS, so those consumers copy or root their raw arguments first, and +`ReadableStream.from` now keeps the iterator and collected chunks in runtime +handles. A `Request` body whose iterable throws rejects `text()`/`json()`/… +with the original error. String chunks are UTF-8 encoded in the Fetch body +drain. + +Tests: `test-parity/node-suite/fetch/request/async-iterable-body.ts`, +`async-iterable-body-send.ts`, `mutated-init-local.ts`.