fix(runtime): settle and dispatch cross-agent async work only on the owning agent (#11433)

changelog.d/11445-cross-agent-pump-ownership.md

@@ -0,0 +1,7 @@
+Fixed `test_gap_turnloop_p9_worker_agent_net` failing intermittently (#11433). A `node:worker_threads` Worker doing `fetch` against a server owned by the main thread printed `immediate 000undefined`, threw `Invalid response handle`, or stalled at `WORKER STOPPED AFTER n/2`, because every JS agent runs the same process-wide pumps and three queues had no owner tag, so whichever agent pumped first drained all of them:
+
+- perry-stdlib's promise-resolution queues (`common/async_bridge.rs`): a Worker's fetch promise could be settled on the main thread, and vice versa.
+- The `worker_threads` Worker-to-parent event queue: a Worker could drain its own `postMessage` events, so the parent never received them.
+- perry-ext-http's server pump: a Worker could run the main thread's `http.Server` handler, so the request was never answered.
+
+Each entry now records its `AgentId`. Pumps, keep-alive checks and the async-bridge GC scanner act only on the calling agent's entries. Pool and fallback threads re-assert the submitting agent with `ResolutionOwnerScope`. `perry_runtime::agent::register_retire_hook` purges a retired agent's leftovers, and `perry_ffi::agent_post::current_agent()` exposes the agent id to ext crates. Under CPU load, the test went from 61/100 failures before the fix to 0/100 after.

crates/perry-ext-http/src/server/mod.rs

@@ -277,6 +277,30 @@ mod tests {
         s
     }
 
+    /// #11433 — a server belongs to the agent that created it, and no other
+    /// agent's pump may treat it as its own. Every agent runs this extension's
+    /// pump, so this is what keeps a Worker from dispatching the primary
+    /// agent's requests on the Worker's thread.
+    #[test]
+    fn a_server_belongs_to_the_agent_that_created_it() {
+        let here = HttpServer::with_handler(0);
+        assert!(here.owned_here());
+        let worker_owner = std::thread::spawn(|| {
+            perry_runtime::agent::enter_worker_agent();
+            let s = HttpServer::with_handler(0);
+            assert!(s.owned_here(), "the creating Worker owns its server");
+            s.owner_agent
+        })
+        .join()
+        .unwrap();
+        let mut foreign = HttpServer::with_handler(0);
+        foreign.owner_agent = worker_owner;
+        assert!(
+            !foreign.owned_here(),
+            "the primary agent must not drain a Worker's server"
+        );
+    }
+
     /// Issue #2210 — `HttpServer::with_handler` seeds Node's
     /// documented timeout defaults so a fresh server reads back the
     /// same numbers Node returns when no options are passed.

crates/perry-ext-http/src/server/server.rs

@@ -134,6 +134,21 @@ pub struct HttpServer {
     pub bun_error_handler: i64,
     /// Observable `Server.development` option.
     pub bun_development: bool,
+    /// The JS agent that created this server (#11433). Its handler and
+    /// listeners are closures in that agent's heap, and its connections live on
+    /// that agent's loop, so only that agent's pump may drain it. Every agent's
+    /// event loop runs this extension's pump — a `node:worker_threads` Worker's
+    /// included — and before this tag a Worker's pump would take a request
+    /// addressed to the primary agent's server and run the primary's handler on
+    /// the Worker's thread.
+    pub owner_agent: u64,
+}
+
+impl HttpServer {
+    /// Whether the calling agent owns this server and may run its JS.
+    pub(crate) fn owned_here(&self) -> bool {
+        self.owner_agent == perry_ffi::agent_post::current_agent()
+    }
 }
 
 impl HttpServer {
@@ -170,6 +185,7 @@ impl HttpServer {
             is_bun_server: false,
             bun_error_handler: 0,
             bun_development: false,
+            owner_agent: perry_ffi::agent_post::current_agent(),
         }
     }
 }
@@ -1025,24 +1041,25 @@ pub unsafe extern "C" fn js_node_http_server_remove_listener(
 #[no_mangle]
 pub extern "C" fn js_node_http_server_has_active() -> i32 {
     let mut active = 0i32;
+    // Only this agent's servers keep this agent alive (#11433).
     iter_handles_of::<HttpServer, _>(|s| {
-        if server_is_active(s) {
+        if s.owned_here() && server_is_active(s) {
             active = 1;
         }
     });
     if active != 0 {
         return 1;
     }
     iter_handles_of::<crate::server::https_server::HttpsServer, _>(|s| {
-        if server_is_active(&s.base) {
+        if s.base.owned_here() && server_is_active(&s.base) {
             active = 1;
         }
     });
     if active != 0 {
         return 1;
     }
     iter_handles_of::<crate::server::http2_server::Http2SecureServer, _>(|s| {
-        if server_is_active(&s.base) {
+        if s.base.owned_here() && server_is_active(&s.base) {
             active = 1;
         }
     });
@@ -1057,6 +1074,19 @@ pub extern "C" fn js_node_http_server_has_active() -> i32 {
     active
 }
 
+/// Whether `server_handle` names an HTTP, HTTPS or HTTP/2 server the calling
+/// agent owns. A handle that names none of them reads as not owned.
+pub(crate) fn server_handle_owned_here(server_handle: i64) -> bool {
+    if let Some(s) = get_handle::<HttpServer>(server_handle) {
+        return s.owned_here();
+    }
+    if let Some(s) = get_handle::<crate::server::https_server::HttpsServer>(server_handle) {
+        return s.base.owned_here();
+    }
+    get_handle::<crate::server::http2_server::Http2SecureServer>(server_handle)
+        .is_some_and(|s| s.base.owned_here())
+}
+
 /// Drain pending requests + upgrades from every registered server,
 /// dispatching to the user handler / `'upgrade'` listener on the
 /// main thread. Called each tick by perry-stdlib's pump (gated on
@@ -1099,9 +1129,17 @@ pub extern "C" fn js_node_http_server_process_pending() -> i32 {
     // #4905 — fire `'connection'` listeners for connections accepted since
     // the last tick, before their requests are dispatched (Node fires
     // `'connection'` ahead of `'request'`).
+    // Only the events of servers this agent owns (#11433); another agent's stay
+    // queued for its own pump.
     let connection_events: Vec<(i64, i64)> = PENDING_CONNECTION_EVENTS
         .lock()
-        .map(|mut q| q.drain(..).collect())
+        .map(|mut q| {
+            let (mine, theirs): (Vec<_>, Vec<_>) = std::mem::take(&mut *q)
+                .into_iter()
+                .partition(|(server_handle, _)| server_handle_owned_here(*server_handle));
+            *q = theirs;
+            mine
+        })
         .unwrap_or_default();
     for (server_handle, socket_handle) in connection_events {
         // The handle may back an HttpServer or an HttpsServer (whose
@@ -1140,8 +1178,10 @@ pub extern "C" fn js_node_http_server_process_pending() -> i32 {
     // Snapshot handle ids first so we can mutate handle state
     // (drain channels, free per-request handles) without the
     // DashMap iterator dangling.
+    // #11433: each loop below visits only the servers this agent created.
     let mut http_handles: Vec<i64> = Vec::new();
     perry_ffi::iter_handle_ids_of::<HttpServer, _>(|id| http_handles.push(id));
+    http_handles.retain(|h| get_handle::<HttpServer>(*h).is_some_and(HttpServer::owned_here));
     for h in http_handles {
         // #4903 — fire the deferred `'listening'` emit + listen callbacks
         // before draining requests: the listen callback is usually what
@@ -1161,6 +1201,10 @@ pub extern "C" fn js_node_http_server_process_pending() -> i32 {
     perry_ffi::iter_handle_ids_of::<crate::server::https_server::HttpsServer, _>(|id| {
         https_handles.push(id)
     });
+    https_handles.retain(|h| {
+        get_handle::<crate::server::https_server::HttpsServer>(*h)
+            .is_some_and(|s| s.base.owned_here())
+    });
     for h in https_handles {
         count += drain_deferred_listen_for::<crate::server::https_server::HttpsServer, _>(h, |s| {
             &mut s.base
@@ -1186,6 +1230,10 @@ pub extern "C" fn js_node_http_server_process_pending() -> i32 {
     perry_ffi::iter_handle_ids_of::<crate::server::http2_server::Http2SecureServer, _>(|id| {
         h2_handles.push(id)
     });
+    h2_handles.retain(|h| {
+        get_handle::<crate::server::http2_server::Http2SecureServer>(*h)
+            .is_some_and(|s| s.base.owned_here())
+    });
     for h in h2_handles {
         count += drain_deferred_listen_for::<crate::server::http2_server::Http2SecureServer, _>(
             h,

crates/perry-ext-http/src/server/server/in_flight.rs

@@ -45,6 +45,9 @@ pub(crate) struct InFlightRequest {
     /// Grace deadline. Past this, synthesize the default response (unless
     /// `skip_default_response`) and free the handles regardless.
     pub(crate) deadline: Instant,
+    /// The agent whose pump parked it — the server's owner (#11433). Only that
+    /// agent reaps it, fires its `'drain'` listeners, or is kept alive by it.
+    owner_agent: u64,
 }
 
 pub(crate) static IN_FLIGHT: Mutex<Vec<InFlightRequest>> = Mutex::new(Vec::new());
@@ -90,7 +93,11 @@ pub(crate) fn finalize_request_handles_deferred(
 /// server's handle "active" so the main loop doesn't exit before the
 /// pending response is flushed.
 pub(crate) fn has_in_flight_requests() -> bool {
-    IN_FLIGHT.lock().map(|g| !g.is_empty()).unwrap_or(false)
+    let agent = perry_ffi::agent_post::current_agent();
+    IN_FLIGHT
+        .lock()
+        .map(|g| g.iter().any(|e| e.owner_agent == agent))
+        .unwrap_or(false)
 }
 
 /// Finalize parked requests whose handler has now called `res.end()` (the
@@ -110,7 +117,13 @@ pub(crate) fn reap_in_flight_requests() {
             return;
         }
         let now = Instant::now();
+        let agent = perry_ffi::agent_post::current_agent();
         guard.retain(|e| {
+            if e.owner_agent != agent {
+                // Another agent's request: its handles and listeners live in
+                // that agent's heap (#11433).
+                return true;
+            }
             let ended = response_writable_ended(e.response_handle);
             if !ended {
                 // Streaming backpressure cleared — fire `'drain'` (outside
@@ -193,6 +206,7 @@ pub(crate) fn finalize_or_park_request(pending: &HttpPendingRequest) {
             response_handle: pending.response_handle,
             skip_default_response: pending.skip_default_response,
             deadline,
+            owner_agent: perry_ffi::agent_post::current_agent(),
         });
     } else {
         // Lock poisoned — fall back to the old immediate behavior so we

crates/perry-ext-http/src/server/turnloop_serve/conn.rs

@@ -125,8 +125,12 @@ fn pending() -> &'static Mutex<HashMap<i64, VecDeque<HttpPendingRequest>>> {
 /// `IncomingMessage` handles whose connection died before their response
 /// completed. Node raises `'aborted'` on the request; the sink cannot run JS,
 /// so the pump drains this and fires the listeners on its own tick.
-fn aborted() -> &'static Mutex<Vec<i64>> {
-    static ABORTED: OnceLock<Mutex<Vec<i64>>> = OnceLock::new();
+///
+/// Each entry is tagged with the agent whose loop queued it — the server's
+/// owner, since the completion sink runs there — and only that agent's pump
+/// takes it (#11433).
+fn aborted() -> &'static Mutex<Vec<(u64, i64)>> {
+    static ABORTED: OnceLock<Mutex<Vec<(u64, i64)>>> = OnceLock::new();
     ABORTED.get_or_init(|| Mutex::new(Vec::new()))
 }
 
@@ -142,35 +146,46 @@ pub(crate) fn note_aborted_handle(handle: i64) {
     aborted()
         .lock()
         .unwrap_or_else(|e| e.into_inner())
-        .push(handle);
+        .push((perry_ffi::agent_post::current_agent(), handle));
 }
 
 /// Take the `IncomingMessage` handles whose connection died mid-request.
 pub(crate) fn take_aborted() -> Vec<i64> {
     let mut queue = aborted().lock().unwrap_or_else(|e| e.into_inner());
-    std::mem::take(&mut *queue)
+    take_owned(&mut queue)
+}
+
+/// Remove and return the calling agent's entries, leaving every other agent's
+/// in place for its own pump (#11433).
+fn take_owned(queue: &mut Vec<(u64, i64)>) -> Vec<i64> {
+    let agent = perry_ffi::agent_post::current_agent();
+    let (mine, theirs): (Vec<_>, Vec<_>) = std::mem::take(queue)
+        .into_iter()
+        .partition(|(owner, _)| *owner == agent);
+    *queue = theirs;
+    mine.into_iter().map(|(_, handle)| handle).collect()
 }
 
 /// Connection-socket handles (`alloc_connection_socket`) whose TCP connection
 /// has fully closed. Same pattern as `aborted()`: the completion sink cannot
 /// run JS, so the pump drains this and fires the socket's `'close'`
 /// listeners on its own tick.
-fn closed_sockets() -> &'static Mutex<Vec<i64>> {
-    static CLOSED: OnceLock<Mutex<Vec<i64>>> = OnceLock::new();
+fn closed_sockets() -> &'static Mutex<Vec<(u64, i64)>> {
+    static CLOSED: OnceLock<Mutex<Vec<(u64, i64)>>> = OnceLock::new();
     CLOSED.get_or_init(|| Mutex::new(Vec::new()))
 }
 
 fn note_closed_socket(socket_handle: i64) {
     closed_sockets()
         .lock()
         .unwrap_or_else(|e| e.into_inner())
-        .push(socket_handle);
+        .push((perry_ffi::agent_post::current_agent(), socket_handle));
 }
 
 /// Take the connection-socket handles due a `'close'` emit.
 pub(crate) fn take_closed_sockets() -> Vec<i64> {
     let mut queue = closed_sockets().lock().unwrap_or_else(|e| e.into_inner());
-    std::mem::take(&mut *queue)
+    take_owned(&mut queue)
 }
 
 /// Note that this connection's in-flight request (if any) will never be
@@ -184,10 +199,7 @@ fn note_aborted(id: i64) {
     .flatten()
     .filter(|h| *h != 0);
     if let Some(handle) = handle {
-        aborted()
-            .lock()
-            .unwrap_or_else(|e| e.into_inner())
-            .push(handle);
+        note_aborted_handle(handle);
     }
 }
 

crates/perry-runtime/src/agent.rs

@@ -54,6 +54,7 @@
 
 use std::cell::Cell;
 use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Mutex, PoisonError};
 
 /// Identity of a JS heap (arena + GC) that queued cross-thread work.
 pub type AgentId = u64;
@@ -111,6 +112,26 @@ pub fn owns(owner: AgentId) -> bool {
     owner == current_agent()
 }
 
+/// Purge hooks for agent-tagged queues that live OUTSIDE this crate (#11433).
+///
+/// `perry-stdlib`'s promise-resolution queues (`common::async_bridge`) are
+/// tagged with the enqueuing agent exactly like the timer and thread-result
+/// queues below, and need the same purge when that agent dies — but they cannot
+/// be named from here. A hook registers once per process and runs inside
+/// [`retire_agent`] with the dying agent's id.
+static RETIRE_HOOKS: Mutex<Vec<fn(AgentId)>> = Mutex::new(Vec::new());
+
+/// Register `hook` to run from [`retire_agent`]. Idempotent per function.
+pub fn register_retire_hook(hook: fn(AgentId)) {
+    let mut hooks = RETIRE_HOOKS.lock().unwrap_or_else(PoisonError::into_inner);
+    if !hooks
+        .iter()
+        .any(|existing| std::ptr::fn_addr_eq(*existing, hook))
+    {
+        hooks.push(hook);
+    }
+}
+
 /// Retire a worker agent at thread exit: its arena is about to be unmapped, so
 /// any queue entry still tagged with it points at memory that is about to go
 /// away. Purge those entries rather than leaving them for a drain that can
@@ -141,6 +162,14 @@ pub fn retire_agent(id: AgentId) {
     crate::event_pump::shutdown_agent_loop();
     crate::timer::purge_agent_timers(id);
     crate::thread::purge_agent_thread_results(id);
+    // Copied out so a hook may itself take locks without holding this one.
+    let hooks: Vec<fn(AgentId)> = RETIRE_HOOKS
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner)
+        .clone();
+    for hook in hooks {
+        hook(id);
+    }
     // Deliberately do NOT clear `CURRENT_AGENT`. Clearing it would make
     // `current_agent()` fall back to `PRIMARY_AGENT` for the rest of this
     // thread's life — i.e. a worker that has just torn down its heap would