fix(child_process): finish output readers after EOF and spawn failure

.github/workflows/test.yml

@@ -1452,6 +1452,12 @@ jobs:
         if: steps.scope.outputs.rust_work == 'true'
         run: rm -rf target/perry-auto-* target/debug/libperry_ext_*.a 2>/dev/null || true
 
+      - name: Install the pinned integration-test Node oracle
+        if: steps.scope.outputs.suites != ''
+        uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7
+        with:
+          node-version-file: .node-version
+
       - name: Run scoped integration suites
         if: steps.scope.outputs.suites != ''
         env:
@@ -1475,7 +1481,7 @@ jobs:
           # RFC-2945 abort guards that a JS throw trips — the opposite of the
           # shipped semantics. See the longer note in `cargo-test`.
           if printf '%s\n' "$SUITES" | grep -qE '^(perry|perry-stdlib) '; then
-            if printf '%s\n' "$SUITES" | grep -qE '^perry (bun_text_modules|import_meta_require_value) '; then
+            if printf '%s\n' "$SUITES" | grep -qE '^perry (bun_text_modules|import_meta_require_value|child_output_late_iterator|async_resource_own_bind) '; then
               # Prepare all require providers in one graph, outside the fixture's
               # timeout. A second perry-dev graph inside cargo test exceeded its
               # ten-minute bound on fresh runners (#9989/#9990).

changelog.d/10042-child-output-late-iterator.md

@@ -0,0 +1,15 @@
+- Retain real child stdout/stderr pipe EOF before dispatching end callbacks, so
+  async iterators first pulled or created after EOF finish instead of waiting
+  forever. Update `readable` and `readableEnded` consistently with that state.
+- Finish empty output pipes after a failed spawn as well: no live reactor entry
+  exists to deliver their EOF. This prevents ENOENT/EACCES cleanup from hanging
+  while awaiting stdout/stderr or extra-pipe collectors.
+- Schedule failed-spawn close after its error callback, preventing overdue
+  close timers from reversing error/end ordering during slow startup.
+- Root and reload child-event receivers and arguments across listener callbacks,
+  including forwarding to the shared stream listener registry after EOF.
+- Add runtime regressions for late readers, delayed first pulls, pending empty
+  pulls, and buffered chunks, plus a bounded real-child Node/native parity fixture
+  at O0, Os, and Oz. The regression is independent of any application bundle.
+- Prepare its native providers in the same CI Cargo graph as stdlib, outside
+  per-fixture timeouts, use the pinned Node oracle, and publish compiler errors.

crates/perry-runtime/src/child_process/builder.rs

