fix(runtime): root event/listener dispatch copies across moving GC

changelog.d/10606-event-emitter-moving-gc-rooting.md

@@ -0,0 +1,26 @@
+### Fixed
+
+Event/listener dispatch loops in several runtime and stdlib sites cloned
+listener callbacks and call arguments into plain Rust locals, then called
+into user code that could allocate and trigger a moving minor collection,
+then reused those unrooted copies for the next listener. A `class X extends
+EventEmitter` with an allocation-heavy listener could segfault dereferencing
+a later listener's stale closure pointer — reproducibly, on a plain default
+build with no GC environment knobs. `PERRY_GC_DIAG=1
+PERRY_GC_PROTECT_FROMSPACE=1` isolated the fault to a retired-from-space
+dereference of a `GC_TYPE_CLOSURE` object, and the crash disappeared under
+`PERRY_GEN_GC=0` (full mark-sweep, non-moving), confirming the moving
+collector as the cause.
+
+The primary site is `crates/perry-runtime/src/node_stream_event_emitter.rs`
+(`emit_stream_event`/`call_listener_args`) — the path every `class X extends
+EventEmitter` subclass and every Node stream class (`Readable`, `Writable`,
+`Duplex`, `Transform`) actually dispatches through. The same pattern is also
+fixed in `perry-stdlib`'s `events.rs` (`js_event_emitter_emit`/`emit0`,
+`dispatch_error_monitor`, `emit_meta_event`), `domain.rs`
+(`emit_domain_event`), `worker_threads/worker_surface.rs`
+(`stream_emit_event`), and `events/warnings.rs` (`emit_warning`). Every
+listener snapshot and call-argument copy that must stay live across a
+dispatch loop is now rooted through a `RuntimeHandleScope`, re-reading each
+value's current (possibly relocated) address before every call instead of
+trusting the pre-call copy.

crates/perry-runtime/src/node_stream_event_emitter.rs

