fix(ext-net): each agent drains only its own socket events; pg/mysql2 in worker_threads no longer hang or misdeliver (#11340)

changelog.d/11444-net-events-per-agent.md

@@ -0,0 +1,7 @@
+**fix(ext-net): each agent drains only its own socket events, so compiled pg and mysql2 inside `worker_threads` no longer hang or misdeliver events (#11340).** perry-ext-net kept one process-wide queue of pending socket events, and every push woke the primary thread. A socket opened on a worker therefore had its events drained by whichever agent's pump ran first.
+
+When the primary took a worker's `'connect'`, `'data'` or `'close'`, it dispatched the event against listeners on the worker's heap. Depending on timing, the event was lost and the worker hung (pg's `connect()`), it ran on the wrong thread (a `TypeError` after the worker exited, as seen with mysql2), or the process crashed.
+
+`pending_events()` now returns the calling agent's queue, keyed by the new `perry_ffi::agent_post::current_agent()`. Sockets, servers and in-flight closes record their owner agent, so an agent's keepalive counts only its own handles. A new gap test, `test_gap_11340_worker_net_events`, runs 25 pg-shaped client sockets from a worker: 2 of 20 runs passed on main, 20 of 20 with this fix. Verified against a live PostgreSQL 16.15 (pg 8.23.0) and MySQL 8.0.46 (mysql2 3.24.4).
+
+A worker that hosts its own server still hangs intermittently. That is a separate defect, filed as #11434.

crates/perry-ext-net/src/adopt.rs

@@ -56,6 +56,7 @@ pub fn adopt_upgraded_tcp_stream(stream: std::net::TcpStream) -> i64 {
         id,
         SocketState {
             tcp_async_id: 0,
+            owner_agent: perry_ffi::agent_post::current_agent(),
             connect_async_id: 0,
             shutdown_async_id: 0,
             awaiting_connect: false,
@@ -144,6 +145,7 @@ pub fn adopt_turnloop_upgrade(id: i64) -> bool {
         id,
         SocketState {
             tcp_async_id: 0,
+            owner_agent: perry_ffi::agent_post::current_agent(),
             connect_async_id: 0,
             shutdown_async_id: 0,
             awaiting_connect: false,

crates/perry-ext-net/src/ipc.rs

@@ -19,6 +19,7 @@ fn allocate_socket() -> i64 {
         id,
         SocketState {
             tcp_async_id: 0,
+            owner_agent: perry_ffi::agent_post::current_agent(),
             connect_async_id: 0,
             shutdown_async_id: 0,
             awaiting_connect: false,

crates/perry-ext-net/src/lib.rs

@@ -175,9 +175,34 @@ pub(crate) mod statics {
         O.get_or_init(|| Mutex::new(HashMap::new()))
     }
 
+    /// The CALLING AGENT's queue of events waiting for its pump.
+    ///
+    /// #11340: this used to be one process-wide queue. Every agent's pump
+    /// drains through `js_ext_net_drain_pending`, and every push wakes the
+    /// primary thread, so a socket opened on a `worker_threads` worker had its
+    /// completions raced for by the primary: whichever pump ran first took
+    /// the worker's `'connect'` / `'data'` / `'close'`. Taken by the primary,
+    /// the event was dispatched against a socket whose listeners live on the
+    /// worker's heap — dropped (the worker then waited forever), delivered to
+    /// the wrong thread (`TypeError`s after the worker exited), or a crash.
+    /// Events are produced on the owning agent's thread (its loop's sink, or
+    /// an FFI call made by its JS), so keying by the agent that pushes is
+    /// keying by the agent that owns the socket.
+    ///
+    /// One queue per agent, created on the agent's first use and never freed:
+    /// the handful of bytes an exited worker's empty queue keeps is the price
+    /// of handing out `&'static` without a teardown hook.
     pub fn pending_events() -> &'static Mutex<Vec<PendingNetEvent>> {
-        static P: OnceLock<Mutex<Vec<PendingNetEvent>>> = OnceLock::new();
-        P.get_or_init(|| Mutex::new(Vec::new()))
+        type Queues = HashMap<u64, &'static Mutex<Vec<PendingNetEvent>>>;
+        static QUEUES: OnceLock<Mutex<Queues>> = OnceLock::new();
+        let agent = perry_ffi::agent_post::current_agent();
+        let mut queues = QUEUES
+            .get_or_init(|| Mutex::new(HashMap::new()))
+            .lock()
+            .unwrap_or_else(|e| e.into_inner());
+        queues
+            .entry(agent)
+            .or_insert_with(|| Box::leak(Box::new(Mutex::new(Vec::new()))))
     }
 
     /// HTTP Agent-owned socket handles are transport facades over the HTTP
@@ -228,6 +253,10 @@ pub(crate) mod statics {
 /// pass instead of needing a second per-server scanner.
 pub(crate) struct ServerState {
     pub async_id: u64,
+    /// The agent that created this server (`perry_ffi::agent_post::current_agent`).
+    /// Only that agent's pump delivers its events, so only that agent's
+    /// keepalive may count it (#11340).
+    pub owner_agent: u64,
     /// Set by `.listen()`. Keeps `has_active_handles` answering yes from the
     /// `listen()` call until the server's registry entry is removed by its
     /// `'close'`, including the window before the bind has been published.
@@ -249,6 +278,10 @@ pub(crate) struct ServerState {
 
 pub(crate) struct SocketState {
     pub(crate) tcp_async_id: u64,
+    /// The agent the socket belongs to — the one whose thread created it
+    /// (`perry_ffi::agent_post::current_agent`). Its events go to that agent's
+    /// queue, so only that agent's keepalive may count it (#11340).
+    pub(crate) owner_agent: u64,
     pub(crate) connect_async_id: u64,
     pub(crate) shutdown_async_id: u64,
     /// True only between `js_net_socket_alloc` and the first
@@ -402,6 +435,7 @@ pub(crate) fn register_turnloop_socket(
         socket_id,
         SocketState {
             tcp_async_id: 0,
+            owner_agent: perry_ffi::agent_post::current_agent(),
             connect_async_id: 0,
             shutdown_async_id: 0,
             awaiting_connect: false,
@@ -444,6 +478,7 @@ impl SocketState {
     pub(crate) fn for_test(awaiting_connect: bool) -> Self {
         SocketState {
             tcp_async_id: 0,
+            owner_agent: perry_ffi::agent_post::current_agent(),
             connect_async_id: 0,
             shutdown_async_id: 0,
             awaiting_connect,
@@ -696,6 +731,7 @@ pub unsafe extern "C" fn js_net_socket_alloc() -> i64 {
         id,
         SocketState {
             tcp_async_id,
+            owner_agent: perry_ffi::agent_post::current_agent(),
             connect_async_id: 0,
             shutdown_async_id: 0,
             awaiting_connect: true,
@@ -753,6 +789,7 @@ pub unsafe extern "C" fn js_net_create_server(
         id,
         ServerState {
             async_id: 0,
+            owner_agent: perry_ffi::agent_post::current_agent(),
             listen_armed: false,
             bound_port: 0,
             bound_host: String::new(),
@@ -1235,6 +1272,7 @@ where
         id,
         SocketState {
             tcp_async_id,
+            owner_agent: perry_ffi::agent_post::current_agent(),
             connect_async_id,
             shutdown_async_id: 0,
             awaiting_connect: false,

crates/perry-ext-net/src/server_state.rs

@@ -395,8 +395,15 @@ pub(crate) fn has_active_handles() -> bool {
     if crate::turnloop_io::close_in_flight() {
         return true;
     }
+    // #11340: only this agent's sockets and servers. A worker's socket is
+    // served by the worker's pump; counting it here kept the primary alive
+    // forever once the worker had exited with that socket's events undrained.
+    let agent = perry_ffi::agent_post::current_agent();
     if statics::sockets().lock().unwrap().values().any(|socket| {
-        socket.refed && !socket.destroyed && (socket.is_open || !socket.awaiting_connect)
+        socket.owner_agent == agent
+            && socket.refed
+            && !socket.destroyed
+            && (socket.is_open || !socket.awaiting_connect)
     }) {
         return true;
     }
@@ -405,7 +412,9 @@ pub(crate) fn has_active_handles() -> bool {
         .unwrap()
         .iter()
         .any(|(id, server)| {
-            (server.listening || server.listen_armed) && crate::bun_tcp::server_keeps_alive(*id)
+            server.owner_agent == agent
+                && (server.listening || server.listen_armed)
+                && crate::bun_tcp::server_keeps_alive(*id)
         })
 }
 

crates/perry-ext-net/src/turnloop_io.rs

@@ -141,25 +141,29 @@ struct Aux {
 /// so a program with nothing else pending exited in between and the
 /// `'close'` never fired. undici's `Client.close()` awaits exactly that
 /// event, so it never settled.
-fn closing() -> &'static Mutex<std::collections::HashSet<i64>> {
-    static CLOSING: OnceLock<Mutex<std::collections::HashSet<i64>>> = OnceLock::new();
-    CLOSING.get_or_init(|| Mutex::new(std::collections::HashSet::new()))
+/// Sockets whose close was submitted and whose `'close'` is still owed, each
+/// with the agent that submitted it (#11340: only that agent waits for it).
+fn closing() -> &'static Mutex<std::collections::HashMap<i64, u64>> {
+    static CLOSING: OnceLock<Mutex<std::collections::HashMap<i64, u64>>> = OnceLock::new();
+    CLOSING.get_or_init(|| Mutex::new(std::collections::HashMap::new()))
 }
 
 /// Whether a submitted socket close still owes its `'close'` event; the
 /// keepalive gate (`server_state::has_active_handles`) waits for it.
 pub(crate) fn close_in_flight() -> bool {
-    !closing()
+    let agent = perry_ffi::agent_post::current_agent();
+    closing()
         .lock()
         .unwrap_or_else(|e| e.into_inner())
-        .is_empty()
+        .values()
+        .any(|owner| *owner == agent)
 }
 
 fn note_close_submitted(id: i64) {
     closing()
         .lock()
         .unwrap_or_else(|e| e.into_inner())
-        .insert(id);
+        .insert(id, perry_ffi::agent_post::current_agent());
 }
 
 fn submit_socket_close(id: i64) -> bool {

crates/perry-ffi/src/agent_post.rs

@@ -46,6 +46,27 @@ extern "C" {
     fn js_perry_agent_post_available() -> i32;
     fn js_perry_agent_post(run: Option<extern "C" fn(*mut c_void)>, ctx: *mut c_void) -> i32;
     fn js_perry_agent_post_dispatched() -> u64;
+    fn js_perry_agent_current() -> u64;
+}
+
+/// The agent the calling thread acts for: its own id on a `worker_threads`
+/// worker, the primary agent's otherwise (including a second thread pumping
+/// on the primary's behalf).
+///
+/// A binding whose loop completions land in a process-wide queue keys that
+/// queue by this, so each agent drains only work that names its own heap
+/// (#11340). The value is an opaque identity: compare it, never interpret it.
+pub fn current_agent() -> u64 {
+    #[cfg(any(not(test), feature = "runtime-link"))]
+    {
+        // SAFETY: a plain thread-local read in the linked runtime.
+        unsafe { js_perry_agent_current() }
+    }
+    #[cfg(all(test, not(feature = "runtime-link")))]
+    {
+        // No runtime is linked: every thread is the one agent there is.
+        0
+    }
 }
 
 /// Work a binding hands to the loop of the agent it is acting for.

crates/perry-runtime/src/turnloop_post/abi.rs

@@ -62,3 +62,15 @@ pub unsafe extern "C" fn js_perry_agent_post(
 pub extern "C" fn js_perry_agent_post_dispatched() -> u64 {
     super::dispatched()
 }
+
+/// The agent the calling thread acts for ([`crate::agent::current_agent`]):
+/// its own if it is a `worker_threads` worker, otherwise the primary agent.
+///
+/// For a binding that keeps a queue of work produced by several agents' loops
+/// and must hand each agent only its own (#11340): a socket event produced on
+/// a worker's loop names objects on the worker's heap, so the primary thread
+/// must never drain it.
+#[no_mangle]
+pub extern "C" fn js_perry_agent_current() -> u64 {
+    crate::agent::current_agent()
+}

scripts/gc_runtime_root_holders.json

@@ -69,12 +69,6 @@
       "verdict": "not_a_gc_pointer",
       "why": "write_tokens(): HashMap<u64 write-token, i64 socket HANDLE>. Each i64 is the same handle-band id ABORTS holds \u2014 the key used to look the socket up in crate::statics::sockets(), never a heap address. Entries are removed on write completion (bun_tcp.rs) and retained-out when a handle closes, so no id outlives its facade."
     },
-    {
-      "file": "crates/perry-ext-net/src/lib.rs",
-      "name": "P",
-      "verdict": "not_a_gc_pointer",
-      "why": "pending_events(): Vec<PendingNetEvent>; every variant carries socket/server ids, Bytes, String, bool, or DropInfo (SocketAddrs) \u2014 no closures or NaN-boxed values. Listener closures live in listeners(), visited by scan_net_roots."
-    },
     {
       "file": "crates/perry-ext-net/src/server_state.rs",
       "name": "STATE",

test-files/_helpers/gap_11340_worker_net_events_worker.ts

@@ -0,0 +1,30 @@
+// The worker half of test_gap_11340_worker_net_events.ts.
+import net from 'node:net';
+import { parentPort } from 'node:worker_threads';
+
+// The parity harness runs a plain-TCP echo server here
+// (test-files/test_net_echo_server.py), the same one test_net_min.ts uses.
+const PORT = 17891;
+const ROUNDS = 25;
+let completed = 0;
+const replies = new Set<string>();
+for (let i = 0; i < ROUNDS; i++) {
+  const reply = await new Promise<string>((resolve) => {
+    // node-postgres' shape (lib/connection.js): construct, connect, THEN subscribe.
+    const s = new net.Socket();
+    s.setNoDelay(true);
+    s.connect(PORT, '127.0.0.1');
+    const want = 'round' + i;
+    let got = '';
+    s.once('connect', () => s.write(want));
+    s.on('data', (d: Buffer) => {
+      got += d.toString();
+      if (got.length >= want.length) s.end();
+    });
+    s.on('close', () => resolve(got));
+    s.on('error', (e: Error) => resolve('error ' + e.message));
+  });
+  if (reply === 'round' + i) completed++;
+  replies.add(reply.replace(/[0-9]+$/, ''));
+}
+parentPort?.postMessage(`rounds ${completed}/${ROUNDS} ${[...replies].sort().join(',')}`);

test-files/test_gap_11340_worker_net_events.ts

@@ -0,0 +1,35 @@
+// #11340: socket events for a socket opened on a `worker_threads` worker must
+// be delivered on that worker. perry-ext-net kept ONE process-wide queue of
+// pending socket events and woke the primary thread on every push, so the
+// primary's pump raced the worker's for them: a worker's `'connect'` taken by
+// the primary was dispatched against listeners on the worker's heap — dropped
+// (the worker waited forever: compiled `pg` hung in `connect()` on a worker),
+// run on the wrong thread (a `TypeError` after the worker exited, as mysql2
+// showed), or a crash.
+//
+// The worker opens 25 sequential client sockets, in node-postgres' shape
+// (construct, `connect()`, then `once('connect')`), to the harness's echo
+// server (test-files/test_net_echo_server.py on 127.0.0.1:17891, the one
+// test_net_min.ts uses). Before the fix the worker stalled within the 25
+// rounds; the primary does nothing but wait.
+import { Worker } from 'node:worker_threads';
+
+const worker = new Worker(new URL('./_helpers/gap_11340_worker_net_events_worker.ts', import.meta.url));
+
+// A lost event presents as a hang; the watchdog turns it into a visible failure.
+const watchdog = setTimeout(() => {
+  console.log('WORKER STOPPED');
+  process.exit(3);
+}, 15000);
+if (typeof (watchdog as { unref?: () => void }).unref === 'function') {
+  (watchdog as { unref: () => void }).unref();
+}
+
+const result = await new Promise<string>((resolve) => {
+  worker.on('message', (value: string) => resolve(String(value)));
+  worker.on('error', (e: Error) => resolve(`worker-error ${e.message}`));
+});
+clearTimeout(watchdog);
+console.log(result);
+await worker.terminate();
+console.log('done');