fix(stream): flush piped transforms before end

changelog.d/transform-pipe-flush.md

@@ -0,0 +1,6 @@
+### Fixed
+
+- **Piped `Transform` streams now emit their flush output before `end`.** Pipe
+  completion runs `_flush` or the `flush` option before closing the readable
+  side, including when the flush callback completes asynchronously. Fixes
+  #10450 in #11037.

crates/perry-runtime/src/node_stream.rs

@@ -982,13 +982,9 @@ extern "C" fn pipe_drain_callback(closure: *const ClosureHeader) -> f64 {
     f64::from_bits(TAG_UNDEFINED)
 }
 
-extern "C" fn pipe_finish_destination_callback(closure: *const ClosureHeader) -> f64 {
-    if closure.is_null() {
-        return f64::from_bits(TAG_UNDEFINED);
-    }
-    let dest = js_closure_get_capture_f64(closure, 0);
+fn finish_pipe_destination(dest: f64) {
     if stream_destroyed(dest) || has_truthy_hidden(dest, hidden_finish_emitted_key()) {
-        return f64::from_bits(TAG_UNDEFINED);
+        return;
     }
     if writable_length(dest) > 0.0 {
         set_hidden_value(
@@ -1002,8 +998,17 @@ extern "C" fn pipe_finish_destination_callback(closure: *const ClosureHeader) ->
             hidden_stream_pipe_end_pending_key(),
             f64::from_bits(TAG_FALSE),
         );
-        finish_stream(dest, None);
+        if !finish_transform_stream(dest, None) {
+            finish_stream(dest, None);
+        }
+    }
+}
+
+extern "C" fn pipe_finish_destination_callback(closure: *const ClosureHeader) -> f64 {
+    if closure.is_null() {
+        return f64::from_bits(TAG_UNDEFINED);
     }
+    finish_pipe_destination(js_closure_get_capture_f64(closure, 0));
     f64::from_bits(TAG_UNDEFINED)
 }
 
@@ -1088,7 +1093,7 @@ fn request_pipe_destination_finish(dest: f64) {
         );
         schedule_pipe_destination_finish_check(dest);
     } else {
-        schedule_pipe_destination_finish(dest);
+        finish_pipe_destination(dest);
     }
 }
 

crates/perry-runtime/src/node_stream_readwrite.rs

@@ -920,7 +920,7 @@ pub(super) fn drain_readable_from_events(stream: f64) {
             }
         }
     }
-    if !stream_destroyed(stream) {
+    if !stream_destroyed(stream) && !has_truthy_hidden(stream, hidden_transform_finishing_key()) {
         emit_readable_end_once(stream);
     }
 }

crates/perry-runtime/src/node_stream_tests.rs

@@ -1018,6 +1018,67 @@ fn transform_flush_callback_pushes_tail_before_finish() {
     });
 }
 
+#[test]
+fn piped_transform_flush_callback_pushes_tail_before_finish() {
+    READABLE_DATA_CAPTURED.with(|captured| captured.borrow_mut().clear());
+    TRANSFORM_THIS_HAS_STREAM_STATE.with(|matches| matches.borrow_mut().clear());
+    TRANSFORM_FLUSH_COUNT.with(|count| *count.borrow_mut() = 0);
+
+    let opts = crate::object::js_object_alloc(0, 2);
+    let transform_cb = js_closure_alloc(transform_identity_callback as *const u8, 0);
+    let flush_cb = js_closure_alloc(transform_flush_tail_callback as *const u8, 0);
+    crate::closure::js_register_closure_arity(transform_identity_callback as *const u8, 3);
+    crate::closure::js_register_closure_arity(transform_flush_tail_callback as *const u8, 1);
+    js_object_set_field_by_name(
+        opts,
+        hidden_key(b"transform"),
+        box_pointer(transform_cb as *const u8),
+    );
+    js_object_set_field_by_name(
+        opts,
+        hidden_key(b"flush"),
+        box_pointer(flush_cb as *const u8),
+    );
+
+    let source = js_node_stream_passthrough_new(f64::from_bits(TAG_UNDEFINED));
+    let destination = js_node_stream_transform_new(box_pointer(opts as *const u8));
+    let destination_handle = raw_ptr_from_value(destination) as i64;
+    let data_closure = js_closure_alloc(capture_data_listener as *const u8, 1);
+    crate::closure::js_register_closure_arity(capture_data_listener as *const u8, 1);
+    crate::closure::js_closure_set_capture_f64(data_closure, 0, destination);
+    let _ = js_node_stream_method_on(
+        destination_handle,
+        string_value("data"),
+        box_pointer(data_closure as *const u8),
+    );
+    let _ = js_node_stream_method_pipe(
+        raw_ptr_from_value(source) as i64,
+        destination,
+        f64::from_bits(TAG_UNDEFINED),
+    );
+
+    let source_handle = raw_ptr_from_value(source) as i64;
+    let _ = js_node_stream_method_write(
+        source_handle,
+        string_value("a"),
+        f64::from_bits(TAG_UNDEFINED),
+        f64::from_bits(TAG_UNDEFINED),
+    );
+    let _ = js_node_stream_method_end(source_handle, f64::from_bits(TAG_UNDEFINED));
+    let _ = crate::promise::js_promise_run_microtasks();
+
+    READABLE_DATA_CAPTURED.with(|captured| {
+        assert_eq!(
+            captured.borrow().as_slice(),
+            &[b"a".to_vec(), b"!".to_vec()]
+        );
+    });
+    TRANSFORM_FLUSH_COUNT.with(|count| assert_eq!(*count.borrow(), 1));
+    TRANSFORM_THIS_HAS_STREAM_STATE.with(|matches| {
+        assert_eq!(matches.borrow().as_slice(), &[true]);
+    });
+}
+
 #[test]
 fn transform_callback_can_push_multiple_outputs_per_input() {
     READABLE_DATA_CAPTURED.with(|captured| captured.borrow_mut().clear());

test-files/test_gap_10450_transform_pipe_flush.ts

@@ -0,0 +1,48 @@
+import { PassThrough, Transform } from "stream";
+
+function collect(label: string, stream: Transform): void {
+  let output = "";
+  stream.on("data", (chunk) => {
+    output += chunk;
+  });
+  stream.on("end", () => {
+    console.log(label, JSON.stringify(output));
+  });
+}
+
+const direct = new Transform({
+  transform(chunk, _encoding, callback) {
+    callback(null, String(chunk).toUpperCase());
+  },
+  flush(callback) {
+    callback(null, "|flushed");
+  },
+});
+collect("direct", direct);
+direct.write("ab");
+direct.end("cd");
+
+const source = new PassThrough();
+const piped = source.pipe(new Transform({
+  transform(chunk, _encoding, callback) {
+    callback(null, String(chunk).toUpperCase());
+  },
+  flush(callback) {
+    callback(null, "|flushed");
+  },
+}));
+collect("piped", piped);
+source.write("ab");
+source.end("cd");
+
+const asyncSource = new PassThrough();
+const asyncPiped = asyncSource.pipe(new Transform({
+  transform(chunk, _encoding, callback) {
+    callback(null, String(chunk).toUpperCase());
+  },
+  flush(callback) {
+    queueMicrotask(() => callback(null, "|async-flushed"));
+  },
+}));
+collect("async piped", asyncPiped);
+asyncSource.end("ef");