@@ -138,6 +138,7 @@ pub(crate) fn cp_build_readable() -> f64 {
     let obj = cp_build_object(&methods, CP_READABLE_SHAPE_ID + methods.len() as u32);
     let val = cp_box_ptr(obj as *const u8);
     cp_set_field(val, b"readable", TAG_TRUE_F64);
+    cp_set_field(val, b"readableEnded", TAG_FALSE_F64);
     cp_set_field(val, b"destroyed", TAG_FALSE_F64);
     // A child's `stdout`/`stderr` must be async-iterable, like Node's: both
     // `for await (const chunk of child.stdout)` and the `isAsyncIterable` probe

crates/perry-runtime/src/child_process/emitter.rs

@@ -34,17 +34,19 @@ pub(crate) fn cp_register(target: f64, event: f64, cb: f64) {
 /// any fired. The listener array is re-read each iteration so a moving GC
 /// during a handler call can't strand us on a stale array pointer.
 pub(crate) fn cp_emit(target: f64, event: &str, args: &[f64]) -> bool {
+    let scope = crate::gc::RuntimeHandleScope::new();
+    let target = scope.root_nanbox_f64(target);
+    let args = scope.root_nanbox_f64_slice(args);
     if event == "message"
-        && args
-            .first()
-            .copied()
-            .is_some_and(|msg| crate::cluster::consume_internal_message(target, msg))
+        && args.first().is_some_and(|msg| {
+            crate::cluster::consume_internal_message(target.get_nanbox_f64(), msg.get_nanbox_f64())
+        })
     {
         return true;
     }
 
-    let async_ids =
-        cp_handle_of(target).and_then(|handle| reactor::cp_async_scope_for_target(handle, target));
+    let async_ids = cp_handle_of(target.get_nanbox_f64())
+        .and_then(|handle| reactor::cp_async_scope_for_target(handle, target.get_nanbox_f64()));
     if let Some(ids) = async_ids {
         crate::async_hooks::enter_resource_scope(ids);
     }
@@ -56,17 +58,18 @@ pub(crate) fn cp_emit(target: f64, event: &str, args: &[f64]) -> bool {
     // #9445: the displaced receiver is rooted ONCE here, not once per callback.
     let prev = this_scope.root_nanbox_f64(crate::object::js_implicit_this_get());
     loop {
-        let arr = match cp_array_ptr(cp_get_field(target, &key)) {
+        let arr = match cp_array_ptr(cp_get_field(target.get_nanbox_f64(), &key)) {
             Some(a) => a,
             None => break,
         };
         if i >= crate::array::js_array_length(arr) {
             break;
         }
         let cb = crate::array::js_array_get_f64(arr, i);
-        js_implicit_this_set(target);
+        js_implicit_this_set(target.get_nanbox_f64());
+        let current_args = crate::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&args);
         unsafe {
-            let _ = js_native_call_value(cb, args.as_ptr(), args.len());
+            let _ = js_native_call_value(cb, current_args.as_ptr(), current_args.len());
         }
         js_implicit_this_set(prev.get_nanbox_f64());
         fired = true;
@@ -77,8 +80,15 @@ pub(crate) fn cp_emit(target: f64, event: &str, args: &[f64]) -> bool {
     // and that iterator registers its `data`/`end`/`error` listeners in node:stream's
     // registry rather than the one above. Forward there too, so a `for await` over a
     // child's output sees the chunks the reactor delivers.
-    if !JSValue::from_bits(cp_get_field(target, b"readable").to_bits()).is_undefined() {
-        crate::node_stream::emit_to_stream_listeners(target, event.as_bytes(), args);
+    if !JSValue::from_bits(cp_get_field(target.get_nanbox_f64(), b"readable").to_bits())
+        .is_undefined()
+    {
+        let current_args = crate::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&args);
+        crate::node_stream::emit_to_stream_listeners(
+            target.get_nanbox_f64(),
+            event.as_bytes(),
+            &current_args,
+        );
     }
 
     if let Some(ids) = async_ids {
@@ -87,6 +97,18 @@ pub(crate) fn cp_emit(target: f64, event: &str, args: &[f64]) -> bool {
     fired
 }
 
+#[cfg(test)]
+mod relocation_tests;
+
+/// Deliver real pipe EOF and retain it for readers attaching after the event.
+/// Keep this separate from public `.emit("end")`, which is not a pipe EOF.
+pub(crate) fn cp_readable_end(stream: f64) {
+    let scope = crate::gc::RuntimeHandleScope::new();
+    let stream = scope.root_nanbox_f64(stream);
+    crate::node_stream::async_iterator::mark_foreign_readable_ended(stream.get_nanbox_f64());
+    cp_emit(stream.get_nanbox_f64(), "end", &[]);
+}
+
 // ----- method bodies (each receives the closure; slot 0 = host `this`) -----
 
 pub(crate) extern "C" fn cp_method_on(closure: *const ClosureHeader, event: f64, cb: f64) -> f64 {

crates/perry-runtime/src/child_process/emitter/relocation_tests.rs

@@ -0,0 +1,99 @@
+use super::*;
+
+/// Simulate the runtime-root rewrite performed by a moving collection inside
+/// the first listener, without requiring a native stack map in a Rust test.
+extern "C" fn relocate(closure: *const ClosureHeader, _arg: f64) -> f64 {
+    for pair in 0..2 {
+        let source = js_closure_get_capture_ptr(closure, pair * 2) as *mut u8;
+        let destination = js_closure_get_capture_ptr(closure, pair * 2 + 1) as *mut u8;
+        unsafe {
+            // Both addresses originate from js_object_alloc in this fixture.
+            let header = crate::gc::header_from_trusted_user_ptr(source).cast_mut();
+            crate::gc::set_forwarding_address(header, destination);
+        }
+    }
+    crate::gc::test_rewrite_runtime_handles_for_forwarded_objects();
+    cp_undefined()
+}
+
+extern "C" fn observe(_closure: *const ClosureHeader, arg: f64) -> f64 {
+    let target = crate::object::js_implicit_this_get();
+    cp_set_field(target, b"seen", arg);
+    cp_undefined()
+}
+
+struct RestoreForwarding([(*mut u8, usize); 2]);
+
+impl Drop for RestoreForwarding {
+    fn drop(&mut self) {
+        for (source, first_word) in self.0 {
+            unsafe {
+                // GC_STORE_AUDIT(POINTER_FREE): restore the original object
+                // header word after this synthetic forwarding-only test.
+                source.cast::<usize>().write(first_word);
+                // Restore the same fixture-owned js_object_alloc allocation.
+                let header = crate::gc::header_from_trusted_user_ptr(source).cast_mut();
+                (*header).gc_flags &= !crate::gc::GC_FLAG_FORWARDED;
+            }
+        }
+    }
+}
+
+#[test]
+fn child_dispatch_reloads_receiver_and_arguments_after_listener_relocation() {
+    cp_register_arities();
+    js_register_closure_arity(relocate as *const u8, 1);
+    js_register_closure_arity(observe as *const u8, 1);
+    let scope = crate::gc::RuntimeHandleScope::new();
+    let source = scope.root_nanbox_f64(cp_box_ptr(crate::object::js_object_alloc(0, 0).cast()));
+    let destination =
+        scope.root_nanbox_f64(cp_box_ptr(crate::object::js_object_alloc(0, 0).cast()));
+    let argument = scope.root_nanbox_f64(cp_box_ptr(crate::object::js_object_alloc(0, 0).cast()));
+    let moved_argument =
+        scope.root_nanbox_f64(cp_box_ptr(crate::object::js_object_alloc(0, 0).cast()));
+    let first = scope.root_nanbox_f64(cp_box_ptr(
+        js_closure_alloc(relocate as *const u8, 4).cast(),
+    ));
+    let second =
+        scope.root_nanbox_f64(cp_box_ptr(js_closure_alloc(observe as *const u8, 0).cast()));
+    let event = scope.root_nanbox_f64(cp_box_string("end"));
+    for target in [&source, &destination] {
+        cp_register(
+            target.get_nanbox_f64(),
+            event.get_nanbox_f64(),
+            first.get_nanbox_f64(),
+        );
+        cp_register(
+            target.get_nanbox_f64(),
+            event.get_nanbox_f64(),
+            second.get_nanbox_f64(),
+        );
+    }
+    let sources = [source.get_nanbox_f64(), argument.get_nanbox_f64()]
+        .map(|v| crate::value::js_nanbox_get_pointer(v) as *mut u8);
+    let destinations = [
+        destination.get_nanbox_f64(),
+        moved_argument.get_nanbox_f64(),
+    ]
+    .map(|v| crate::value::js_nanbox_get_pointer(v) as *mut u8);
+    let _restore = RestoreForwarding(sources.map(|p| (p, unsafe { p.cast::<usize>().read() })));
+    let callback =
+        crate::value::js_nanbox_get_pointer(first.get_nanbox_f64()) as *mut ClosureHeader;
+    for pair in 0..2 {
+        js_closure_set_capture_ptr(callback, pair * 2, sources[pair as usize] as i64);
+        js_closure_set_capture_ptr(callback, pair * 2 + 1, destinations[pair as usize] as i64);
+    }
+    assert!(cp_emit(
+        source.get_nanbox_f64(),
+        "end",
+        &[argument.get_nanbox_f64()]
+    ));
+    assert_eq!(
+        source.get_nanbox_f64().to_bits(),
+        destination.get_nanbox_f64().to_bits()
+    );
+    assert_eq!(
+        cp_get_field(destination.get_nanbox_f64(), b"seen").to_bits(),
+        moved_argument.get_nanbox_f64().to_bits()
+    );
+}

crates/perry-runtime/src/child_process/failed_spawn.rs

@@ -0,0 +1,80 @@
+//! A failed spawn has no live reactor entry to deliver output-pipe EOF.
+//! Finish those empty streams after the child error and before child close.
+use super::*;
+
+/// Schedule close only after delivering the error. A close timer armed at spawn
+/// time can already be overdue before the error's setImmediate callback runs.
+/// Keep fork's separate failure contract unchanged by using this for spawn only.
+pub(super) extern "C" fn emit_error_then_close(closure: *const ClosureHeader) -> f64 {
+    let scope = crate::gc::RuntimeHandleScope::new();
+    let cp = scope.root_nanbox_f64(cp_this(closure));
+    reactor::cp_emit_spawn_error(closure);
+    let close = crate::closure::js_closure_alloc(reactor::cp_emit_spawn_close as *const u8, 1);
+    crate::closure::js_closure_set_capture_ptr(close, 0, cp.get_nanbox_f64().to_bits() as i64);
+    crate::timer::js_set_timeout_callback(close as i64, 1.0);
+    cp_undefined()
+}
+
+pub(super) fn finish_outputs(cp: f64) {
+    let scope = crate::gc::RuntimeHandleScope::new();
+    let cp = scope.root_nanbox_f64(cp);
+    let stdio = scope.root_nanbox_f64(cp_get_field(cp.get_nanbox_f64(), b"stdio"));
+    let count = cp_array_ptr(stdio.get_nanbox_f64())
+        .map(|array| crate::array::js_array_length(array))
+        .unwrap_or(0);
+    // fd 0 is writable stdin; every other pipe built by spawn is readable.
+    // Ignored/inherited fds are null and must not receive synthetic events.
+    for fd in 1..count {
+        let Some(array) = cp_array_ptr(stdio.get_nanbox_f64()) else {
+            break;
+        };
+        let stream = crate::array::js_array_get_f64(array, fd);
+        finish_output(stream);
+    }
+}
+
+fn finish_output(stream: f64) {
+    if cp_object_ptr(stream).is_none() {
+        return;
+    }
+    let scope = crate::gc::RuntimeHandleScope::new();
+    let stream = scope.root_nanbox_f64(stream);
+    if cp_get_field(stream.get_nanbox_f64(), b"closed").to_bits() == TAG_TRUE_F64.to_bits() {
+        return;
+    }
+    cp_readable_end(stream.get_nanbox_f64());
+    cp_set_field(stream.get_nanbox_f64(), b"destroyed", TAG_TRUE_F64);
+    cp_set_field(stream.get_nanbox_f64(), b"closed", TAG_TRUE_F64);
+    cp_emit(stream.get_nanbox_f64(), "close", &[]);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn failed_output_retains_end_and_closed_state() {
+        cp_register_arities();
+        let scope = crate::gc::RuntimeHandleScope::new();
+        let stream = scope.root_nanbox_f64(cp_build_readable());
+        for _ in 0..2 {
+            finish_output(stream.get_nanbox_f64());
+            assert_eq!(
+                cp_get_field(stream.get_nanbox_f64(), b"readable").to_bits(),
+                TAG_FALSE_F64.to_bits()
+            );
+            for key in [b"readableEnded".as_slice(), b"destroyed", b"closed"] {
+                assert_eq!(
+                    cp_get_field(stream.get_nanbox_f64(), key).to_bits(),
+                    TAG_TRUE_F64.to_bits()
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn absent_failed_output_is_ignored() {
+        finish_output(TAG_NULL_F64);
+        finish_output(cp_undefined());
+    }
+}

crates/perry-runtime/src/child_process/mod.rs

@@ -55,6 +55,7 @@ use crate::value::JSValue;
 mod builder;
 mod emitter;
 mod exec;
+mod failed_spawn;
 mod options;
 mod output;
 mod registry;
@@ -94,7 +95,7 @@ pub(crate) use emitter::{
     cp_method_kill, cp_method_on, cp_method_pipe, cp_method_read, cp_method_ref,
     cp_method_remove_all_listeners, cp_method_remove_listener, cp_method_send,
     cp_method_set_encoding, cp_method_stdin_end, cp_method_stdin_write, cp_method_this0,
-    cp_method_this1, cp_method_unref, cp_register, cp_send_callback_thunk,
+    cp_method_this1, cp_method_unref, cp_readable_end, cp_register, cp_send_callback_thunk,
     cp_stream_callback_thunk, js_fork_child,
 };
 

crates/perry-runtime/src/child_process/reactor.rs

@@ -1171,13 +1171,12 @@ pub extern "C" fn js_child_process_spawn_streams(
             cp_set_field(cp, b"__cpError", err);
             cp_set_field(cp, b"__cpSpawnErrno", super::cp_errno_number(code));
             cp_set_field(cp, b"exitCode", super::cp_errno_number(code));
-            let emit_closure =
-                crate::closure::js_closure_alloc(cp_emit_spawn_error as *const u8, 1);
+            let emit_closure = crate::closure::js_closure_alloc(
+                super::failed_spawn::emit_error_then_close as *const u8,
+                1,
+            );
             crate::closure::js_closure_set_capture_ptr(emit_closure, 0, cp.to_bits() as i64);
             crate::timer::js_set_immediate_callback(emit_closure as i64);
-            let close = crate::closure::js_closure_alloc(cp_emit_spawn_close as *const u8, 1);
-            crate::closure::js_closure_set_capture_ptr(close, 0, cp.to_bits() as i64);
-            crate::timer::js_set_timeout_callback(close as i64, 1.0);
         }
     }
 
@@ -1187,22 +1186,30 @@ pub extern "C" fn js_child_process_spawn_streams(
 /// Deferred single-`error` emit for the spawn/fork failure path. Slot 0
 /// captures the ChildProcess value.
 pub(super) extern "C" fn cp_emit_spawn_error(closure: *const ClosureHeader) -> f64 {
-    let cp = cp_this(closure);
-    let err = cp_get_field(cp, b"__cpError");
-    if !JSValue::from_bits(err.to_bits()).is_undefined() {
-        cp_emit(cp, "error", &[err]);
-        cp_set_field(cp, b"signalCode", TAG_NULL_F64);
+    let scope = crate::gc::RuntimeHandleScope::new();
+    let cp = scope.root_nanbox_f64(cp_this(closure));
+    let err = scope.root_nanbox_f64(cp_get_field(cp.get_nanbox_f64(), b"__cpError"));
+    if !JSValue::from_bits(err.get_nanbox_f64().to_bits()).is_undefined() {
+        cp_emit(cp.get_nanbox_f64(), "error", &[err.get_nanbox_f64()]);
+        cp_set_field(cp.get_nanbox_f64(), b"signalCode", TAG_NULL_F64);
     }
     cp_undefined()
 }
 
-extern "C" fn cp_emit_spawn_close(closure: *const ClosureHeader) -> f64 {
-    let cp = cp_this(closure);
-    cp_emit(cp, "close", &[cp_get_field(cp, b"exitCode"), TAG_NULL_F64]);
+pub(super) extern "C" fn cp_emit_spawn_close(closure: *const ClosureHeader) -> f64 {
+    let scope = crate::gc::RuntimeHandleScope::new();
+    let cp = scope.root_nanbox_f64(cp_this(closure));
+    super::failed_spawn::finish_outputs(cp.get_nanbox_f64());
+    let code = cp_get_field(cp.get_nanbox_f64(), b"exitCode");
+    cp_emit(cp.get_nanbox_f64(), "close", &[code, TAG_NULL_F64]);
     cp_undefined()
 }
 
 pub(super) fn cp_register_reactor_arities() {
+    crate::closure::js_register_closure_arity(
+        super::failed_spawn::emit_error_then_close as *const u8,
+        0,
+    );
     crate::closure::js_register_closure_arity(cp_emit_spawn_error as *const u8, 0);
     crate::closure::js_register_closure_arity(cp_emit_spawn_close as *const u8, 0);
     crate::closure::js_register_closure_arity(cp_abort_listener as *const u8, 0);
@@ -1631,7 +1638,7 @@ fn cp_reactor_pump_inner() {
                     for fd in end_fds {
                         let stream = cp_stdio_stream(cp, fd);
                         if super::cp_object_ptr(stream).is_some() {
-                            cp_emit(stream, "end", &[]);
+                            super::cp_readable_end(stream);
                         }
                     }
                 }