@@ -746,6 +746,22 @@ pub(super) fn emit_stream_event(stream: f64, event: f64, args: &[f64]) -> f64 {
     if event_identity_bytes(event).is_none() {
         return f64::from_bits(super::TAG_FALSE);
     }
+    // #10600: `listener_snapshot`'s Vec, `stream`, `event` and `args` are all
+    // plain Rust locals, not GC roots. A listener can allocate enough to
+    // trigger a moving minor collection; an unrooted copy then holds a
+    // retired from-space address for the NEXT listener dispatched from this
+    // same loop (reproduced: a `class X extends EventEmitter` whose second
+    // listener allocates heavily segfaults dereferencing the third
+    // listener's stale closure pointer under the default generational GC —
+    // confirmed gone under `PERRY_GEN_GC=0`). Root the whole dispatch
+    // window through one handle scope and re-read every value's current
+    // (possibly relocated) bits before each call, the pattern `events.rs`'s
+    // async branch already uses.
+    let scope = crate::gc::RuntimeHandleScope::new();
+    let stream_h = scope.root_nanbox_f64(stream);
+    let event_h = scope.root_nanbox_f64(event);
+    let arg_handles = scope.root_nanbox_f64_slice(args);
+
     if super::string_value_eq(event, b"error") {
         if let Some(first) = args.first() {
             super::set_hidden_value(stream, super::hidden_error_key(), *first);
@@ -756,14 +772,21 @@ pub(super) fn emit_stream_event(stream: f64, event: f64, args: &[f64]) -> f64 {
         if monitor_snapshot.iter().any(|(_, once)| *once) {
             remove_once_listeners(stream, monitor_event);
         }
-        for (listener, _) in monitor_snapshot {
-            call_listener_args(stream, listener, args);
+        let monitor_listener_values: Vec<f64> = monitor_snapshot.iter().map(|(l, _)| *l).collect();
+        let monitor_handles = scope.root_nanbox_f64_slice(&monitor_listener_values);
+        for handle in &monitor_handles {
+            let live_args = crate::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&arg_handles);
+            call_listener_args(
+                stream_h.get_nanbox_f64(),
+                handle.get_nanbox_f64(),
+                &live_args,
+            );
         }
     }
 
-    let snapshot = listener_snapshot(stream, event);
+    let snapshot = listener_snapshot(stream_h.get_nanbox_f64(), event_h.get_nanbox_f64());
     if snapshot.is_empty() {
-        if super::string_value_eq(event, b"error") {
+        if super::string_value_eq(event_h.get_nanbox_f64(), b"error") {
             let err = args
                 .first()
                 .copied()
@@ -773,17 +796,25 @@ pub(super) fn emit_stream_event(stream: f64, event: f64, args: &[f64]) -> f64 {
         return f64::from_bits(super::TAG_FALSE);
     }
     if snapshot.iter().any(|(_, once)| *once) {
-        remove_once_listeners(stream, event);
+        remove_once_listeners(stream_h.get_nanbox_f64(), event_h.get_nanbox_f64());
     }
     // Node's Readable data delivery path does not route async `data` listener
     // rejections through captureRejections; custom EventEmitter-style events do.
-    let is_data = super::string_value_eq(event, b"data");
-    let capture_rejections =
-        capture_rejections_enabled(stream) && !super::string_value_eq(event, b"error") && !is_data;
-    for (listener, _) in snapshot {
-        let result = call_listener_args(stream, listener, args);
+    let is_data = super::string_value_eq(event_h.get_nanbox_f64(), b"data");
+    let capture_rejections = capture_rejections_enabled(stream_h.get_nanbox_f64())
+        && !super::string_value_eq(event_h.get_nanbox_f64(), b"error")
+        && !is_data;
+    let listener_values: Vec<f64> = snapshot.iter().map(|(l, _)| *l).collect();
+    let listener_handles = scope.root_nanbox_f64_slice(&listener_values);
+    for handle in &listener_handles {
+        let live_args = crate::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&arg_handles);
+        let result = call_listener_args(
+            stream_h.get_nanbox_f64(),
+            handle.get_nanbox_f64(),
+            &live_args,
+        );
         if capture_rejections {
-            capture_listener_rejection(stream, result);
+            capture_listener_rejection(stream_h.get_nanbox_f64(), result);
         } else if is_data {
             // Node's Readable swallows a rejection returned by an async `data`
             // listener — it is neither captured to `error` nor surfaced as an

crates/perry-stdlib/src/domain.rs

@@ -270,9 +270,24 @@ unsafe fn emit_domain_event(handle: Handle, event: &str, args: &[f64]) -> bool {
         return false;
     }
     let receiver = nanbox_handle(handle);
-    for listener in listeners {
+    // #10600: `listeners` and `args` are plain Rust locals cloned out of the
+    // live domain, not GC roots. A listener can allocate enough to trigger a
+    // moving minor collection; an unrooted copy then holds a retired
+    // from-space address for the NEXT listener in this same loop. Root both
+    // through one handle scope and re-read the current bits before every
+    // call.
+    let scope = perry_runtime::gc::RuntimeHandleScope::new();
+    let listener_handles = scope.root_nanbox_f64_slice(&listeners);
+    let arg_handles = scope.root_nanbox_f64_slice(args);
+    for listener_handle in &listener_handles {
+        let live_args =
+            perry_runtime::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&arg_handles);
         let previous_this = perry_runtime::object::js_implicit_this_set(receiver);
-        let _ = perry_runtime::closure::js_native_call_value(listener, args.as_ptr(), args.len());
+        let _ = perry_runtime::closure::js_native_call_value(
+            listener_handle.get_nanbox_f64(),
+            live_args.as_ptr(),
+            live_args.len(),
+        );
         perry_runtime::object::js_implicit_this_set(previous_this);
     }
     true

crates/perry-stdlib/src/events.rs

@@ -261,6 +261,27 @@ struct Listener {
     once: bool,
 }
 
+/// #10600: `Listener.callback` is a raw, untagged heap pointer to a closure.
+/// `snapshot` (cloned out of the live event map before dispatch, so a
+/// once-listener removal mid-dispatch doesn't affect the emit already in
+/// progress) is a plain Rust `Vec`, not a GC root. A listener can allocate
+/// enough to trigger a moving minor collection; an unrooted `snapshot` entry
+/// then holds a retired from-space address for the NEXT listener in the same
+/// dispatch loop. Root every live callback through `scope` up front and read
+/// each one back through its handle — never through `snapshot` itself — at
+/// call time.
+fn root_listener_callbacks<'scope>(
+    scope: &'scope perry_runtime::gc::RuntimeHandleScope,
+    snapshot: &[Listener],
+) -> Vec<perry_runtime::gc::RuntimeHandle<'scope>> {
+    let raw: Vec<u64> = snapshot
+        .iter()
+        .filter(|l| l.callback != 0)
+        .map(|l| l.callback as u64)
+        .collect();
+    scope.root_heap_word_u64_slice(&raw)
+}
+
 #[derive(Copy, Clone)]
 struct PendingOnce {
     promise: *mut Promise,
@@ -406,11 +427,17 @@ impl EventEmitterHandle {
         let str_ptr = js_string_from_bytes(bytes.as_ptr(), bytes.len() as u32);
         let event_arg = js_nanbox_string(str_ptr as i64);
         let listener_arg = js_nanbox_pointer(listener_arg);
-        for l in snapshot {
-            if l.callback != 0 {
-                let closure_ptr = l.callback as *const ClosureHeader;
-                js_closure_call2(closure_ptr, event_arg, listener_arg);
-            }
+        let scope = perry_runtime::gc::RuntimeHandleScope::new();
+        let event_arg_h = scope.root_nanbox_f64(event_arg);
+        let listener_arg_h = scope.root_nanbox_f64(listener_arg);
+        let callback_handles = root_listener_callbacks(&scope, &snapshot);
+        for handle in &callback_handles {
+            let closure_ptr = handle.get_heap_word_u64() as *const ClosureHeader;
+            js_closure_call2(
+                closure_ptr,
+                event_arg_h.get_nanbox_f64(),
+                listener_arg_h.get_nanbox_f64(),
+            );
         }
     }
 
@@ -778,14 +805,15 @@ unsafe fn dispatch_error_monitor(emitter: &mut EventEmitterHandle, arg: Option<f
         emitter.prune_event_if_empty(ERROR_MONITOR_EVENT_NAME);
     }
 
-    for l in snapshot {
-        if l.callback != 0 {
-            let closure_ptr = l.callback as *const ClosureHeader;
-            if let Some(arg) = arg {
-                js_closure_call1(closure_ptr, arg);
-            } else {
-                js_closure_call0(closure_ptr);
-            }
+    let scope = perry_runtime::gc::RuntimeHandleScope::new();
+    let arg_handle = arg.map(|a| scope.root_nanbox_f64(a));
+    let callback_handles = root_listener_callbacks(&scope, &snapshot);
+    for handle in &callback_handles {
+        let closure_ptr = handle.get_heap_word_u64() as *const ClosureHeader;
+        if let Some(arg_handle) = &arg_handle {
+            js_closure_call1(closure_ptr, arg_handle.get_nanbox_f64());
+        } else {
+            js_closure_call0(closure_ptr);
         }
     }
 }
@@ -1070,13 +1098,20 @@ pub unsafe extern "C" fn js_event_emitter_emit(
 
             let capture_rejections = emitter.capture_rejections && event_name != "error";
             let async_handle = emitter.async_resource_handle;
-            for l in snapshot {
-                if l.callback != 0 {
-                    let result =
-                        call_emitter_listener(handle, async_handle, l.callback, &emitted_args);
-                    if capture_rejections {
-                        capture_listener_rejection(handle, result);
-                    }
+            let scope = perry_runtime::gc::RuntimeHandleScope::new();
+            let arg_handles = scope.root_nanbox_f64_slice(&emitted_args);
+            let callback_handles = root_listener_callbacks(&scope, &snapshot);
+            for handle_cb in &callback_handles {
+                let live_args =
+                    perry_runtime::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&arg_handles);
+                let result = call_emitter_listener(
+                    handle,
+                    async_handle,
+                    handle_cb.get_heap_word_u64() as i64,
+                    &live_args,
+                );
+                if capture_rejections {
+                    capture_listener_rejection(handle, result);
                 }
             }
         }
@@ -1143,12 +1178,17 @@ pub unsafe extern "C" fn js_event_emitter_emit0(handle: Handle, event_bits: i64)
 
             let capture_rejections = emitter.capture_rejections && event_name != "error";
             let async_handle = emitter.async_resource_handle;
-            for l in snapshot {
-                if l.callback != 0 {
-                    let result = call_emitter_listener(handle, async_handle, l.callback, &[]);
-                    if capture_rejections {
-                        capture_listener_rejection(handle, result);
-                    }
+            let scope = perry_runtime::gc::RuntimeHandleScope::new();
+            let callback_handles = root_listener_callbacks(&scope, &snapshot);
+            for handle_cb in &callback_handles {
+                let result = call_emitter_listener(
+                    handle,
+                    async_handle,
+                    handle_cb.get_heap_word_u64() as i64,
+                    &[],
+                );
+                if capture_rejections {
+                    capture_listener_rejection(handle, result);
                 }
             }
         }

crates/perry-stdlib/src/events/warnings.rs

@@ -48,9 +48,24 @@ unsafe fn emit_warning(warning: f64) {
         let key_ptr = js_string_from_bytes(key.as_ptr(), key.len() as u32);
         let emit_warning = js_object_get_field_by_name_f64(process_obj, key_ptr);
         if closure_ptr_from_value(emit_warning).is_some() {
-            let args = [warning];
-            let previous_this = perry_runtime::object::js_implicit_this_set(process);
-            perry_runtime::closure::js_native_call_value(emit_warning, args.as_ptr(), args.len());
+            // #10600: `emit_warning`, `process` and `warning` are plain Rust
+            // locals, not GC roots. `js_native_call_value` can itself
+            // allocate in its own dispatch preamble before it reads the
+            // callee/args, which would otherwise leave these pointing at a
+            // retired from-space address by the time they're dereferenced.
+            let scope = perry_runtime::gc::RuntimeHandleScope::new();
+            let callback_h = scope.root_nanbox_f64(emit_warning);
+            let process_h = scope.root_nanbox_f64(process);
+            let arg_handles = scope.root_nanbox_f64_slice(&[warning]);
+            let previous_this =
+                perry_runtime::object::js_implicit_this_set(process_h.get_nanbox_f64());
+            let live_args =
+                perry_runtime::gc::RuntimeHandleScope::refreshed_nanbox_f64_slice(&arg_handles);
+            perry_runtime::closure::js_native_call_value(
+                callback_h.get_nanbox_f64(),
+                live_args.as_ptr(),
+                live_args.len(),
+            );
             perry_runtime::object::js_implicit_this_set(previous_this);
             return;
         }

crates/perry-stdlib/src/worker_threads/worker_surface.rs

@@ -387,12 +387,22 @@ fn stream_emit_event(event: f64, arg: f64) -> f64 {
     let Some(arr) = array_ptr_from_value(get_object_field_from_value(this, &key)) else {
         return js_bool(false);
     };
-    let args = [arg];
-    let len = perry_runtime::array::js_array_length(arr);
+    // #10600: `this`, `arr` and `arg` are plain Rust locals, not GC roots.
+    // `arr` is the listener array's own raw pointer, so a stale copy after a
+    // listener allocates enough to trigger a moving minor collection
+    // corrupts every read for the rest of this loop, not just one listener.
+    // Root all three through one handle scope and re-read the current bits
+    // before every dispatch.
+    let scope = perry_runtime::gc::RuntimeHandleScope::new();
+    let this_h = scope.root_nanbox_f64(this);
+    let arr_h = scope.root_raw_mut_ptr(arr);
+    let arg_h = scope.root_nanbox_f64(arg);
+    let len = perry_runtime::array::js_array_length(arr_h.get_raw_mut_ptr());
     for i in 0..len {
-        let callback = perry_runtime::array::js_array_get_f64(arr, i);
-        let prev_this = perry_runtime::object::js_implicit_this_set(this);
+        let callback = perry_runtime::array::js_array_get_f64(arr_h.get_raw_mut_ptr(), i);
+        let prev_this = perry_runtime::object::js_implicit_this_set(this_h.get_nanbox_f64());
         unsafe {
+            let args = [arg_h.get_nanbox_f64()];
             let _ =
                 perry_runtime::closure::js_native_call_value(callback, args.as_ptr(), args.len());
         }

test-files/test_gap_10600_event_emitter_dispatch_rooting.ts

@@ -0,0 +1,39 @@
+import { EventEmitter } from "node:events";
+
+// #10600: `class X extends EventEmitter` dispatches through
+// perry-runtime's node_stream_event_emitter.rs, whose emit loop cloned
+// listener callbacks (and the call args) into plain Rust locals, then
+// called into user code that can allocate, then reused those unrooted
+// copies for the next listener. A moving minor collection triggered by
+// an allocation-heavy listener left the NEXT listener's snapshot entry
+// pointing at retired from-space memory — no GC env knobs needed, this
+// crashes under the plain default build.
+class Bus extends EventEmitter {}
+
+const bus = new Bus();
+const seen: string[] = [];
+
+bus.on("tick", function (this: any, tag: string) {
+  seen.push("first:" + tag);
+});
+
+bus.on("tick", function (this: any, tag: string) {
+  // Allocate enough that a moving minor collection lands here, while the
+  // THIRD listener's closure is still a live, unrooted pointer captured
+  // by the emit loop's listener snapshot.
+  const churn: any[] = [];
+  for (let i = 0; i < 6000; i++) {
+    churn.push({ k: i, s: "x" + i, o: { i, s2: "y" + i } });
+  }
+  seen.push("second:" + tag + ":" + churn.length);
+});
+
+bus.on("tick", function (this: any, tag: string) {
+  seen.push("third:" + tag);
+});
+
+for (let i = 0; i < 300; i++) {
+  bus.emit("tick", "t" + i);
+}
+
+console.log(seen.length, seen[seen.length - 1]);