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`. 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/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); 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,