fix(stream): wait for both finished sides

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`.

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);

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));

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);

crates/perry-runtime/src/node_stream_tests.rs

@@ -28,12 +28,41 @@ thread_local! {
     static TRANSFORM_THIS_HAS_STREAM_STATE: RefCell<Vec<bool>> = const { RefCell::new(Vec::new()) };
     static TRANSFORM_FLUSH_COUNT: RefCell<usize> = const { RefCell::new(0) };
     static UNCAUGHT_STREAM_ERROR_COUNT: RefCell<usize> = const { RefCell::new(0) };
+    static FINISHED_CALLBACK_COUNT: RefCell<usize> = 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)