From 23f0e503ee99b18a9b7b882f63acd65db0d184d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ralph=20K=C3=BCpper?= Date: Sun, 6 Sep 2026 19:50:52 +0200 Subject: [PATCH] fix(stream): wait for both finished sides --- changelog.d/9906-stream-finished-duplex.md | 5 ++++ crates/perry-runtime/src/node_stream.rs | 25 ++++++++++++++++ .../src/node_stream_constructors/pipeline.rs | 2 +- .../perry-runtime/src/node_stream_dispatch.rs | 1 + crates/perry-runtime/src/node_stream_tests.rs | 29 +++++++++++++++++++ 5 files changed, 61 insertions(+), 1 deletion(-) create mode 100644 changelog.d/9906-stream-finished-duplex.md diff --git a/changelog.d/9906-stream-finished-duplex.md b/changelog.d/9906-stream-finished-duplex.md new file mode 100644 index 0000000000..6931f15687 --- /dev/null +++ b/changelog.d/9906-stream-finished-duplex.md @@ -0,0 +1,5 @@ +### Fixed + +- Callback-form `stream.finished()` now waits for both sides of a duplex stream, + so ending an unread `PassThrough` does not report completion before its + readable side emits `end`. diff --git a/crates/perry-runtime/src/node_stream.rs b/crates/perry-runtime/src/node_stream.rs index 722648d1a6..dc13ddf430 100644 --- a/crates/perry-runtime/src/node_stream.rs +++ b/crates/perry-runtime/src/node_stream.rs @@ -373,6 +373,31 @@ extern "C" fn ns_finished_error_false_close(closure: *const ClosureHeader) -> f6 f64::from_bits(TAG_UNDEFINED) } +extern "C" fn ns_finished_default_completion(closure: *const ClosureHeader) -> f64 { + if closure.is_null() || js_closure_get_capture_f64(closure, 2).to_bits() == TAG_TRUE { + return f64::from_bits(TAG_UNDEFINED); + } + let stream = js_closure_get_capture_f64(closure, 0); + let readable_done = !js_node_stream_has_readable_side(stream) + || has_truthy_hidden(stream, hidden_end_emitted_key()); + let writable_done = !js_node_stream_has_writable_side(stream) + || has_truthy_hidden(stream, hidden_finish_emitted_key()); + let closed = has_truthy_hidden(stream, hidden_key(b"closed")); + let error = readable_hidden_error(stream); + if error.is_none() && !closed && !(readable_done && writable_done) { + return f64::from_bits(TAG_UNDEFINED); + } + + js_closure_set_capture_f64(closure as *mut ClosureHeader, 2, f64::from_bits(TAG_TRUE)); + let callback = js_closure_get_capture_f64(closure, 1); + if let Some(error) = error { + call_listener_args(stream, callback, &[error]); + } else { + call_listener_args(stream, callback, &[]); + } + f64::from_bits(TAG_UNDEFINED) +} + extern "C" fn ns_finished_signal_abort(closure: *const ClosureHeader) -> f64 { if closure.is_null() { return f64::from_bits(TAG_UNDEFINED); diff --git a/crates/perry-runtime/src/node_stream_constructors/pipeline.rs b/crates/perry-runtime/src/node_stream_constructors/pipeline.rs index 27ab7fa395..39c95c9388 100644 --- a/crates/perry-runtime/src/node_stream_constructors/pipeline.rs +++ b/crates/perry-runtime/src/node_stream_constructors/pipeline.rs @@ -182,7 +182,7 @@ pub(super) fn add_finished_signal_abort_listener(stream: f64, signal: f64, callb } pub(super) fn add_finished_cleanup_completion_listener(stream: f64, callback: f64) { - let listener = js_closure_alloc(ns_finished_error_false_close as *const u8, 3); + let listener = js_closure_alloc(ns_finished_default_completion as *const u8, 3); js_closure_set_capture_f64(listener, 0, stream); js_closure_set_capture_f64(listener, 1, callback); js_closure_set_capture_f64(listener, 2, f64::from_bits(TAG_FALSE)); diff --git a/crates/perry-runtime/src/node_stream_dispatch.rs b/crates/perry-runtime/src/node_stream_dispatch.rs index 7fc0cc5ef2..7bdc6e3c8e 100644 --- a/crates/perry-runtime/src/node_stream_dispatch.rs +++ b/crates/perry-runtime/src/node_stream_dispatch.rs @@ -604,6 +604,7 @@ pub(super) fn register_stub_arities() { 1, ); register(ns_finished_error_false_close as *const u8, 0); + register(ns_finished_default_completion as *const u8, 0); register(ns_finished_signal_abort as *const u8, 0); register(ns_iter_to_array as *const u8, 1); register(ns_iter_map as *const u8, 2); diff --git a/crates/perry-runtime/src/node_stream_tests.rs b/crates/perry-runtime/src/node_stream_tests.rs index 19aa3f717a..1ba98c5c90 100644 --- a/crates/perry-runtime/src/node_stream_tests.rs +++ b/crates/perry-runtime/src/node_stream_tests.rs @@ -28,12 +28,41 @@ thread_local! { static TRANSFORM_THIS_HAS_STREAM_STATE: RefCell> = const { RefCell::new(Vec::new()) }; static TRANSFORM_FLUSH_COUNT: RefCell = const { RefCell::new(0) }; static UNCAUGHT_STREAM_ERROR_COUNT: RefCell = const { RefCell::new(0) }; + static FINISHED_CALLBACK_COUNT: RefCell = const { RefCell::new(0) }; } fn catches_runtime_throw(f: impl FnOnce()) -> bool { crate::exception::catch_js_throw(f).is_err() } +extern "C" fn capture_finished_callback(_closure: *const ClosureHeader) -> f64 { + FINISHED_CALLBACK_COUNT.with(|count| *count.borrow_mut() += 1); + f64::from_bits(TAG_UNDEFINED) +} + +#[test] +fn finished_waits_for_both_passthrough_sides() { + FINISHED_CALLBACK_COUNT.with(|count| *count.borrow_mut() = 0); + crate::closure::js_register_closure_arity(capture_finished_callback as *const u8, 0); + + let stream = js_node_stream_passthrough_new(f64::from_bits(TAG_UNDEFINED)); + let callback = + box_pointer(js_closure_alloc(capture_finished_callback as *const u8, 0) as *const u8); + let mut args = crate::array::js_array_alloc(2); + args = crate::array::js_array_push_f64(args, stream); + args = crate::array::js_array_push_f64(args, callback); + js_node_stream_finished(args); + + let handle = raw_ptr_from_value(stream) as i64; + js_node_stream_method_end(handle, string_value("done")); + let _ = crate::promise::js_promise_run_microtasks(); + FINISHED_CALLBACK_COUNT.with(|count| assert_eq!(*count.borrow(), 0)); + + js_node_stream_method_resume(handle); + let _ = crate::promise::js_promise_run_microtasks(); + FINISHED_CALLBACK_COUNT.with(|count| assert_eq!(*count.borrow(), 1)); +} + pub(super) fn string_value(s: &str) -> f64 { let ptr = crate::string::js_string_from_bytes(s.as_ptr(), s.len() as u32); box_string(ptr)