From 2c8b2df82bebe1f0620e371161650af7bfe447ce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Thu, 11 Jun 2026 07:31:50 +0200 Subject: [PATCH] feat(streams): BYOB readers + real ByteLengthQueuingStrategy accounting (#4915) Implements the Web Streams residuals left after #237/#1545: - ReadableStreamBYOBReader: getReader({ mode: "byob" }) and new ReadableStreamBYOBReader(stream) mint readers whose read(view) fills the caller-supplied TypedArray/DataView/Buffer (draining queued byte chunks across chunk boundaries, remainder requeued); done resolves with an empty view per spec. - ReadableByteStreamController.byobRequest: { view, respond(bytesWritten), respondWithNewView(view) } while a read(view) is parked, null otherwise; pull sources write straight into the caller's buffer. - Real QueuingStrategy accounting: per-chunk size() results (or byteLength on byte streams) are summed into desiredSize for ReadableStream, WritableStream, and TransformStream; the WritableStream/TransformStream constructors now accept strategy objects (number-or-object parsing in streams/strategy.rs). - Byte streams default to highWaterMark 0 per spec (no eager pull at construction; ShouldCallPull fires on parked read requests). - Spec-shaped errors: TypeError for BYOB mode on non-byte streams / invalid views / released readers, RangeError for out-of-range bytesWritten; removed the dead js_streams_throw_*_not_implemented stubs (perry-ext-streams + stdlib) and the three #4915 manifest stub notes; regenerated API docs. Verified byte-for-byte against node --experimental-strip-types (test-files/test_parity_streams_byob.ts) and guarded by crates/perry/tests/issue_4915_streams_byob.rs. Streams regression sweep (pipe/user-source/blob/readable/subclass/consumers/stream_web) is byte-identical to origin/main. Fixes #4915 --- crates/perry-api-manifest/src/entries.rs | 22 +- .../tests/stub_inventory.rs | 5 +- crates/perry-codegen/src/ext_registry.rs | 4 + .../perry-codegen/src/lower_call/builtin.rs | 65 +- .../src/lower_call/options/fetch.rs | 23 + .../src/runtime_decls/strings_part2.rs | 16 +- crates/perry-ext-streams/src/lib.rs | 24 +- .../perry-hir/src/destructuring/var_decl.rs | 11 + .../perry-hir/src/lower/expr_call/stream.rs | 1 + crates/perry-hir/src/lower/expr_member.rs | 9 +- crates/perry-hir/src/lower/expr_new.rs | 1 + crates/perry-stdlib/src/streams.rs | 240 +++++--- crates/perry-stdlib/src/streams/byob.rs | 578 ++++++++++++++++++ crates/perry-stdlib/src/streams/pipe.rs | 2 +- crates/perry-stdlib/src/streams/strategy.rs | 103 ++++ crates/perry-stdlib/src/streams/subclass.rs | 15 +- crates/perry-stdlib/src/streams/tests.rs | 2 + crates/perry-stdlib/src/streams/transform.rs | 73 ++- crates/perry-stdlib/src/streams/writable.rs | 50 +- crates/perry/tests/issue_4915_streams_byob.rs | 114 ++++ docs/api/perry.d.ts | 8 +- docs/src/api/reference.md | 8 +- test-files/test_parity_streams_byob.ts | 131 ++++ 23 files changed, 1306 insertions(+), 199 deletions(-) create mode 100644 crates/perry-stdlib/src/streams/byob.rs create mode 100644 crates/perry-stdlib/src/streams/strategy.rs create mode 100644 crates/perry/tests/issue_4915_streams_byob.rs create mode 100644 test-files/test_parity_streams_byob.ts diff --git a/crates/perry-api-manifest/src/entries.rs b/crates/perry-api-manifest/src/entries.rs index 3a8c2f8c61f..908bc6ccea7 100644 --- a/crates/perry-api-manifest/src/entries.rs +++ b/crates/perry-api-manifest/src/entries.rs @@ -4170,10 +4170,12 @@ pub static API_MANIFEST: &[ApiEntry] = &[ property("stream/web", "default"), class("stream/web", "ReadableStream"), class("stream/web", "ReadableStreamDefaultReader"), - class("stream/web", "ReadableStreamBYOBReader") - .stub_note("constructor/use throws: not yet implemented (#4915)"), - class("stream/web", "ReadableStreamBYOBRequest") - .stub_note("constructor/use throws: not yet implemented (#4915)"), + // #4915: BYOB readers are real — `new ReadableStreamBYOBReader(stream)` / + // `getReader({ mode: "byob" })` mint a reader whose `read(view)` fills the + // caller-supplied buffer; the byte-stream controller's `byobRequest` + // exposes `view` / `respond(bytesWritten)` / `respondWithNewView(view)`. + class("stream/web", "ReadableStreamBYOBReader"), + class("stream/web", "ReadableStreamBYOBRequest"), class("stream/web", "ReadableByteStreamController"), class("stream/web", "ReadableStreamDefaultController"), class("stream/web", "TransformStream"), @@ -4181,8 +4183,9 @@ pub static API_MANIFEST: &[ApiEntry] = &[ class("stream/web", "WritableStream"), class("stream/web", "WritableStreamDefaultWriter"), class("stream/web", "WritableStreamDefaultController"), - class("stream/web", "ByteLengthQueuingStrategy") - .stub_note("constructor/use throws: not yet implemented (#4915)"), + // #4915: real byteLength accounting — per-chunk size() results are summed + // into desiredSize for ReadableStream/WritableStream/TransformStream. + class("stream/web", "ByteLengthQueuingStrategy"), class("stream/web", "CountQueuingStrategy"), class("stream/web", "TextEncoderStream"), class("stream/web", "TextDecoderStream"), @@ -5125,9 +5128,10 @@ pub static API_MANIFEST: &[ApiEntry] = &[ class("streams", "TextEncoder"), class("streams", "TextDecoder"), class("streams", "DecompressionStream"), - // node:stream/web QueuingStrategy classes (#1545). - class("streams", "ByteLengthQueuingStrategy") - .stub_note("constructor/use throws: not yet implemented (#4915)"), + // node:stream/web QueuingStrategy classes (#1545). #4915: the + // constructor lowers through the same stdlib builtin arm as the + // node:stream/web form, with real byteLength desiredSize accounting. + class("streams", "ByteLengthQueuingStrategy"), class("streams", "CountQueuingStrategy"), // --- node:http server (issue #577) --- method("http", "createServer", false, None), diff --git a/crates/perry-api-manifest/tests/stub_inventory.rs b/crates/perry-api-manifest/tests/stub_inventory.rs index 4cde67ec404..1e6c12b524d 100644 --- a/crates/perry-api-manifest/tests/stub_inventory.rs +++ b/crates/perry-api-manifest/tests/stub_inventory.rs @@ -69,7 +69,10 @@ fn stub_inventory_matches_known_clusters() { // listening port via SO_REUSEPORT binds + a fork-IPC 'listening' // round-trip; SCHED_RR fd-passing + shared ephemeral `listen(0)` // port are tracked in #4962 and are policy fidelity, not lies. - ("#4915", 4), // BYOB readers + ByteLengthQueuingStrategy (throw) + // #4915 (BYOB readers + ByteLengthQueuingStrategy) intentionally + // absent: read(view), controller.byobRequest respond/ + // respondWithNewView, and real byteLength desiredSize accounting + // landed (perry-stdlib/src/streams/byob.rs). ("#4916", 2), // v8 get/writeHeapSnapshot (empty graph) ("#4917", 18), // stdlib adapters: zlib(11) + http.Agent(3) + worker ref/unref(2) + mongodb(1) + backoff(1) ]; diff --git a/crates/perry-codegen/src/ext_registry.rs b/crates/perry-codegen/src/ext_registry.rs index c0fc6c7ab08..c640a344177 100644 --- a/crates/perry-codegen/src/ext_registry.rs +++ b/crates/perry-codegen/src/ext_registry.rs @@ -138,6 +138,10 @@ const FFI_REGISTRY: &[(&str, OwnerKind)] = &[ ("js_writer_ready", OwnerKind::Stdlib { feature: Some("bundled-streams") }), ("js_writer_desired_size", OwnerKind::Stdlib { feature: Some("bundled-streams") }), ("js_reader_read", OwnerKind::Stdlib { feature: Some("bundled-streams") }), + // #4915: BYOB reader surface. + ("js_reader_read_with_view", OwnerKind::Stdlib { feature: Some("bundled-streams") }), + ("js_readable_stream_get_byob_reader", OwnerKind::Stdlib { feature: Some("bundled-streams") }), + ("js_readable_stream_controller_byob_request", OwnerKind::Stdlib { feature: Some("bundled-streams") }), ("js_reader_release_lock", OwnerKind::Stdlib { feature: Some("bundled-streams") }), ("js_reader_closed", OwnerKind::Stdlib { feature: Some("bundled-streams") }), ("js_reader_cancel", OwnerKind::Stdlib { feature: Some("bundled-streams") }), diff --git a/crates/perry-codegen/src/lower_call/builtin.rs b/crates/perry-codegen/src/lower_call/builtin.rs index 82d0bda8959..c255657b554 100644 --- a/crates/perry-codegen/src/lower_call/builtin.rs +++ b/crates/perry-codegen/src/lower_call/builtin.rs @@ -1274,20 +1274,11 @@ pub(super) fn lower_builtin_new( } } if args.len() >= 2 { - if let Some(qprops) = extract_options_fields(ctx, &args[1]) { - for (k, vexpr) in &qprops { - if k == "highWaterMark" { - hwm = lower_expr(ctx, vexpr)?; - } - } - } else { - let strategy = lower_expr(ctx, &args[1])?; - hwm = ctx.block().call( - DOUBLE, - "js_streams_strategy_high_water_mark", - &[(DOUBLE, &strategy)], - ); - } + // #4915: pass the whole strategy value through — the runtime + // accepts a plain highWaterMark number or a strategy object + // (e.g. ByteLengthQueuingStrategy) and reads highWaterMark + + // size() from it. + hwm = lower_expr(ctx, &args[1])?; } if let Some(sink) = sink_object { let h = ctx.block().call( @@ -1340,20 +1331,26 @@ pub(super) fn lower_builtin_new( transformer_object = Some(lower_expr(ctx, &args[0])?); } } + // #4915: writableStrategy (arg 1) / readableStrategy (arg 2) — + // each may be a plain highWaterMark number or a strategy object; + // the runtime parses either form. + let mut writable_strategy = hwm; + let mut readable_strategy = double_literal(1.0); if args.len() >= 2 { - if let Some(qprops) = extract_options_fields(ctx, &args[1]) { - for (k, vexpr) in &qprops { - if k == "highWaterMark" { - hwm = lower_expr(ctx, vexpr)?; - } - } - } + writable_strategy = lower_expr(ctx, &args[1])?; + } + if args.len() >= 3 { + readable_strategy = lower_expr(ctx, &args[2])?; } if let Some(transformer) = transformer_object { let h = ctx.block().call( DOUBLE, "js_transform_stream_new_from_transformer_object", - &[(DOUBLE, &transformer), (DOUBLE, &hwm)], + &[ + (DOUBLE, &transformer), + (DOUBLE, &writable_strategy), + (DOUBLE, &readable_strategy), + ], ); return Ok(Some(h)); } @@ -1364,7 +1361,8 @@ pub(super) fn lower_builtin_new( (DOUBLE, &start), (DOUBLE, &transform), (DOUBLE, &flush), - (DOUBLE, &hwm), + (DOUBLE, &writable_strategy), + (DOUBLE, &readable_strategy), ], ); Ok(Some(h)) @@ -1420,6 +1418,27 @@ pub(super) fn lower_builtin_new( Ok(Some(h)) } + // #4915: `new ReadableStreamBYOBReader(stream)` — equivalent to + // `stream.getReader({ mode: "byob" })`. The runtime validates that + // the argument is a byte stream (TypeError otherwise) and returns a + // reader handle whose `read(view)` fills the caller's buffer. + "ReadableStreamBYOBReader" => { + let stream = if !args.is_empty() { + lower_expr(ctx, &args[0])? + } else { + double_literal(f64::from_bits(crate::nanbox::TAG_UNDEFINED)) + }; + for a in args.iter().skip(1) { + let _ = lower_expr(ctx, a)?; + } + let h = ctx.block().call( + DOUBLE, + "js_readable_stream_get_byob_reader", + &[(DOUBLE, &stream)], + ); + Ok(Some(h)) + } + // node:stream/web QueuingStrategy classes (#1545). Both take a single // `{ highWaterMark }` options object; the runtime reads // `opts.highWaterMark` and builds a `{ highWaterMark, size }` object. diff --git a/crates/perry-codegen/src/lower_call/options/fetch.rs b/crates/perry-codegen/src/lower_call/options/fetch.rs index db1f5f4bac6..72993045598 100644 --- a/crates/perry-codegen/src/lower_call/options/fetch.rs +++ b/crates/perry-codegen/src/lower_call/options/fetch.rs @@ -964,6 +964,17 @@ pub(in crate::lower_call) fn lower_fetch_native_method( ); return Ok(Some(v)); } + // #4915: byte-stream controller's BYOB request — `{ view, + // respond, respondWithNewView }` while a BYOB read is parked, + // null otherwise. + "byobRequest" => { + let v = ctx.block().call( + DOUBLE, + "js_readable_stream_controller_byob_request", + &[(DOUBLE, &recv_handle)], + ); + return Ok(Some(v)); + } _ => return Ok(None), } } @@ -971,6 +982,18 @@ pub(in crate::lower_call) fn lower_fetch_native_method( if module == "readable_stream_reader" { let recv_handle = lower_expr(ctx, recv)?; match method { + // #4915: `read(view)` is the BYOB form — the runtime fills the + // caller-supplied view (default readers ignore the argument). + "read" if !args.is_empty() => { + let view = lower_expr(ctx, &args[0])?; + let blk = ctx.block(); + let promise = blk.call( + I64, + "js_reader_read_with_view", + &[(DOUBLE, &recv_handle), (DOUBLE, &view)], + ); + return Ok(Some(nanbox_pointer_inline(blk, &promise))); + } "read" => { let blk = ctx.block(); let promise = blk.call(I64, "js_reader_read", &[(DOUBLE, &recv_handle)]); diff --git a/crates/perry-codegen/src/runtime_decls/strings_part2.rs b/crates/perry-codegen/src/runtime_decls/strings_part2.rs index 7aa296d03e4..abfa75916b3 100644 --- a/crates/perry-codegen/src/runtime_decls/strings_part2.rs +++ b/crates/perry-codegen/src/runtime_decls/strings_part2.rs @@ -998,6 +998,13 @@ pub(crate) fn declare_phase_b_strings_part2(module: &mut LlModule) { DOUBLE, &[DOUBLE, DOUBLE], ); + // #4915: BYOB reader surface. + module.declare_function("js_readable_stream_get_byob_reader", DOUBLE, &[DOUBLE]); + module.declare_function( + "js_readable_stream_controller_byob_request", + DOUBLE, + &[DOUBLE], + ); // #1645: ReadableStream.from(iterable) — builds a pre-loaded stream. module.declare_function("js_readable_stream_from_iterable", DOUBLE, &[DOUBLE]); module.declare_function("js_readable_stream_locked", DOUBLE, &[DOUBLE]); @@ -1027,6 +1034,8 @@ pub(crate) fn declare_phase_b_strings_part2(module: &mut LlModule) { ); // ReadableStreamDefaultReader. module.declare_function("js_reader_read", I64, &[DOUBLE]); + // #4915: BYOB `read(view)` — fills the caller-supplied view. + module.declare_function("js_reader_read_with_view", I64, &[DOUBLE, DOUBLE]); module.declare_function("js_reader_release_lock", DOUBLE, &[DOUBLE]); module.declare_function("js_reader_closed", I64, &[DOUBLE]); module.declare_function("js_reader_cancel", I64, &[DOUBLE, DOUBLE]); @@ -1059,16 +1068,17 @@ pub(crate) fn declare_phase_b_strings_part2(module: &mut LlModule) { module.declare_function("js_writer_ready", I64, &[DOUBLE]); module.declare_function("js_writer_desired_size", DOUBLE, &[DOUBLE]); // TransformStream. - // #1644: leading arg is the `start` hook. + // #1644: leading arg is the `start` hook. #4915: the two trailing args + // are the writable / readable strategies (number or strategy object). module.declare_function( "js_transform_stream_new", DOUBLE, - &[DOUBLE, DOUBLE, DOUBLE, DOUBLE], + &[DOUBLE, DOUBLE, DOUBLE, DOUBLE, DOUBLE], ); module.declare_function( "js_transform_stream_new_from_transformer_object", DOUBLE, - &[DOUBLE, DOUBLE], + &[DOUBLE, DOUBLE, DOUBLE], ); module.declare_function("js_transform_stream_readable", DOUBLE, &[DOUBLE]); module.declare_function("js_transform_stream_writable", DOUBLE, &[DOUBLE]); diff --git a/crates/perry-ext-streams/src/lib.rs b/crates/perry-ext-streams/src/lib.rs index 5f52b067e91..e9864b5dcc1 100644 --- a/crates/perry-ext-streams/src/lib.rs +++ b/crates/perry-ext-streams/src/lib.rs @@ -27,10 +27,9 @@ //! returns the blob handle as-is. A v0.6.0 followup needs a //! cross-wrapper handle-bytes exchange (e.g. perry-ffi //! `wrapper_get_bytes(symbol, handle)`) to wire them up. -//! - **BYOB readers** — `getReader({ mode: "byob" })`. Throws via -//! `js_streams_throw_byob_not_implemented`. -//! - **`ByteLengthQueuingStrategy`** — custom `size()` callbacks per -//! chunk. Throws via `js_streams_throw_byte_length_not_implemented`. +//! - **BYOB readers / `ByteLengthQueuingStrategy`** — implemented in the +//! node:stream/web path (`perry-stdlib/src/streams/byob.rs`, #4915); +//! this perry-ffi port doesn't expose them yet. //! - **`ReadableStream.from(asyncIterable)`** — needs cooperative //! async iteration which the spawn-blocking-only perry-ffi v0.5.x //! surface can't express; throws. @@ -1388,19 +1387,10 @@ unsafe fn transform_close(writable_id: usize) -> *mut Promise { promise } -// ───────────────────────────────────────────────────────────────────── -// Stubs for deferred surface (issue #237 followups) -// ───────────────────────────────────────────────────────────────────── - -#[no_mangle] -pub unsafe extern "C" fn js_streams_throw_byob_not_implemented() -> f64 { - throw_with_message("BYOB readers are not yet implemented (issue #237 followup)"); -} - -#[no_mangle] -pub unsafe extern "C" fn js_streams_throw_byte_length_not_implemented() -> f64 { - throw_with_message("ByteLengthQueuingStrategy is not yet implemented (issue #237 followup)"); -} +// BYOB readers and ByteLengthQueuingStrategy are implemented in the +// node:stream/web path (perry-stdlib/src/streams/byob.rs, #4915); the old +// `js_streams_throw_*_not_implemented` stubs here were never reachable from +// codegen and are gone. // ───────────────────────────────────────────────────────────────────── // Public helpers diff --git a/crates/perry-hir/src/destructuring/var_decl.rs b/crates/perry-hir/src/destructuring/var_decl.rs index 09863027a32..7782449e457 100644 --- a/crates/perry-hir/src/destructuring/var_decl.rs +++ b/crates/perry-hir/src/destructuring/var_decl.rs @@ -999,6 +999,17 @@ pub(crate) fn lower_var_decl_with_destructuring( ); ctx.uses_fetch = true; } + // #4915: `new ReadableStreamBYOBReader(stream)` — + // the handle is a reader, same module tag as + // `stream.getReader({ mode: "byob" })`. + "ReadableStreamBYOBReader" => { + ctx.register_native_instance( + name.clone(), + "readable_stream_reader".to_string(), + "ReadableStreamBYOBReader".to_string(), + ); + ctx.uses_fetch = true; + } "WritableStream" => { ctx.register_native_instance( name.clone(), diff --git a/crates/perry-hir/src/lower/expr_call/stream.rs b/crates/perry-hir/src/lower/expr_call/stream.rs index d3531153cd5..d7038ccf558 100644 --- a/crates/perry-hir/src/lower/expr_call/stream.rs +++ b/crates/perry-hir/src/lower/expr_call/stream.rs @@ -28,6 +28,7 @@ pub(super) fn is_stream_api_method(module: &str, method: &str) -> bool { | "close" | "error" | "desiredSize" + | "byobRequest" ), "readable_stream_reader" => { matches!(method, "read" | "releaseLock" | "cancel" | "closed") diff --git a/crates/perry-hir/src/lower/expr_member.rs b/crates/perry-hir/src/lower/expr_member.rs index 32233d7f4fc..1009a6c369b 100644 --- a/crates/perry-hir/src/lower/expr_member.rs +++ b/crates/perry-hir/src/lower/expr_member.rs @@ -1464,7 +1464,13 @@ fn lower_member_inner(ctx: &mut LoweringContext, member: &ast::MemberExpr) -> Re // Getter properties keep the 0-arg NativeMethodCall below // (they really are getters); everything else here is a // callable method. - "locked" | "desiredSize" | "closed" | "ready" | "readable" | "writable" + "locked" + | "desiredSize" + | "closed" + | "ready" + | "readable" + | "writable" + | "byobRequest" ) { // #1642: a value-read of a Web Streams *method* (not a // getter) must yield a callable bound-method reference, not @@ -2662,6 +2668,7 @@ fn is_stream_api_member(module: &str, prop: &str) -> bool { | "close" | "error" | "desiredSize" + | "byobRequest" ), "readable_stream_reader" => { matches!(prop, "read" | "releaseLock" | "cancel" | "closed") diff --git a/crates/perry-hir/src/lower/expr_new.rs b/crates/perry-hir/src/lower/expr_new.rs index ca9f637ec51..8de57b1e87a 100644 --- a/crates/perry-hir/src/lower/expr_new.rs +++ b/crates/perry-hir/src/lower/expr_new.rs @@ -1170,6 +1170,7 @@ pub(super) fn lower_new(ctx: &mut LoweringContext, new_expr: &ast::NewExpr) -> R | "Blob" | "File" | "ReadableStream" + | "ReadableStreamBYOBReader" | "WritableStream" | "TransformStream" ) { diff --git a/crates/perry-stdlib/src/streams.rs b/crates/perry-stdlib/src/streams.rs index d02bee1a93a..02d6555265f 100644 --- a/crates/perry-stdlib/src/streams.rs +++ b/crates/perry-stdlib/src/streams.rs @@ -19,8 +19,11 @@ //! fetch path eagerly buffers the whole response anyway, so the user- //! visible contract is identical for the consumers we expose here. //! -//! Stubs: BYOB readers and full custom `QueuingStrategy` size accounting — -//! see the inline comment on each site. +//! BYOB readers (`getReader({ mode: "byob" })`, `read(view)`, +//! `controller.byobRequest.respond/respondWithNewView`) and real +//! `QueuingStrategy` size accounting (per-chunk `size()` results summed +//! into `desiredSize`) live in `streams/byob.rs` and the queue helpers on +//! `ReadableStreamData` (#4915). use perry_runtime::{ js_array_alloc, js_array_push, js_closure_call0, js_closure_call1, js_closure_call2, @@ -32,13 +35,26 @@ use std::collections::{HashMap, VecDeque}; use std::os::raw::c_int; use std::sync::Mutex; +mod byob; mod pipe; +mod strategy; mod subclass; #[cfg(test)] mod tests; mod transform; mod writable; +pub use self::byob::{ + js_readable_stream_controller_byob_request, js_readable_stream_get_byob_reader, + js_reader_read_with_view, +}; +pub(crate) use self::strategy::parse_strategy_value; +pub use self::strategy::{ + js_byte_length_queuing_strategy_new, js_count_queuing_strategy_new, + js_streams_strategy_high_water_mark, +}; +use self::strategy::{read_high_water_mark, read_queuing_strategy_size}; + use self::pipe::js_readable_stream_pipe_to; use self::subclass::{box_promise, js_stream_unwrap_handle}; pub(crate) use self::subclass::{dispatch_stream_method, dispatch_stream_property}; @@ -49,7 +65,6 @@ pub use self::subclass::{ pub use self::transform::{ js_stream_web_compression_stream_new, js_stream_web_decompression_stream_new, js_stream_web_text_decoder_stream_new, js_stream_web_text_encoder_stream_new, - js_streams_throw_byob_not_implemented, js_streams_throw_byte_length_not_implemented, js_transform_stream_new, js_transform_stream_new_from_transformer_object, js_transform_stream_readable, js_transform_stream_writable, }; @@ -91,6 +106,16 @@ struct ReadableStreamData { state: ReadableState, /// Queued chunks as NaN-boxed pointers (typically Uint8Array via POINTER_TAG). chunks: VecDeque, + /// Per-chunk strategy sizes, parallel to `chunks` (#4915). The size of a + /// chunk is `strategy.size(chunk)` when a custom strategy is installed, + /// `chunk.byteLength` on byte streams, else 1 — computed once at enqueue + /// time per spec. Mutate the queue only through `push_chunk` / + /// `pop_chunk` / `clear_chunks` / `drain_chunks` so the running + /// `queue_total_size` stays in sync. + chunk_sizes: VecDeque, + /// Running sum of `chunk_sizes` — what `desiredSize` subtracts from the + /// highWaterMark (real ByteLengthQueuingStrategy accounting, #4915). + queue_total_size: f64, /// FIFO of read() promises waiting for a chunk. pending_reads: VecDeque<*mut Promise>, start_cb: i64, @@ -111,14 +136,47 @@ struct ReadableStreamData { canceled: bool, } +impl ReadableStreamData { + fn push_chunk(&mut self, bits: u64, size: f64) { + self.chunks.push_back(bits); + self.chunk_sizes.push_back(size); + self.queue_total_size += size; + } + + fn pop_chunk(&mut self) -> Option { + let bits = self.chunks.pop_front()?; + let size = self.chunk_sizes.pop_front().unwrap_or(1.0); + self.queue_total_size = (self.queue_total_size - size).max(0.0); + Some(bits) + } + + fn clear_chunks(&mut self) { + self.chunks.clear(); + self.chunk_sizes.clear(); + self.queue_total_size = 0.0; + } + + fn drain_chunks(&mut self) -> Vec { + self.chunk_sizes.clear(); + self.queue_total_size = 0.0; + self.chunks.drain(..).collect() + } +} + #[allow(dead_code)] struct WritableStreamData { state: WritableState, write_cb: i64, close_cb: i64, abort_cb: i64, - /// Backlog of writes while the sink's previous `write()` Promise is pending. - write_queue: VecDeque<(u64, *mut Promise)>, + /// Custom `strategy.size(chunk)` callback (#4915). 0 when absent; each + /// chunk then counts as 1 toward `desiredSize`. + strategy_size_cb: i64, + /// Backlog of writes while the sink's previous `write()` Promise is + /// pending: `(chunk_bits, write_promise, strategy_size)`. + write_queue: VecDeque<(u64, *mut Promise, f64)>, + /// Strategy size of the chunk currently in flight (0.0 when idle). + in_flight_size: f64, in_flight: bool, high_water_mark: f64, writer_handle: Option, @@ -165,6 +223,10 @@ struct ReaderData { stream_handle: usize, locked: bool, closed_promise: *mut Promise, + /// True for readers minted by `getReader({ mode: "byob" })` / + /// `new ReadableStreamBYOBReader(stream)` (#4915). BYOB readers route + /// `read(view)` through `byob::js_reader_read_with_view`. + is_byob: bool, } struct WriterData { @@ -258,12 +320,14 @@ fn scan_stream_roots_mut(visitor: &mut perry_runtime::gc::RuntimeRootVisitor<'_> } } } + byob::scan_byob_roots(visitor); if let Ok(mut map) = WRITABLE_STREAMS.lock() { for s in map.values_mut() { visitor.visit_i64_slot(&mut s.write_cb); visitor.visit_i64_slot(&mut s.close_cb); visitor.visit_i64_slot(&mut s.abort_cb); - for (chunk, p) in s.write_queue.iter_mut() { + visitor.visit_i64_slot(&mut s.strategy_size_cb); + for (chunk, p, _size) in s.write_queue.iter_mut() { visit_stream_value_slot(visitor, chunk); visitor.visit_raw_mut_ptr_slot(p); } @@ -493,12 +557,28 @@ fn alloc_readable_with_strategy( ReadableStreamData { state: ReadableState::Readable, chunks: VecDeque::new(), + chunk_sizes: VecDeque::new(), + queue_total_size: 0.0, pending_reads: VecDeque::new(), start_cb, pull_cb, cancel_cb, strategy_size_cb, - high_water_mark: if hwm.is_nan() || hwm <= 0.0 { 1.0 } else { hwm }, + // Byte streams default to highWaterMark 0 (per spec — no eager + // pull at construction; the first pull fires when a read + // request arrives, #4915). Default streams keep the legacy + // clamp-to-1 behavior. + high_water_mark: if hwm.is_nan() { + if is_byte_stream { + 0.0 + } else { + 1.0 + } + } else if !is_byte_stream && hwm <= 0.0 { + 1.0 + } else { + hwm.max(0.0) + }, is_byte_stream, pull_returns_byte_chunk: false, pulling: false, @@ -513,6 +593,16 @@ fn alloc_readable_with_strategy( } fn alloc_writable(write_cb: i64, close_cb: i64, abort_cb: i64, hwm: f64) -> usize { + alloc_writable_with_strategy(write_cb, close_cb, abort_cb, hwm, 0) +} + +fn alloc_writable_with_strategy( + write_cb: i64, + close_cb: i64, + abort_cb: i64, + hwm: f64, + strategy_size_cb: i64, +) -> usize { let id = next_id(&NEXT_STREAM_ID); let ready = js_promise_new(); let closed = js_promise_new(); @@ -524,7 +614,9 @@ fn alloc_writable(write_cb: i64, close_cb: i64, abort_cb: i64, hwm: f64) -> usiz write_cb, close_cb, abort_cb, + strategy_size_cb, write_queue: VecDeque::new(), + in_flight_size: 0.0, in_flight: false, high_water_mark: if hwm.is_nan() || hwm <= 0.0 { 1.0 } else { hwm }, writer_handle: None, @@ -587,11 +679,19 @@ extern "C" fn readable_pull_microtask(closure: *const ClosureHeader) -> f64 { } pub(super) unsafe fn maybe_pull(stream_id: usize) { + // ShouldCallPull (#4915): a parked read request always justifies a + // pull (this is what drives byte streams with highWaterMark 0 — the + // pull only fires once a `read()` / `read(view)` is waiting); + // otherwise pull while the queue is under the highWaterMark. + let has_byob_pending = byob::has_pending(stream_id); let (cb, controller, should_pull, pull_returns_byte_chunk) = { let mut g = READABLE_STREAMS.lock().unwrap(); match g.get_mut(&stream_id) { Some(s) if s.state == ReadableState::Readable && !s.pulling && s.started => { - let need = s.chunks.is_empty() || (s.chunks.len() as f64) < s.high_water_mark; + let has_read_request = !s.pending_reads.is_empty() || has_byob_pending; + let need = has_read_request + || (s.chunks.is_empty() && s.high_water_mark > 0.0) + || (!s.chunks.is_empty() && s.queue_total_size < s.high_water_mark); if need && s.pull_cb != 0 { s.pulling = true; (s.pull_cb, stream_id as f64, true, s.pull_returns_byte_chunk) @@ -639,6 +739,7 @@ unsafe fn close_pending(stream_id: usize) { let result = build_iter_result(TAG_UNDEFINED, true); js_promise_resolve(p, f64::from_bits(result)); } + byob::close_pending_byob(stream_id); } unsafe fn error_pending(stream_id: usize, reason_bits: u64) { @@ -652,6 +753,7 @@ unsafe fn error_pending(stream_id: usize, reason_bits: u64) { for p in promises { js_promise_reject(p, f64::from_bits(reason_bits)); } + byob::error_pending_byob(stream_id, reason_bits); } // ───────────────────────────────────────────────────────────────────── @@ -779,83 +881,6 @@ pub unsafe extern "C" fn js_readable_stream_new_from_source_object( id as f64 } -// ── #1545: node:stream/web QueuingStrategy classes ────────────────────── -// -// `new CountQueuingStrategy({ highWaterMark })` and -// `new ByteLengthQueuingStrategy({ highWaterMark })` produce plain objects -// with a numeric `highWaterMark` field and a `size` method, matching the -// WHATWG built-ins. CountQueuingStrategy.size always returns 1 (chunks are -// counted one-by-one); ByteLengthQueuingStrategy.size returns -// `chunk.byteLength`. Both are surfaced through codegen's builtin-`new` -// dispatch (lower_call/builtin.rs); the import binding lives in -// node_submodules. - -/// `CountQueuingStrategy.prototype.size` — every chunk counts as 1. -extern "C" fn count_queuing_strategy_size(_c: *const ClosureHeader, _chunk: f64) -> f64 { - 1.0 -} - -/// `ByteLengthQueuingStrategy.prototype.size` — `chunk.byteLength`. -extern "C" fn byte_length_queuing_strategy_size(_c: *const ClosureHeader, chunk: f64) -> f64 { - // Mirror Node's `return chunk.byteLength`: the generic property getter - // resolves `.byteLength` for both registered buffers/typed arrays and - // plain `{ byteLength }` objects. - unsafe { perry_runtime::value::js_get_property(chunk, b"byteLength".as_ptr() as i64, 10) } -} - -/// Build a `{ highWaterMark, size }` object for a queuing strategy. `hwm_bits` -/// is the raw JSValue bits read from the caller's options object. -unsafe fn build_queuing_strategy( - hwm_bits: u64, - size_fn: extern "C" fn(*const ClosureHeader, f64) -> f64, -) -> f64 { - let obj = js_object_alloc(0, 2); - let keys = js_array_alloc(2); - let k_hwm = js_string_from_bytes(b"highWaterMark".as_ptr(), 13); - let k_size = js_string_from_bytes(b"size".as_ptr(), 4); - js_array_push(keys, JSValue::string_ptr(k_hwm)); - js_array_push(keys, JSValue::string_ptr(k_size)); - js_object_set_field(obj, 0, JSValue::from_bits(hwm_bits)); - // `size` is a 1-arg native function value. Register the arity so closure - // dispatch pads/forwards the single `chunk` argument correctly. - let fn_ptr = size_fn as *const u8; - perry_runtime::closure::js_register_closure_arity(fn_ptr, 1); - let closure = perry_runtime::closure::js_closure_alloc(fn_ptr, 0); - js_object_set_field(obj, 1, JSValue::pointer(closure as *const u8)); - js_object_set_keys(obj, keys); - f64::from_bits(JSValue::object_ptr(obj as *mut u8).bits()) -} - -/// Read `opts.highWaterMark` (raw JSValue bits) from a strategy's options -/// object; undefined when absent (matches `new CountQueuingStrategy({})`). -unsafe fn read_high_water_mark(opts: f64) -> u64 { - perry_runtime::value::js_get_property(opts, b"highWaterMark".as_ptr() as i64, 13).to_bits() -} - -unsafe fn read_queuing_strategy_size(strategy: f64) -> i64 { - let size = perry_runtime::value::js_get_property(strategy, b"size".as_ptr() as i64, 4); - closure_from_bits(size.to_bits()) -} - -#[no_mangle] -pub unsafe extern "C" fn js_streams_strategy_high_water_mark(strategy: f64) -> f64 { - f64::from_bits(read_high_water_mark(strategy)) -} - -/// `new CountQueuingStrategy({ highWaterMark })`. -#[no_mangle] -pub unsafe extern "C" fn js_count_queuing_strategy_new(opts: f64) -> f64 { - let hwm = read_high_water_mark(opts); - build_queuing_strategy(hwm, count_queuing_strategy_size) -} - -/// `new ByteLengthQueuingStrategy({ highWaterMark })`. -#[no_mangle] -pub unsafe extern "C" fn js_byte_length_queuing_strategy_new(opts: f64) -> f64 { - let hwm = read_high_water_mark(opts); - build_queuing_strategy(hwm, byte_length_queuing_strategy_size) -} - /// Internal helper: build a single-chunk readable stream from an owned /// byte buffer. Used by `blob.stream()` and `response.body`. pub fn alloc_readable_from_bytes(bytes: Vec) -> usize { @@ -867,7 +892,7 @@ pub fn alloc_readable_from_bytes(bytes: Vec) -> usize { if let Some(s) = g.get_mut(&id) { s.started = true; if !bytes.is_empty() { - s.chunks.push_back(chunk_bits); + s.push_chunk(chunk_bits, 1.0); } s.state = ReadableState::Closed; } @@ -923,6 +948,7 @@ pub unsafe extern "C" fn js_readable_stream_get_reader_with_options( stream_handle: id, locked: true, closed_promise: closed_p, + is_byob: byob_requested, }, ); return reader_id as f64; @@ -1017,7 +1043,7 @@ unsafe fn js_readable_stream_cancel_inner( } else { s.canceled = true; s.state = ReadableState::Closed; - s.chunks.clear(); + s.clear_chunks(); s.cancel_cb } } @@ -1315,7 +1341,9 @@ pub unsafe extern "C" fn js_readable_stream_from_iterable(value: f64) -> f64 { { let mut g = READABLE_STREAMS.lock().unwrap(); if let Some(s) = g.get_mut(&id) { - s.chunks.extend(source.chunks); + for bits in source.chunks { + s.push_chunk(bits, 1.0); + } s.started = true; if let Some(error) = source.error { if s.chunks.is_empty() { @@ -1373,7 +1401,7 @@ unsafe fn error_readable_stream(stream_id: usize, reason_bits: u64) { s.state = ReadableState::Errored; s.error_value = reason_bits; s.pending_error_after_chunks = None; - s.chunks.clear(); + s.clear_chunks(); s.reader_handle } None => return, @@ -1415,6 +1443,11 @@ pub unsafe extern "C" fn js_readable_stream_controller_enqueue( "The \"buffer\" argument must be an instance of Buffer, TypedArray, or DataView", ); } + // A pending BYOB read (byte streams only) takes the chunk before the + // default-read queue: the bytes land directly in the caller's view. + if is_byte_stream && byob::service_pending_with_chunk(id, chunk_bits) { + return f64::from_bits(TAG_UNDEFINED); + } let (popped, strategy_size_cb) = { let mut g = READABLE_STREAMS.lock().unwrap(); match g.get_mut(&id) { @@ -1432,7 +1465,9 @@ pub unsafe extern "C" fn js_readable_stream_controller_enqueue( let result = build_iter_result(chunk_bits, false); js_promise_resolve(p, f64::from_bits(result)); } else { - if strategy_size_cb != 0 { + // Per spec the strategy's size(chunk) runs once at enqueue time; the + // result is the chunk's contribution to desiredSize accounting. + let size = if strategy_size_cb != 0 { let size = readable_strategy_size_to_number(js_closure_call1( strategy_size_cb as *const ClosureHeader, chunk, @@ -1440,11 +1475,16 @@ pub unsafe extern "C" fn js_readable_stream_controller_enqueue( if size.is_nan() || size < 0.0 || size.is_infinite() { throw_invalid_readable_strategy_size(id, size); } - } + size + } else if is_byte_stream { + byob::chunk_byte_length(chunk_bits) + } else { + 1.0 + }; let mut g = READABLE_STREAMS.lock().unwrap(); if let Some(s) = g.get_mut(&id) { if s.state == ReadableState::Readable { - s.chunks.push_back(chunk_bits); + s.push_chunk(chunk_bits, size); } } } @@ -1498,9 +1538,7 @@ pub unsafe extern "C" fn js_readable_stream_controller_desired_size(stream_handl let id = stream_handle as usize; let g = READABLE_STREAMS.lock().unwrap(); match g.get(&id) { - Some(s) if s.state == ReadableState::Readable => { - (s.high_water_mark - s.chunks.len() as f64).max(0.0) - } + Some(s) if s.state == ReadableState::Readable => s.high_water_mark - s.queue_total_size, Some(s) if s.state == ReadableState::Errored => f64::NAN, _ => 0.0, } @@ -1579,13 +1617,11 @@ pub unsafe extern "C" fn js_reader_read(reader_handle: f64) -> *mut Promise { let stream_id = match READERS.lock().unwrap().get(&reader_id) { Some(r) if r.locked => r.stream_handle, Some(_) => { - let err = make_error_with_message("Reader is no longer locked to a stream"); - js_promise_reject(promise, f64::from_bits(err)); + reject_type_error(promise, "Reader is no longer locked to a stream"); return promise; } None => { - let err = make_error_with_message("Invalid reader"); - js_promise_reject(promise, f64::from_bits(err)); + reject_type_error(promise, "Invalid reader"); return promise; } }; @@ -1595,7 +1631,7 @@ pub unsafe extern "C" fn js_reader_read(reader_handle: f64) -> *mut Promise { let mut g = READABLE_STREAMS.lock().unwrap(); match g.get_mut(&stream_id) { Some(s) => { - if let Some(c) = s.chunks.pop_front() { + if let Some(c) = s.pop_chunk() { if s.chunks.is_empty() { if let Some(error) = s.pending_error_after_chunks.take() { s.state = ReadableState::Errored; @@ -1818,7 +1854,7 @@ pub unsafe extern "C" fn js_readable_stream_tee(stream_handle: f64) -> f64 { match g.get_mut(&id) { Some(s) if s.reader_handle.is_none() => { is_byte_stream = s.is_byte_stream; - let drained: Vec = s.chunks.drain(..).collect(); + let drained = s.drain_chunks(); s.state = ReadableState::Closed; drained } @@ -1843,6 +1879,8 @@ pub unsafe extern "C" fn js_readable_stream_tee(stream_handle: f64) -> f64 { ReadableStreamData { state: ReadableState::Closed, chunks: chunks.iter().copied().collect(), + chunk_sizes: chunks.iter().map(|_| 1.0).collect(), + queue_total_size: chunks.len() as f64, pending_reads: VecDeque::new(), start_cb: 0, pull_cb: 0, diff --git a/crates/perry-stdlib/src/streams/byob.rs b/crates/perry-stdlib/src/streams/byob.rs new file mode 100644 index 00000000000..0142667444b --- /dev/null +++ b/crates/perry-stdlib/src/streams/byob.rs @@ -0,0 +1,578 @@ +// ───────────────────────────────────────────────────────────────────── +// ReadableStreamBYOBReader + ReadableByteStreamController.byobRequest +// (#4915 — Web Streams residuals after #237/#1545) +// ───────────────────────────────────────────────────────────────────── +// +// `getReader({ mode: "byob" })` / `new ReadableStreamBYOBReader(stream)` +// mint a reader whose `read(view)` fills the caller-supplied +// TypedArray / DataView / Buffer in place. While a BYOB read is +// outstanding, the byte-stream controller exposes `byobRequest` — +// `{ view, respond(bytesWritten), respondWithNewView(view) }` — so pull +// sources can write straight into the caller's buffer instead of +// allocating an intermediate chunk. +// +// Perry's typed arrays own their storage inline (no detachable +// ArrayBuffer aliasing), so the spec's buffer-transfer dance is +// simplified: the caller's view is mutated in place and the fulfilled +// `value` is a fresh view of the same element kind holding exactly the +// bytes read. The common consumer patterns — +// `const { value } = await reader.read(new Uint8Array(n))` and +// pull-sources driving `byobRequest.respond(n)` — observe Node-shaped +// results either way. + +use super::*; +use std::collections::HashMap; + +/// One outstanding `read(view)` — the read promise plus the caller's view. +struct ByobPending { + promise: *mut Promise, + view_bits: u64, +} + +unsafe impl Send for ByobPending {} + +lazy_static::lazy_static! { + static ref BYOB_PENDING: Mutex>> = + Mutex::new(HashMap::new()); +} + +/// True while at least one `read(view)` is parked on this stream — feeds +/// the ShouldCallPull check in `maybe_pull`. +pub(super) fn has_pending(stream_id: usize) -> bool { + BYOB_PENDING + .lock() + .map(|m| m.get(&stream_id).map(|q| !q.is_empty()).unwrap_or(false)) + .unwrap_or(false) +} + +pub(super) fn scan_byob_roots(visitor: &mut perry_runtime::gc::RuntimeRootVisitor<'_>) { + if let Ok(mut map) = BYOB_PENDING.lock() { + for queue in map.values_mut() { + for pending in queue.iter_mut() { + visitor.visit_raw_mut_ptr_slot(&mut pending.promise); + visit_stream_value_slot(visitor, &mut pending.view_bits); + } + } + } +} + +/// A writable window over a caller-supplied view: TypedArray (any kind), +/// DataView, or Node Buffer. `kind` is the typed-array element kind used +/// to mint the fulfilled value (Uint8 for DataView/Buffer receivers). +struct ViewInfo { + data: *mut u8, + byte_len: usize, + kind: u8, + elem_size: usize, +} + +unsafe fn view_info(view_bits: u64) -> Option { + let addr = raw_pointer_addr(view_bits)?; + if addr < 0x1000 { + return None; + } + if let Some(kind) = perry_runtime::typedarray::lookup_typed_array_kind(addr) { + let ta = addr as *mut perry_runtime::typedarray::TypedArrayHeader; + let bytes = perry_runtime::typedarray::typed_array_bytes_mut(ta)?; + return Some(ViewInfo { + data: bytes.as_mut_ptr(), + byte_len: bytes.len(), + kind, + elem_size: perry_runtime::typedarray::elem_size_for_kind(kind).max(1), + }); + } + if perry_runtime::buffer::is_registered_buffer(addr) + && !perry_runtime::buffer::is_any_array_buffer(addr) + { + // DataView and Uint8Array/Buffer registrations both carry their + // byte storage in a BufferHeader. + let buf = addr as *mut perry_runtime::buffer::BufferHeader; + let len = (*buf).length as usize; + return Some(ViewInfo { + data: perry_runtime::buffer::buffer_data_mut(buf), + byte_len: len, + kind: 0, // KIND_U8 — fulfilled values surface as Uint8Array + elem_size: 1, + }); + } + None +} + +/// `chunk.byteLength` for desiredSize accounting on byte streams; 1.0 for +/// values whose byte length can't be derived (matches the count fallback). +pub(super) unsafe fn chunk_byte_length(chunk_bits: u64) -> f64 { + match read_bytes_from_chunk(chunk_bits) { + Some(bytes) => bytes.len() as f64, + None => 1.0, + } +} + +/// Build the fulfilled `value` for a BYOB read: a fresh view of the same +/// element kind holding `bytes` (already copied into the caller's view). +unsafe fn alloc_view_of_kind(kind: u8, elem_size: usize, bytes: &[u8]) -> u64 { + if kind == 0 { + return alloc_uint8array_from_bytes(bytes); + } + let elems = bytes.len() / elem_size.max(1); + let ta = perry_runtime::typedarray::typed_array_alloc(kind, elems as u32); + if ta.is_null() { + return TAG_UNDEFINED; + } + if let Some(dst) = perry_runtime::typedarray::typed_array_bytes_mut(ta) { + let n = dst.len().min(bytes.len()); + dst[..n].copy_from_slice(&bytes[..n]); + } + JSValue::pointer(ta as *const u8).bits() +} + +/// Copy queued chunk bytes into `view`, returning the number of bytes +/// filled. Partially-consumed chunks are put back at the queue head as a +/// fresh Uint8Array remainder. +unsafe fn fill_view_from_queue(stream_id: usize, view: &ViewInfo) -> usize { + let mut filled = 0usize; + // Round capacity down to a whole number of elements. + let cap = (view.byte_len / view.elem_size) * view.elem_size; + while filled < cap { + let chunk = { + let mut g = READABLE_STREAMS.lock().unwrap(); + match g.get_mut(&stream_id) { + Some(s) => match s.pop_chunk() { + Some(c) => c, + None => break, + }, + None => break, + } + }; + let bytes = match read_bytes_from_chunk(chunk) { + Some(b) => b, + None => continue, // non-byte chunk on a byte stream — skip + }; + let take = bytes.len().min(cap - filled); + std::ptr::copy_nonoverlapping(bytes.as_ptr(), view.data.add(filled), take); + filled += take; + if take < bytes.len() { + let remainder = alloc_uint8array_from_bytes(&bytes[take..]); + let mut g = READABLE_STREAMS.lock().unwrap(); + if let Some(s) = g.get_mut(&stream_id) { + let size = (bytes.len() - take) as f64; + s.chunks.push_front(remainder); + s.chunk_sizes.push_front(size); + s.queue_total_size += size; + } + break; + } + } + filled +} + +unsafe fn reject_with_type_error(promise: *mut Promise, message: &str) -> *mut Promise { + reject_type_error(promise, message); + promise +} + +/// `getReader({ mode: "byob" })` routed through a direct +/// `new ReadableStreamBYOBReader(stream)` construction (#4915). +#[no_mangle] +pub unsafe extern "C" fn js_readable_stream_get_byob_reader(stream_handle: f64) -> f64 { + let stream_handle = js_stream_unwrap_handle(stream_handle); + let id = stream_handle as usize; + let is_byte_stream = { + let g = READABLE_STREAMS.lock().unwrap(); + match g.get(&id) { + Some(s) => s.is_byte_stream, + None => { + throw_type_error( + "ReadableStreamBYOBReader constructor requires a ReadableStream argument", + ); + } + } + }; + if !is_byte_stream { + throw_type_error( + "Cannot construct a ReadableStreamBYOBReader for a stream not constructed with a byte source", + ); + } + // Reuse the lock/closed-promise bookkeeping in the shared path; the + // mode check above already proved this is a byte stream. + let reader = get_reader_for_stream(id, true); + reader as f64 +} + +/// Shared reader-minting path used by `js_readable_stream_get_reader_with_options` +/// (via the byob_requested flag) and the BYOB constructor above. +pub(super) unsafe fn get_reader_for_stream(stream_id: usize, is_byob: bool) -> usize { + ensure_gc_registered(); + let mut g = READABLE_STREAMS.lock().unwrap(); + let s = match g.get_mut(&stream_id) { + Some(s) => s, + None => return 0, + }; + if s.reader_handle.is_some() { + drop(g); + throw_type_error("ReadableStream is locked"); + } + let reader_id = next_id(&NEXT_STREAM_ID); + let closed_p = js_promise_new(); + if s.state == ReadableState::Closed { + js_promise_resolve(closed_p, f64::from_bits(TAG_UNDEFINED)); + } else if s.state == ReadableState::Errored { + js_promise_reject(closed_p, f64::from_bits(s.error_value)); + } + s.reader_handle = Some(reader_id); + drop(g); + READERS.lock().unwrap().insert( + reader_id, + ReaderData { + stream_handle: stream_id, + locked: true, + closed_promise: closed_p, + is_byob, + }, + ); + reader_id +} + +/// `reader.read(view)` — BYOB read into a caller-supplied buffer. +/// Default readers ignore the argument and behave like `read()`. +#[no_mangle] +pub unsafe extern "C" fn js_reader_read_with_view(reader_handle: f64, view: f64) -> *mut Promise { + let unwrapped = js_stream_unwrap_handle(reader_handle); + let reader_id = unwrapped as usize; + let (stream_id, is_byob) = match READERS.lock().unwrap().get(&reader_id) { + Some(r) if r.locked => (r.stream_handle, r.is_byob), + Some(_) => { + let promise = js_promise_new(); + return reject_with_type_error(promise, "Reader is no longer locked to a stream"); + } + None => { + let promise = js_promise_new(); + return reject_with_type_error(promise, "Invalid reader"); + } + }; + let view_bits = view.to_bits(); + if !is_byob || view_bits == TAG_UNDEFINED { + // Default reader: `read()` semantics (extra args ignored). + return js_reader_read(reader_handle); + } + let promise = js_promise_new(); + let info = match view_info(view_bits) { + Some(info) => info, + None => { + return reject_with_type_error( + promise, + "ReadableStreamBYOBReader.read(view) requires a TypedArray or DataView", + ); + } + }; + if info.byte_len == 0 { + return reject_with_type_error( + promise, + "ReadableStreamBYOBReader.read(view) view cannot be zero-length", + ); + } + + enum Outcome { + Errored(u64), + Closed, + TryFill, + } + let outcome = { + let g = READABLE_STREAMS.lock().unwrap(); + match g.get(&stream_id) { + Some(s) => match s.state { + ReadableState::Errored => Outcome::Errored(s.error_value), + ReadableState::Closed if s.chunks.is_empty() => Outcome::Closed, + _ => Outcome::TryFill, + }, + None => Outcome::Closed, + } + }; + match outcome { + Outcome::Errored(e) => { + js_promise_reject(promise, f64::from_bits(e)); + return promise; + } + Outcome::Closed => { + let empty = alloc_view_of_kind(info.kind, info.elem_size, &[]); + let result = build_iter_result(empty, true); + js_promise_resolve(promise, f64::from_bits(result)); + return promise; + } + Outcome::TryFill => {} + } + + let filled = fill_view_from_queue(stream_id, &info); + if filled > 0 { + let bytes = std::slice::from_raw_parts(info.data, filled); + let value = alloc_view_of_kind(info.kind, info.elem_size, bytes); + let (closed_now, reader) = { + let g = READABLE_STREAMS.lock().unwrap(); + match g.get(&stream_id) { + Some(s) => ( + s.state == ReadableState::Closed && s.chunks.is_empty(), + s.reader_handle, + ), + None => (false, None), + } + }; + let result = build_iter_result(value, false); + js_promise_resolve(promise, f64::from_bits(result)); + if closed_now { + if let Some(rid) = reader { + let p = READERS.lock().unwrap().get(&rid).map(|r| r.closed_promise); + if let Some(p) = p { + js_promise_resolve(p, f64::from_bits(TAG_UNDEFINED)); + } + } + } + maybe_pull(stream_id); + return promise; + } + + // Nothing buffered: park the read so the controller's byobRequest / + // enqueue can service it, then ask the source to pull. + BYOB_PENDING + .lock() + .unwrap() + .entry(stream_id) + .or_default() + .push_back(ByobPending { promise, view_bits }); + maybe_pull(stream_id); + promise +} + +/// Controller property `byobRequest` — non-null while a BYOB read is +/// outstanding on this byte stream. +#[no_mangle] +pub unsafe extern "C" fn js_readable_stream_controller_byob_request(stream_handle: f64) -> f64 { + let id = js_stream_unwrap_handle(stream_handle) as usize; + let view_bits = { + let g = BYOB_PENDING.lock().unwrap(); + match g.get(&id).and_then(|q| q.front()) { + Some(p) => p.view_bits, + None => return f64::from_bits(TAG_NULL), + } + }; + + let obj = js_object_alloc(0, 3); + let keys = js_array_alloc(3); + let k_view = js_string_from_bytes(b"view".as_ptr(), 4); + let k_respond = js_string_from_bytes(b"respond".as_ptr(), 7); + let k_rwnv = js_string_from_bytes(b"respondWithNewView".as_ptr(), 18); + js_array_push(keys, JSValue::string_ptr(k_view)); + js_array_push(keys, JSValue::string_ptr(k_respond)); + js_array_push(keys, JSValue::string_ptr(k_rwnv)); + js_object_set_field(obj, 0, JSValue::from_bits(view_bits)); + + let respond_fn = byob_request_respond as *const u8; + perry_runtime::closure::js_register_closure_arity(respond_fn, 1); + let respond = perry_runtime::closure::js_closure_alloc(respond_fn, 1); + perry_runtime::closure::js_closure_set_capture_f64(respond, 0, id as f64); + js_object_set_field(obj, 1, JSValue::pointer(respond as *const u8)); + + let rwnv_fn = byob_request_respond_with_new_view as *const u8; + perry_runtime::closure::js_register_closure_arity(rwnv_fn, 1); + let rwnv = perry_runtime::closure::js_closure_alloc(rwnv_fn, 1); + perry_runtime::closure::js_closure_set_capture_f64(rwnv, 0, id as f64); + js_object_set_field(obj, 2, JSValue::pointer(rwnv as *const u8)); + + js_object_set_keys(obj, keys); + f64::from_bits(JSValue::object_ptr(obj as *mut u8).bits()) +} + +extern "C" fn byob_request_respond(closure: *const ClosureHeader, bytes_written: f64) -> f64 { + unsafe { + let stream_id = perry_runtime::closure::js_closure_get_capture_f64(closure, 0) as usize; + let n = JSValue::from_bits(bytes_written.to_bits()).to_number(); + if n.is_nan() || n < 0.0 || n.fract() != 0.0 { + throw_range_error_with_code( + "bytesWritten must be a non-negative integer", + "ERR_OUT_OF_RANGE", + ); + } + let pending = match BYOB_PENDING + .lock() + .unwrap() + .get_mut(&stream_id) + .and_then(|q| q.pop_front()) + { + Some(p) => p, + None => { + throw_type_error("There is no pending BYOB request to respond to"); + } + }; + let info = match view_info(pending.view_bits) { + Some(info) => info, + None => { + reject_type_error(pending.promise, "BYOB request view is no longer valid"); + return f64::from_bits(TAG_UNDEFINED); + } + }; + let n = n as usize; + if n > info.byte_len { + // Put the request back so the source can retry with a valid count. + BYOB_PENDING + .lock() + .unwrap() + .entry(stream_id) + .or_default() + .push_front(pending); + throw_range_error_with_code( + "bytesWritten exceeds the BYOB view's byteLength", + "ERR_OUT_OF_RANGE", + ); + } + let stream_closed = { + let g = READABLE_STREAMS.lock().unwrap(); + g.get(&stream_id) + .map(|s| s.state != ReadableState::Readable) + .unwrap_or(true) + }; + if n == 0 { + if !stream_closed { + BYOB_PENDING + .lock() + .unwrap() + .entry(stream_id) + .or_default() + .push_front(pending); + throw_type_error("bytesWritten must be positive while the stream is readable"); + } + let empty = alloc_view_of_kind(info.kind, info.elem_size, &[]); + let result = build_iter_result(empty, true); + js_promise_resolve(pending.promise, f64::from_bits(result)); + return f64::from_bits(TAG_UNDEFINED); + } + let bytes = std::slice::from_raw_parts(info.data, n); + let value = alloc_view_of_kind(info.kind, info.elem_size, bytes); + let result = build_iter_result(value, false); + js_promise_resolve(pending.promise, f64::from_bits(result)); + } + f64::from_bits(TAG_UNDEFINED) +} + +extern "C" fn byob_request_respond_with_new_view(closure: *const ClosureHeader, view: f64) -> f64 { + unsafe { + let stream_id = perry_runtime::closure::js_closure_get_capture_f64(closure, 0) as usize; + let view_bits = view.to_bits(); + let info = match view_info(view_bits) { + Some(info) => info, + None => { + throw_type_error("respondWithNewView requires a TypedArray or DataView argument"); + } + }; + let pending = match BYOB_PENDING + .lock() + .unwrap() + .get_mut(&stream_id) + .and_then(|q| q.pop_front()) + { + Some(p) => p, + None => { + throw_type_error("There is no pending BYOB request to respond to"); + } + }; + let stream_closed = { + let g = READABLE_STREAMS.lock().unwrap(); + g.get(&stream_id) + .map(|s| s.state != ReadableState::Readable) + .unwrap_or(true) + }; + let done = info.byte_len == 0; + if done && !stream_closed { + BYOB_PENDING + .lock() + .unwrap() + .entry(stream_id) + .or_default() + .push_front(pending); + throw_type_error( + "respondWithNewView requires a non-empty view while the stream is readable", + ); + } + let result = build_iter_result(view_bits, done); + js_promise_resolve(pending.promise, f64::from_bits(result)); + } + f64::from_bits(TAG_UNDEFINED) +} + +/// Controller `enqueue(chunk)` on a byte stream while a BYOB read is +/// parked: copy the chunk straight into the caller's view (overflow goes +/// back on the queue). Returns true when the chunk was consumed. +pub(super) unsafe fn service_pending_with_chunk(stream_id: usize, chunk_bits: u64) -> bool { + let pending = match BYOB_PENDING + .lock() + .unwrap() + .get_mut(&stream_id) + .and_then(|q| q.pop_front()) + { + Some(p) => p, + None => return false, + }; + let info = match view_info(pending.view_bits) { + Some(info) => info, + None => { + reject_type_error(pending.promise, "BYOB request view is no longer valid"); + return true; + } + }; + let bytes = match read_bytes_from_chunk(chunk_bits) { + Some(b) => b, + None => { + // Shouldn't happen (enqueue validated the chunk); requeue the read. + BYOB_PENDING + .lock() + .unwrap() + .entry(stream_id) + .or_default() + .push_front(pending); + return false; + } + }; + let cap = (info.byte_len / info.elem_size) * info.elem_size; + let take = bytes.len().min(cap); + std::ptr::copy_nonoverlapping(bytes.as_ptr(), info.data, take); + if take < bytes.len() { + let remainder = alloc_uint8array_from_bytes(&bytes[take..]); + let mut g = READABLE_STREAMS.lock().unwrap(); + if let Some(s) = g.get_mut(&stream_id) { + if s.state == ReadableState::Readable { + s.push_chunk(remainder, (bytes.len() - take) as f64); + } + } + } + let filled = std::slice::from_raw_parts(info.data, take); + let value = alloc_view_of_kind(info.kind, info.elem_size, filled); + let result = build_iter_result(value, false); + js_promise_resolve(pending.promise, f64::from_bits(result)); + true +} + +/// Stream closed: outstanding BYOB reads settle `{ value: , +/// done: true }`. +pub(super) unsafe fn close_pending_byob(stream_id: usize) { + let drained: Vec = match BYOB_PENDING.lock().unwrap().get_mut(&stream_id) { + Some(q) => q.drain(..).collect(), + None => return, + }; + for pending in drained { + let (kind, elem_size) = match view_info(pending.view_bits) { + Some(info) => (info.kind, info.elem_size), + None => (0, 1), + }; + let empty = alloc_view_of_kind(kind, elem_size, &[]); + let result = build_iter_result(empty, true); + js_promise_resolve(pending.promise, f64::from_bits(result)); + } +} + +/// Stream errored: outstanding BYOB reads reject with the stored reason. +pub(super) unsafe fn error_pending_byob(stream_id: usize, reason_bits: u64) { + let drained: Vec = match BYOB_PENDING.lock().unwrap().get_mut(&stream_id) { + Some(q) => q.drain(..).collect(), + None => return, + }; + for pending in drained { + js_promise_reject(pending.promise, f64::from_bits(reason_bits)); + } +} diff --git a/crates/perry-stdlib/src/streams/pipe.rs b/crates/perry-stdlib/src/streams/pipe.rs index b0c47d262d8..a84043a087b 100644 --- a/crates/perry-stdlib/src/streams/pipe.rs +++ b/crates/perry-stdlib/src/streams/pipe.rs @@ -187,7 +187,7 @@ unsafe fn pipe_next_read(readable_id: usize) -> PipeReadStep { let mut g = READABLE_STREAMS.lock().unwrap(); match g.get_mut(&readable_id) { Some(s) => { - if let Some(c) = s.chunks.pop_front() { + if let Some(c) = s.pop_chunk() { PipeReadStep::Chunk(c) } else if s.state == ReadableState::Closed { PipeReadStep::Done diff --git a/crates/perry-stdlib/src/streams/strategy.rs b/crates/perry-stdlib/src/streams/strategy.rs new file mode 100644 index 00000000000..3ad9b92e21d --- /dev/null +++ b/crates/perry-stdlib/src/streams/strategy.rs @@ -0,0 +1,103 @@ +// QueuingStrategy classes (#1545) and constructor strategy parsing (#4915), +// extracted from streams.rs (2000-line CI cap). + +use super::*; + +// ── #1545: node:stream/web QueuingStrategy classes ────────────────────── +// +// `new CountQueuingStrategy({ highWaterMark })` and +// `new ByteLengthQueuingStrategy({ highWaterMark })` produce plain objects +// with a numeric `highWaterMark` field and a `size` method, matching the +// WHATWG built-ins. CountQueuingStrategy.size always returns 1 (chunks are +// counted one-by-one); ByteLengthQueuingStrategy.size returns +// `chunk.byteLength`. Both are surfaced through codegen's builtin-`new` +// dispatch (lower_call/builtin.rs); the import binding lives in +// node_submodules. + +/// `CountQueuingStrategy.prototype.size` — every chunk counts as 1. +extern "C" fn count_queuing_strategy_size(_c: *const ClosureHeader, _chunk: f64) -> f64 { + 1.0 +} + +/// `ByteLengthQueuingStrategy.prototype.size` — `chunk.byteLength`. +extern "C" fn byte_length_queuing_strategy_size(_c: *const ClosureHeader, chunk: f64) -> f64 { + // Mirror Node's `return chunk.byteLength`: the generic property getter + // resolves `.byteLength` for both registered buffers/typed arrays and + // plain `{ byteLength }` objects. + unsafe { perry_runtime::value::js_get_property(chunk, b"byteLength".as_ptr() as i64, 10) } +} + +/// Build a `{ highWaterMark, size }` object for a queuing strategy. `hwm_bits` +/// is the raw JSValue bits read from the caller's options object. +unsafe fn build_queuing_strategy( + hwm_bits: u64, + size_fn: extern "C" fn(*const ClosureHeader, f64) -> f64, +) -> f64 { + let obj = js_object_alloc(0, 2); + let keys = js_array_alloc(2); + let k_hwm = js_string_from_bytes(b"highWaterMark".as_ptr(), 13); + let k_size = js_string_from_bytes(b"size".as_ptr(), 4); + js_array_push(keys, JSValue::string_ptr(k_hwm)); + js_array_push(keys, JSValue::string_ptr(k_size)); + js_object_set_field(obj, 0, JSValue::from_bits(hwm_bits)); + // `size` is a 1-arg native function value. Register the arity so closure + // dispatch pads/forwards the single `chunk` argument correctly. + let fn_ptr = size_fn as *const u8; + perry_runtime::closure::js_register_closure_arity(fn_ptr, 1); + let closure = perry_runtime::closure::js_closure_alloc(fn_ptr, 0); + js_object_set_field(obj, 1, JSValue::pointer(closure as *const u8)); + js_object_set_keys(obj, keys); + f64::from_bits(JSValue::object_ptr(obj as *mut u8).bits()) +} + +/// Read `opts.highWaterMark` (raw JSValue bits) from a strategy's options +/// object; undefined when absent (matches `new CountQueuingStrategy({})`). +pub(super) unsafe fn read_high_water_mark(opts: f64) -> u64 { + perry_runtime::value::js_get_property(opts, b"highWaterMark".as_ptr() as i64, 13).to_bits() +} + +pub(super) unsafe fn read_queuing_strategy_size(strategy: f64) -> i64 { + let size = perry_runtime::value::js_get_property(strategy, b"size".as_ptr() as i64, 4); + closure_from_bits(size.to_bits()) +} + +/// Interpret a constructor "strategy" argument that may be a plain +/// highWaterMark number (the legacy ABI), a strategy object +/// (`{ highWaterMark, size }` — e.g. a ByteLengthQueuingStrategy), or +/// undefined. Returns `(high_water_mark, size_cb)` with the defaults +/// `(1.0, 0)` (#4915). +pub(crate) unsafe fn parse_strategy_value(value: f64) -> (f64, i64) { + let bits = value.to_bits(); + let top = bits >> 48; + if top >= 0x7FF8 { + if bits == TAG_UNDEFINED || bits == TAG_NULL { + return (1.0, 0); + } + // INT32-tagged plain number (e.g. a literal highWaterMark). + if top == 0x7FFE { + return ((bits as u32 as i32) as f64, 0); + } + let hwm = f64::from_bits(read_high_water_mark(value)); + return (hwm, read_queuing_strategy_size(value)); + } + (value, 0) +} + +#[no_mangle] +pub unsafe extern "C" fn js_streams_strategy_high_water_mark(strategy: f64) -> f64 { + f64::from_bits(read_high_water_mark(strategy)) +} + +/// `new CountQueuingStrategy({ highWaterMark })`. +#[no_mangle] +pub unsafe extern "C" fn js_count_queuing_strategy_new(opts: f64) -> f64 { + let hwm = read_high_water_mark(opts); + build_queuing_strategy(hwm, count_queuing_strategy_size) +} + +/// `new ByteLengthQueuingStrategy({ highWaterMark })`. +#[no_mangle] +pub unsafe extern "C" fn js_byte_length_queuing_strategy_new(opts: f64) -> f64 { + let hwm = read_high_water_mark(opts); + build_queuing_strategy(hwm, byte_length_queuing_strategy_size) +} diff --git a/crates/perry-stdlib/src/streams/subclass.rs b/crates/perry-stdlib/src/streams/subclass.rs index 7d5bfdafbf6..b75bfd3740c 100644 --- a/crates/perry-stdlib/src/streams/subclass.rs +++ b/crates/perry-stdlib/src/streams/subclass.rs @@ -169,6 +169,13 @@ pub(crate) unsafe fn dispatch_stream_method( let is_reader = READERS.lock().unwrap().contains_key(&id); if is_reader { match method { + // BYOB readers fill the caller-supplied view (#4915); default + // readers ignore the argument inside read_with_view. + "read" if !args.is_empty() => { + return Some(box_promise(super::byob::js_reader_read_with_view( + handle, arg0, + ))) + } "read" => return Some(box_promise(js_reader_read(handle))), "releaseLock" => return Some(js_reader_release_lock(handle)), "cancel" => return Some(box_promise(js_reader_cancel(handle, arg0))), @@ -243,6 +250,11 @@ pub(crate) unsafe fn dispatch_stream_property(handle: f64, name: &str) -> f64 { // undefined). `locked` is the one #1670 exercises (`res.body.locked`). match (kind, name) { (1, "locked") => return js_readable_stream_locked(handle), + (1, "desiredSize") => return js_readable_stream_controller_desired_size(handle), + // Non-null only while a BYOB read is parked on this byte stream (#4915). + (1, "byobRequest") => { + return super::byob::js_readable_stream_controller_byob_request(handle) + } (2, "locked") => return js_writable_stream_locked(handle), (3, "closed") => return box_promise(js_reader_closed(handle)), _ => {} @@ -348,6 +360,7 @@ pub unsafe extern "C" fn js_transform_stream_subclass_init( transform_bits, flush_bits, hwm, + hwm, ); attach_handle_to_this(this_bits, handle as usize); f64::from_bits(TAG_UNDEFINED) @@ -364,7 +377,7 @@ pub fn drain_readable_into_bytes(stream_id: usize) -> Vec { let mut g = READABLE_STREAMS.lock().unwrap(); match g.get_mut(&stream_id) { Some(s) => { - let drained: Vec = s.chunks.drain(..).collect(); + let drained = s.drain_chunks(); s.state = ReadableState::Closed; drained } diff --git a/crates/perry-stdlib/src/streams/tests.rs b/crates/perry-stdlib/src/streams/tests.rs index 587943b8058..90fc69757da 100644 --- a/crates/perry-stdlib/src/streams/tests.rs +++ b/crates/perry-stdlib/src/streams/tests.rs @@ -24,6 +24,8 @@ fn root_scanner_emits_callbacks_chunks_and_promises() { ReadableStreamData { state: ReadableState::Errored, chunks: VecDeque::from([0x7FFD_0000_0000_1234]), + chunk_sizes: VecDeque::from([1.0]), + queue_total_size: 1.0, pending_reads: VecDeque::from([0x2345_6780 as *mut Promise]), start_cb: 0x3456_7890, pull_cb: 0, diff --git a/crates/perry-stdlib/src/streams/transform.rs b/crates/perry-stdlib/src/streams/transform.rs index 26b596a5aa7..ee87996232d 100644 --- a/crates/perry-stdlib/src/streams/transform.rs +++ b/crates/perry-stdlib/src/streams/transform.rs @@ -9,30 +9,62 @@ use flate2::read::{ use flate2::Compression; use std::io::Read; +/// `new TransformStream(transformer, writableStrategy, readableStrategy)` +/// (#4915): both strategy params accept a plain highWaterMark number, a +/// strategy object (`{ highWaterMark, size }`, e.g. a +/// ByteLengthQueuingStrategy), or undefined. #[no_mangle] pub unsafe extern "C" fn js_transform_stream_new( start_bits: f64, transform_bits: f64, flush_bits: f64, - hwm: f64, + writable_strategy: f64, + readable_strategy: f64, ) -> f64 { let start_cb = closure_from_bits(start_bits.to_bits()); let transform_cb = closure_from_bits(transform_bits.to_bits()); let flush_cb = closure_from_bits(flush_bits.to_bits()); - alloc_transform_stream(start_cb, transform_cb, flush_cb, None, hwm) + let writable = parse_strategy_value(writable_strategy); + let readable = parse_strategy_value(readable_strategy); + alloc_transform_stream_with_strategies( + start_cb, + transform_cb, + flush_cb, + None, + writable, + readable, + ) } -unsafe fn alloc_transform_stream( +pub(super) unsafe fn alloc_transform_stream( start_cb: i64, transform_cb: i64, flush_cb: i64, native: Option, hwm: f64, +) -> f64 { + alloc_transform_stream_with_strategies( + start_cb, + transform_cb, + flush_cb, + native, + (hwm, 0), + (hwm, 0), + ) +} + +unsafe fn alloc_transform_stream_with_strategies( + start_cb: i64, + transform_cb: i64, + flush_cb: i64, + native: Option, + (w_hwm, w_size_cb): (f64, i64), + (r_hwm, r_size_cb): (f64, i64), ) -> f64 { ensure_gc_registered(); // Allocate the readable side empty (controller is its own handle). - let readable_id = alloc_readable(0, 0, 0, hwm); + let readable_id = alloc_readable_with_strategy(0, 0, 0, r_hwm, false, r_size_cb); { let mut g = READABLE_STREAMS.lock().unwrap(); if let Some(s) = g.get_mut(&readable_id) { @@ -56,9 +88,15 @@ unsafe fn alloc_transform_stream( write_cb: 0, close_cb: 0, abort_cb: 0, + strategy_size_cb: w_size_cb, write_queue: VecDeque::new(), + in_flight_size: 0.0, in_flight: false, - high_water_mark: if hwm.is_nan() || hwm <= 0.0 { 1.0 } else { hwm }, + high_water_mark: if w_hwm.is_nan() || w_hwm <= 0.0 { + 1.0 + } else { + w_hwm + }, writer_handle: None, error_value: 0, ready_promise: ready, @@ -94,14 +132,16 @@ unsafe fn alloc_transform_stream( #[no_mangle] pub unsafe extern "C" fn js_transform_stream_new_from_transformer_object( transformer: f64, - hwm: f64, + writable_strategy: f64, + readable_strategy: f64, ) -> f64 { ensure_gc_registered(); js_transform_stream_new( stream_object_field(transformer, b"start"), stream_object_field(transformer, b"transform"), stream_object_field(transformer, b"flush"), - hwm, + writable_strategy, + readable_strategy, ) } @@ -579,20 +619,5 @@ pub unsafe extern "C" fn js_stream_web_decompression_stream_new(format: f64) -> ) } -// ───────────────────────────────────────────────────────────────────── -// Stubs for deferred surface (issue #237 followups) -// ───────────────────────────────────────────────────────────────────── - -#[no_mangle] -pub unsafe extern "C" fn js_streams_throw_byob_not_implemented() -> f64 { - let err = make_error_with_message("BYOB readers are not yet implemented (issue #237 followup)"); - perry_runtime::exception::js_throw(f64::from_bits(err)); -} - -#[no_mangle] -pub unsafe extern "C" fn js_streams_throw_byte_length_not_implemented() -> f64 { - let err = make_error_with_message( - "ByteLengthQueuingStrategy is not yet implemented (issue #237 followup)", - ); - perry_runtime::exception::js_throw(f64::from_bits(err)); -} +// BYOB readers and ByteLengthQueuingStrategy accounting are implemented +// (#4915) — the old `js_streams_throw_*_not_implemented` stubs are gone. diff --git a/crates/perry-stdlib/src/streams/writable.rs b/crates/perry-stdlib/src/streams/writable.rs index 998fc992be6..0d0228fb39c 100644 --- a/crates/perry-stdlib/src/streams/writable.rs +++ b/crates/perry-stdlib/src/streams/writable.rs @@ -39,11 +39,14 @@ pub unsafe extern "C" fn js_writable_stream_new_with_sink_type( } ensure_gc_registered(); - let id = alloc_writable( + // `hwm` may be a plain number or a whole strategy object (#4915). + let (hwm, size_cb) = parse_strategy_value(hwm); + let id = alloc_writable_with_strategy( closure_from_bits(write_bits.to_bits()), closure_from_bits(close_bits.to_bits()), closure_from_bits(abort_bits.to_bits()), hwm, + size_cb, ); // #1545: WritableStream `start(controller)` fires synchronously at // construction (before any write), matching the WHATWG order @@ -66,11 +69,13 @@ pub unsafe extern "C" fn js_writable_stream_new_from_sink_object(sink: f64, hwm: ); } - let id = alloc_writable( + let (hwm, size_cb) = parse_strategy_value(hwm); + let id = alloc_writable_with_strategy( stream_object_closure(sink, b"write"), stream_object_closure(sink, b"close"), stream_object_closure(sink, b"abort"), hwm, + size_cb, ); let start_cb = stream_object_closure(sink, b"start"); if start_cb != 0 { @@ -221,7 +226,8 @@ pub(super) unsafe fn js_writable_stream_abort_inner( // ───────────────────────────────────────────────────────────────────── fn writable_desired_size(s: &WritableStreamData) -> f64 { - s.high_water_mark - if s.in_flight { 1.0 } else { 0.0 } - s.write_queue.len() as f64 + let queued: f64 = s.write_queue.iter().map(|(_, _, size)| size).sum(); + s.high_water_mark - s.in_flight_size - queued } fn sync_writer_ready_promise(stream_id: usize, writer_id: usize, ready: *mut Promise) { @@ -521,6 +527,26 @@ pub(super) unsafe fn writable_stream_write( return transform_write(stream_id, chunk); } let promise = js_promise_new(); + // The strategy's size(chunk) is user JS — run it before taking the + // registry lock (it may re-enter the streams FFI). + let size_cb = WRITABLE_STREAMS + .lock() + .unwrap() + .get(&stream_id) + .map(|s| s.strategy_size_cb) + .unwrap_or(0); + let chunk_size = if size_cb != 0 { + let size = + JSValue::from_bits(js_closure_call1(size_cb as *const ClosureHeader, chunk).to_bits()) + .to_number(); + if size.is_nan() || size < 0.0 || size.is_infinite() { + let message = invalid_size_message(size); + throw_range_error_with_code(&message, "ERR_INVALID_ARG_VALUE"); + } + size + } else { + 1.0 + }; let mut start_write = None; let needs_pending_ready; { @@ -533,16 +559,17 @@ pub(super) unsafe fn writable_stream_write( return promise; } _ => { - let err = make_error_with_message("Stream is closed or closing"); - js_promise_reject(promise, f64::from_bits(err)); + reject_type_error(promise, "Stream is closed or closing"); return promise; } }; let before = writable_desired_size(s); if s.in_flight { - s.write_queue.push_back((chunk.to_bits(), promise)); + s.write_queue + .push_back((chunk.to_bits(), promise, chunk_size)); } else { s.in_flight = true; + s.in_flight_size = chunk_size; start_write = Some((s.write_cb, chunk, promise)); } let after = writable_desired_size(s); @@ -583,10 +610,12 @@ unsafe fn finish_writable_write_success(stream_id: usize, writer_id: usize, prom match g.get_mut(&stream_id) { Some(s) => { s.in_flight = false; + s.in_flight_size = 0.0; let next = if s.state == WritableState::Writable || s.state == WritableState::Closing { - s.write_queue.pop_front().map(|(chunk, p)| { + s.write_queue.pop_front().map(|(chunk, p, size)| { s.in_flight = true; + s.in_flight_size = size; (s.write_cb, f64::from_bits(chunk), p) }) } else { @@ -624,12 +653,14 @@ unsafe fn finish_writable_write_error(stream_id: usize, promise: *mut Promise, r match g.get_mut(&stream_id) { Some(s) => { s.in_flight = false; + s.in_flight_size = 0.0; s.state = WritableState::Errored; s.error_value = reason.to_bits(); let close_request = s.close_request_promise; s.close_request_promise = std::ptr::null_mut(); s.close_started = false; - let queued: Vec<*mut Promise> = s.write_queue.drain(..).map(|(_, p)| p).collect(); + let queued: Vec<*mut Promise> = + s.write_queue.drain(..).map(|(_, p, _)| p).collect(); (s.ready_promise, s.closed_promise, close_request, queued) } None => ( @@ -667,8 +698,7 @@ pub unsafe extern "C" fn js_writer_write(writer_handle: f64, chunk: f64) -> *mut Some(w) if w.locked => w.stream_handle, _ => { let promise = js_promise_new(); - let err = make_error_with_message("Writer is no longer locked to a stream"); - js_promise_reject(promise, f64::from_bits(err)); + reject_type_error(promise, "Writer is no longer locked to a stream"); return promise; } }; diff --git a/crates/perry/tests/issue_4915_streams_byob.rs b/crates/perry/tests/issue_4915_streams_byob.rs new file mode 100644 index 00000000000..02307467e36 --- /dev/null +++ b/crates/perry/tests/issue_4915_streams_byob.rs @@ -0,0 +1,114 @@ +//! Regression test for #4915: Web Streams residuals — BYOB readers and +//! ByteLengthQueuingStrategy were registered surface that threw +//! "not yet implemented". Now `getReader({ mode: "byob" })` / +//! `new ReadableStreamBYOBReader(stream)` mint readers whose `read(view)` +//! fills the caller-supplied buffer, the byte-stream controller exposes +//! `byobRequest` (`view` / `respond(bytesWritten)` / `respondWithNewView`), +//! and queuing strategies do real per-chunk size accounting in +//! `desiredSize` for ReadableStream / WritableStream / TransformStream. + +use std::path::PathBuf; +use std::process::Command; + +fn perry_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_perry")) +} + +#[test] +fn byob_readers_and_byte_length_strategy() { + let dir = tempfile::tempdir().expect("tempdir"); + let entry = dir.path().join("main.ts"); + let output = dir.path().join("main_bin"); + + std::fs::write( + &entry, + r#" +async function main() { + // BYOB read(view) drains pre-enqueued byte chunks across chunk + // boundaries, then reports done with an empty view. + const rs = new ReadableStream({ + type: "bytes", + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3, 4, 5])); + controller.enqueue(new Uint8Array([6, 7])); + controller.close(); + }, + }); + const reader = rs.getReader({ mode: "byob" }); + const a = await reader.read(new Uint8Array(4)); + console.log("a", a.done, Array.from(a.value).join(",")); + const b = await reader.read(new Uint8Array(4)); + console.log("b", b.done, Array.from(b.value).join(",")); + const c = await reader.read(new Uint8Array(4)); + console.log("c", c.done, c.value.byteLength); + + // BYOB mode on a non-byte stream is a TypeError (spec-shaped, so + // feature detection takes the right path). + const plain = new ReadableStream({ start(ctl) { ctl.close(); } }); + try { + plain.getReader({ mode: "byob" }); + console.log("mode", "no-throw"); + } catch (e: any) { + console.log("mode", e instanceof TypeError); + } + + // Pull sources write through controller.byobRequest. + const pulled = new ReadableStream({ + type: "bytes", + pull(controller) { + const req = controller.byobRequest; + if (req !== null && req !== undefined) { + req.view[0] = 42; + req.respond(1); + } + }, + }); + const r2 = pulled.getReader({ mode: "byob" }); + const d = await r2.read(new Uint8Array(2)); + console.log("d", d.done, Array.from(d.value).join(",")); + + // ByteLengthQueuingStrategy: desiredSize counts bytes, not chunks. + let ctl: any; + new ReadableStream( + { start(c) { ctl = c; } }, + new ByteLengthQueuingStrategy({ highWaterMark: 16 }), + ); + ctl.enqueue(new Uint8Array(6)); + console.log("size", ctl.desiredSize); + + console.log("ok"); +} +main(); +"#, + ) + .expect("write entry"); + + let compile = Command::new(perry_bin()) + .current_dir(dir.path()) + .arg("compile") + .arg(&entry) + .arg("-o") + .arg(&output) + .output() + .expect("run perry compile"); + assert!( + compile.status.success(), + "perry compile failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&compile.stdout), + String::from_utf8_lossy(&compile.stderr) + ); + + let run = Command::new(&output).output().expect("run compiled binary"); + assert!( + run.status.success(), + "compiled binary failed\nstatus: {:?}\nstdout:\n{}\nstderr:\n{}", + run.status, + String::from_utf8_lossy(&run.stdout), + String::from_utf8_lossy(&run.stderr) + ); + let stdout = String::from_utf8_lossy(&run.stdout); + assert_eq!( + stdout, "a false 1,2,3,4\nb false 5,6,7\nc true 0\nmode true\nd false 42\nsize 10\nok\n", + "BYOB read/byobRequest/strategy accounting regressed" + ); +} diff --git a/docs/api/perry.d.ts b/docs/api/perry.d.ts index 35af2e4e2f8..373e8d3f092 100644 --- a/docs/api/perry.d.ts +++ b/docs/api/perry.d.ts @@ -3461,7 +3461,7 @@ declare module "stream/promises" { } declare module "stream/web" { - /** stdlib @perryStub constructor/use throws: not yet implemented (#4915) */ + /** stdlib */ export class ByteLengthQueuingStrategy { [key: string]: any; } /** stdlib */ export class CompressionStream { [key: string]: any; } @@ -3473,9 +3473,9 @@ declare module "stream/web" { export class ReadableByteStreamController { [key: string]: any; } /** stdlib */ export class ReadableStream { [key: string]: any; } - /** stdlib @perryStub constructor/use throws: not yet implemented (#4915) */ + /** stdlib */ export class ReadableStreamBYOBReader { [key: string]: any; } - /** stdlib @perryStub constructor/use throws: not yet implemented (#4915) */ + /** stdlib */ export class ReadableStreamBYOBRequest { [key: string]: any; } /** stdlib */ export class ReadableStreamDefaultController { [key: string]: any; } @@ -3501,7 +3501,7 @@ declare module "stream/web" { } declare module "streams" { - /** stdlib @perryStub constructor/use throws: not yet implemented (#4915) */ + /** stdlib */ export class ByteLengthQueuingStrategy { [key: string]: any; } /** stdlib */ export class CountQueuingStrategy { [key: string]: any; } diff --git a/docs/src/api/reference.md b/docs/src/api/reference.md index a24226512d8..fc200fb3454 100644 --- a/docs/src/api/reference.md +++ b/docs/src/api/reference.md @@ -3105,14 +3105,14 @@ Total: 2781 entries across 114 modules. ### Classes -- `ByteLengthQueuingStrategy` ⚠ **stub** — constructor/use throws: not yet implemented (#4915) +- `ByteLengthQueuingStrategy` - `CompressionStream` - `CountQueuingStrategy` - `DecompressionStream` - `ReadableByteStreamController` - `ReadableStream` -- `ReadableStreamBYOBReader` ⚠ **stub** — constructor/use throws: not yet implemented (#4915) -- `ReadableStreamBYOBRequest` ⚠ **stub** — constructor/use throws: not yet implemented (#4915) +- `ReadableStreamBYOBReader` +- `ReadableStreamBYOBRequest` - `ReadableStreamDefaultController` - `ReadableStreamDefaultReader` - `TextDecoderStream` @@ -3131,7 +3131,7 @@ Total: 2781 entries across 114 modules. ### Classes -- `ByteLengthQueuingStrategy` ⚠ **stub** — constructor/use throws: not yet implemented (#4915) +- `ByteLengthQueuingStrategy` - `CountQueuingStrategy` - `DecompressionStream` - `ReadableStream` diff --git a/test-files/test_parity_streams_byob.ts b/test-files/test_parity_streams_byob.ts new file mode 100644 index 00000000000..180df3200ee --- /dev/null +++ b/test-files/test_parity_streams_byob.ts @@ -0,0 +1,131 @@ +// #4915: ReadableStreamBYOBReader + ByteLengthQueuingStrategy parity. +// Run with: perry vs `node --experimental-strip-types `. + +async function main() { + // ── BYOB read(view) over a byte stream with pre-enqueued chunks ── + { + const rs = new ReadableStream({ + type: "bytes", + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3, 4, 5])); + controller.enqueue(new Uint8Array([6, 7])); + controller.close(); + }, + }); + const reader = rs.getReader({ mode: "byob" }); + const a = await reader.read(new Uint8Array(4)); + console.log("byob-1", a.done, Array.from(a.value)); + const b = await reader.read(new Uint8Array(4)); + console.log("byob-2", b.done, Array.from(b.value)); + const c = await reader.read(new Uint8Array(4)); + console.log("byob-3", c.done, c.value.byteLength); + } + + // ── BYOB reader on a non-byte stream throws TypeError ── + { + const plain = new ReadableStream({ + start(controller) { + controller.enqueue("x"); + controller.close(); + }, + }); + try { + plain.getReader({ mode: "byob" }); + console.log("mode-check", "no-throw"); + } catch (e: any) { + console.log("mode-check", e instanceof TypeError); + } + } + + // ── byobRequest.respond from the pull source ── + { + let pulls = 0; + const rs = new ReadableStream({ + type: "bytes", + pull(controller) { + pulls++; + const req = controller.byobRequest; + if (req !== null && req !== undefined) { + const view = req.view; + view[0] = 42; + view[1] = 43; + req.respond(2); + } else { + controller.enqueue(new Uint8Array([9])); + } + }, + }); + const reader = rs.getReader({ mode: "byob" }); + const r = await reader.read(new Uint8Array(8)); + console.log("respond", r.done, Array.from(r.value)); + } + + // ── ByteLengthQueuingStrategy: real desiredSize accounting ── + { + let ctl: any; + const rs = new ReadableStream( + { + start(controller) { + ctl = controller; + }, + }, + new ByteLengthQueuingStrategy({ highWaterMark: 16 }), + ); + console.log("bls-0", ctl.desiredSize); + ctl.enqueue(new Uint8Array(6)); + console.log("bls-1", ctl.desiredSize); + ctl.enqueue(new Uint8Array(4)); + console.log("bls-2", ctl.desiredSize); + const reader = rs.getReader(); + await reader.read(); + console.log("bls-3", ctl.desiredSize); + } + + // ── ByteLengthQueuingStrategy object shape ── + { + const s = new ByteLengthQueuingStrategy({ highWaterMark: 1024 }); + console.log("strategy", s.highWaterMark, s.size(new Uint8Array(7))); + const c = new CountQueuingStrategy({ highWaterMark: 4 }); + console.log("count", c.highWaterMark, c.size(new Uint8Array(7))); + } + + // ── WritableStream constructed with a ByteLengthQueuingStrategy ── + { + const written: number[] = []; + const ws = new WritableStream( + { + write(chunk: Uint8Array) { + written.push(chunk.byteLength); + }, + }, + new ByteLengthQueuingStrategy({ highWaterMark: 64 }), + ); + const writer = ws.getWriter(); + await writer.write(new Uint8Array(10)); + await writer.write(new Uint8Array(20)); + await writer.close(); + console.log("ws", written.join(","), writer.desiredSize); + } + + // ── TransformStream accepts strategies ── + { + const ts = new TransformStream( + { + transform(chunk: Uint8Array, controller: any) { + controller.enqueue(new Uint8Array(chunk.byteLength * 2)); + }, + }, + new ByteLengthQueuingStrategy({ highWaterMark: 32 }), + new ByteLengthQueuingStrategy({ highWaterMark: 32 }), + ); + const writer = ts.writable.getWriter(); + const reader = ts.readable.getReader(); + await writer.write(new Uint8Array(3)); + const out = await reader.read(); + console.log("ts", out.done, out.value.byteLength); + } + + console.log("done"); +} + +main();