fix(http): pass the connection socket to server 'connection' listeners

changelog.d/11328-http-connection-event-socket.md

@@ -0,0 +1,38 @@
+Fixed `http.Server`/`https.Server` `'connection'` listeners receiving no
+argument — Node passes the accepted `net.Socket`, the same object later
+exposed as `req.socket` for every request on that connection, and libraries
+key open connections by `conn.remoteAddress + ':' + conn.remotePort` and drop
+them on the socket's `'close'` (e.g. the `server-destroy` package, used by
+Astro's `@astrojs/node` adapter). Perry fired the listener with no args
+(`emit_no_arg_to_listeners`); user code reading `conn.remoteAddress` then
+threw `TypeError: Cannot read properties of undefined`.
+
+Root cause / fix: `PENDING_CONNECTION_EVENTS` queued only the server handle.
+Every accepted turnloop connection now also builds a connection-socket object
+at accept time (`alloc_connection_socket`, reusing the existing
+`IncomingMessage` type rather than a new one — it already has
+`remoteAddress`/`remotePort`, an `on`/`once`/`destroy` dispatch surface, and a
+`'close'` emit), and the queue carries `(server_handle, socket_handle)`. The
+`'connection'` listener now receives that socket; `finish_request` assigns it
+as `req.socket`/`req.connection` for every request on the connection, through
+the override mechanism `#4904` already added for `new
+http.IncomingMessage(socket)`, so `req.socket === conn`. When the connection's
+terminal `Closed` completion arrives, its socket handle is queued (the
+completion sink cannot run JS) and the pump fires `'close'` on it before
+draining the server's own deferred `'close'` callback. HTTP/2's accept path
+gets a socket argument too, through the same shared queue, without wiring
+`req.socket` on H2 streams (out of scope: `Http2ServerRequest` has no such
+alias).
+
+Known gaps left out of scope: `res.socket`/`res.connection` still self-refer
+instead of aliasing the connection socket; `conn.destroy()` fires the socket's
+`'close'` immediately without tearing down the underlying TCP connection (it
+reuses the generic `IncomingMessage.destroy()`, which never closed the
+connection for a plain `req.destroy()` either); the `http2` ALPN-to-HTTP/1.1
+handoff (`adopt_alpn_http1`) still never fires `'connection'` at all (a
+pre-existing gap, unrelated to this fix — `req.socket` is still wired there
+for consistency).
+
+Test: `test-parity/node-suite/http/server/connection-event-socket.ts`
+(argument shape, `req.socket === conn`, `'close'` firing and connection
+bookkeeping via the `server-destroy` idiom, event ordering).

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

@@ -858,6 +858,27 @@ pub(crate) fn emit_no_arg_to_listeners(listeners: &[i64]) {
     }
 }
 
+/// Fire a one-arg event (`server.on('connection', socket)`).
+pub(crate) fn emit_one_arg_to_listeners(listeners: &[i64], arg: f64) {
+    // #8082: the snapshot AND the arg cross every callback — root both.
+    let scope = perry_ffi::TransientRootScope::enter();
+    let rooted = scope.root_addrs(listeners);
+    let arg = scope.root_nanbox(arg);
+    for cb in &rooted {
+        let addr = cb.get();
+        if addr == 0 {
+            continue;
+        }
+        unsafe {
+            let raw = addr as *const RawClosureHeader;
+            let closure = JsClosure::from_raw(raw);
+            if !closure.is_null() {
+                let _ = closure.call1(arg.get());
+            }
+        }
+    }
+}
+
 /// Mark the request as closed, abort `req.signal`, and fire `'close'` once.
 pub(crate) fn close_incoming_message(handle: i64) {
     let close_listeners;
@@ -886,6 +907,32 @@ pub(crate) fn alloc_incoming_message(im: IncomingMessage) -> i64 {
     register_handle(im)
 }
 
+/// Build the JS-visible object exposed to `'connection'` listeners and, for
+/// every request on this connection, as `req.socket`/`req.connection`. One
+/// per accepted connection, shared by every request that connection carries.
+///
+/// Deliberately reuses `IncomingMessage` rather than a new type: it already
+/// has `remoteAddress`/`remotePort`, an `on`/`once`/`destroy` dispatch
+/// surface, and a `'close'` emit (`close_incoming_message`) — exactly the
+/// net.Socket-shaped surface Node exposes here, with no request-specific
+/// state ever populated on it (`method`/`url`/`headers` stay at their
+/// `IncomingMessage::new` defaults and are never Node-accurate for this
+/// object; nothing reads them, since real `net.Socket` has no such
+/// properties either).
+pub(crate) fn alloc_connection_socket(remote_address: String, remote_port: u16) -> i64 {
+    let mut socket = IncomingMessage::new(
+        String::new(),
+        String::new(),
+        HashMap::new(),
+        Vec::new(),
+        Vec::new(),
+        remote_address,
+        remote_port,
+    );
+    socket.complete = true;
+    register_handle(socket)
+}
+
 /// Record only that the response wrote at least one byte. The req/res facade
 /// does not currently share the transport's exact byte counter, so this is a
 /// deliberate nonzero approximation rather than Node's precise total.

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

@@ -14,7 +14,7 @@ use perry_ffi::{
 };
 
 use crate::server::ensure_gc_scanner_registered;
-use crate::server::request::{emit_no_arg_to_listeners, handle_to_pointer_f64, with_implicit_this};
+use crate::server::request::{handle_to_pointer_f64, with_implicit_this};
 use crate::server::response::{ResponseShape, ServerResponse};
 use crate::server::types::{
     extract_host, extract_port, js_handle_clear_side_tables, js_promise_run_microtasks,
@@ -233,13 +233,14 @@ pub struct HttpPendingUpgrade {
     pub head: Vec<u8>,
 }
 
-/// Server handles whose accept loop saw a new connection since the last
-/// pump tick. Drained by `js_node_http_server_process_pending` to fire
-/// `'connection'` listeners on the main thread (#4905). Node passes the
-/// socket as the listener argument; we don't model a net.Socket for
-/// these connections yet, so listeners fire with no args — enough for
-/// the canonical connection-counting idiom.
-pub(crate) static PENDING_CONNECTION_EVENTS: Mutex<Vec<i64>> = Mutex::new(Vec::new());
+/// `(server_handle, socket_handle)` pairs whose accept loop saw a new
+/// connection since the last pump tick. Drained by
+/// `js_node_http_server_process_pending` to fire `'connection'` listeners on
+/// the main thread (#4905). `socket_handle` is the connection's identity
+/// (`crate::server::request::alloc_connection_socket`): the `IncomingMessage`-shaped object Node passes as the
+/// listener's argument and that every request on the connection shares as
+/// `req.socket`.
+pub(crate) static PENDING_CONNECTION_EVENTS: Mutex<Vec<(i64, i64)>> = Mutex::new(Vec::new());
 
 /// Read the `HttpServer` behind a JS server handle, whichever flavour it is.
 ///
@@ -1098,11 +1099,11 @@ 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'`).
-    let connection_events: Vec<i64> = PENDING_CONNECTION_EVENTS
+    let connection_events: Vec<(i64, i64)> = PENDING_CONNECTION_EVENTS
         .lock()
         .map(|mut q| q.drain(..).collect())
         .unwrap_or_default();
-    for server_handle in connection_events {
+    for (server_handle, socket_handle) in connection_events {
         // The handle may back an HttpServer or an HttpsServer (whose
         // accept loop pushes here too since #4971) — probe both.
         let listeners = get_handle_mut::<HttpServer>(server_handle)
@@ -1116,7 +1117,19 @@ pub extern "C" fn js_node_http_server_process_pending() -> i32 {
             continue;
         }
         let this_val = handle_to_pointer_f64(server_handle);
-        with_implicit_this(this_val, || emit_no_arg_to_listeners(&listeners));
+        let socket_val = handle_to_pointer_f64(socket_handle);
+        with_implicit_this(this_val, || {
+            crate::server::request::emit_one_arg_to_listeners(&listeners, socket_val)
+        });
+        count += 1;
+    }
+
+    // Fire `'close'` on every connection
+    // socket whose TCP connection fully closed since the last tick — before
+    // this tick's server `'close'` callback below (`drain_deferred_close_for`),
+    // matching Node's ordering.
+    for socket_handle in crate::server::turnloop_serve::take_closed_sockets() {
+        crate::server::request::close_incoming_message(socket_handle);
         count += 1;
     }
 

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

@@ -40,10 +40,12 @@ pub(crate) fn note_turnloop_request_aborted(request_handle: i64) {
 
 /// Queue the `'connection'` event for a turnloop-accepted connection (P5).
 ///
-/// Drained by the pump, whose listeners fire with no args.
-pub(crate) fn queue_turnloop_connection_event(server_handle: i64) {
+/// Drained by the pump, whose listeners fire with `socket_handle` — the
+/// connection's socket object — as the sole
+/// argument.
+pub(crate) fn queue_turnloop_connection_event(server_handle: i64, socket_handle: i64) {
     if let Ok(mut q) = PENDING_CONNECTION_EVENTS.lock() {
-        q.push(server_handle);
+        q.push((server_handle, socket_handle));
     }
 }
 

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

@@ -494,7 +494,16 @@ fn on_accept(listener_id: i64, conn_id: i64) {
         }
     }
     insert(conn);
-    crate::server::server::queue_turnloop_connection_event(server_handle);
+    // Give HTTP/2's `'connection'` listener a
+    // real socket argument too, same shared drain path as HTTP/1.1. Not
+    // stored on `H2Conn` — HTTP/2 streams don't wire `req.socket` from it
+    // (out of scope here; Node's own `Http2ServerRequest` has no `.socket`
+    // alias to a per-connection object the way `http`'s does).
+    let connection_socket = crate::server::request::alloc_connection_socket(
+        peer.as_ref().map(|e| e.address.clone()).unwrap_or_default(),
+        peer.as_ref().map(|e| e.port).unwrap_or(0),
+    );
+    crate::server::server::queue_turnloop_connection_event(server_handle, connection_socket);
     if let Err(_err) = tl::read_start(conn_id) {
         destroy_connection(conn_id);
         return;

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

@@ -19,7 +19,10 @@ use perry_ffi::turnloop_net as tl;
 use turnloop_http::http1;
 
 use super::wire::{self, Framing};
-use crate::server::request::{alloc_incoming_message, IncomingMessage};
+use crate::server::request::{
+    alloc_connection_socket, alloc_incoming_message, handle_to_pointer_f64, incoming_socket_assign,
+    IncomingMessage,
+};
 use crate::server::response::{alloc_server_response_for_turnloop, ResponseShape};
 use crate::server::server::{with_base_server, HttpPendingRequest};
 
@@ -72,6 +75,11 @@ pub(crate) struct Conn {
     server_handle: i64,
     peer_address: String,
     peer_port: u16,
+    /// The `IncomingMessage`-shaped object
+    /// handed to `server.on('connection', ...)` and shared as `req.socket`
+    /// by every request this connection carries. One per connection, built
+    /// once in `start_connection`/`adopt_alpn_http1`.
+    socket_handle: i64,
     decoder: http1::Decoder,
     input: Vec<u8>,
     building: Option<Building>,
@@ -143,6 +151,28 @@ pub(crate) fn take_aborted() -> Vec<i64> {
     std::mem::take(&mut *queue)
 }
 
+/// 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();
+    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);
+}
+
+/// 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)
+}
+
 /// Note that this connection's in-flight request (if any) will never be
 /// answered, exactly once per request.
 fn note_aborted(id: i64) {
@@ -267,15 +297,19 @@ pub(crate) fn start_connection(
         }
     }
     let peer = tl::peer_address(conn_id);
+    let peer_address = peer.as_ref().map(|e| e.address.clone()).unwrap_or_default();
+    let peer_port = peer.as_ref().map(|e| e.port).unwrap_or(0);
     let keep_alive_timeout_ms =
         with_base_server(server_handle, |s| s.keep_alive_timeout).unwrap_or(5_000.0);
+    let socket_handle = alloc_connection_socket(peer_address.clone(), peer_port);
     conns().lock().unwrap_or_else(|e| e.into_inner()).insert(
         conn_id,
         Conn {
             id: conn_id,
             server_handle,
-            peer_address: peer.as_ref().map(|e| e.address.clone()).unwrap_or_default(),
-            peer_port: peer.as_ref().map(|e| e.port).unwrap_or(0),
+            peer_address,
+            peer_port,
+            socket_handle,
             decoder: http1::Decoder::new(http1::Mode::Request, Default::default()),
             input: Vec::with_capacity(8 * 1024),
             building: None,
@@ -293,7 +327,7 @@ pub(crate) fn start_connection(
             websocket: false,
         },
     );
-    crate::server::server::queue_turnloop_connection_event(server_handle);
+    crate::server::server::queue_turnloop_connection_event(server_handle, socket_handle);
     arm_idle(conn_id);
     if let Err(_err) = tl::read_start(conn_id) {
         destroy_connection(conn_id);
@@ -327,13 +361,19 @@ pub(crate) fn adopt_alpn_http1(
         with_base_server(server_handle, |s| s.keep_alive_timeout).unwrap_or(5_000.0);
     let mut input = Vec::with_capacity(8 * 1024);
     input.extend_from_slice(&leftover);
+    // Pre-existing gap, not this fix's scope: this ALPN handoff never queued
+    // `'connection'` (no `queue_turnloop_connection_event` call below) even
+    // before connection sockets existed. `req.socket` on it is still wired so a
+    // request landing here is consistent with the ordinary accept path.
+    let socket_handle = alloc_connection_socket(peer_address.clone(), peer_port);
     conns().lock().unwrap_or_else(|e| e.into_inner()).insert(
         id,
         Conn {
             id,
             server_handle,
             peer_address,
             peer_port,
+            socket_handle,
             decoder: http1::Decoder::new(http1::Mode::Request, Default::default()),
             input,
             building: None,
@@ -621,6 +661,10 @@ fn finish_request(c: &mut Conn, building: Building) -> (HttpPendingRequest, bool
         "1.1".to_string()
     };
     let im_handle = alloc_incoming_message(im);
+    // `req.socket === conn` for the
+    // `'connection'` listener's argument — same handle, every request on
+    // this connection.
+    incoming_socket_assign(im_handle, handle_to_pointer_f64(c.socket_handle));
     let sr_handle = alloc_server_response_for_turnloop(c.id, c.seq, im_handle);
 
     let is_check_continue = building.expects_continue
@@ -1057,11 +1101,21 @@ fn on_closed(id: i64) {
     // ever passing through `destroy_connection`.
     note_aborted(id);
     cancel_idle(id);
-    let owned = conns()
+    let removed = conns()
         .lock()
         .unwrap_or_else(|e| e.into_inner())
-        .remove(&id)
-        .is_some();
+        .remove(&id);
+    let owned = removed.is_some();
+    // The connection is fully gone — queue its
+    // socket's `'close'` for the pump's next tick. This sink runs inside
+    // `dispatch_staged` and must not run JS (same rule as `note_aborted`
+    // above), so `js_node_http_server_process_pending` (via
+    // `take_closed_sockets`) fires the listeners instead.
+    if let Some(conn) = removed {
+        if conn.socket_handle != 0 {
+            note_closed_socket(conn.socket_handle);
+        }
+    }
     crate::server::server::turnloop_connection_closed(id);
     if owned {
         // The rustls session has to go BEFORE the id does. `turnloop_tls_io`

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

@@ -58,7 +58,8 @@ use perry_ffi::agent_post::{self, AgentJob};
 
 pub(crate) use conn::{
     adopt_alpn_http1, begin_stream, connections_of, destroy_connection, finish_body, is_busy,
-    note_aborted_handle, send_body, send_interim, send_response, take_aborted, take_pending,
+    note_aborted_handle, send_body, send_interim, send_response, take_aborted, take_closed_sockets,
+    take_pending,
 };
 
 /// This crate's slot in the runtime's completion-sink registry.

scripts/gc_runtime_root_holders.json

@@ -43,7 +43,7 @@
       "file": "crates/perry-ext-http/src/server/server.rs",
       "name": "PENDING_CONNECTION_EVENTS",
       "verdict": "not_a_gc_pointer",
-      "why": "Vec of server-handle registry ids drained by js_node_http_server_process_pending to fire 'connection' listeners; the listeners themselves are in HttpServer.listeners, scanned by scan_http_server_roots."
+      "why": "Vec of (server_handle, socket_handle) registry-id pairs drained by js_node_http_server_process_pending to fire 'connection' listeners; socket_handle is a connection-socket IncomingMessage handle (alloc_connection_socket). Both are perry-ffi registry ids, not NaN-boxed values — the listeners themselves are in HttpServer.listeners, scanned by scan_http_server_roots, and the socket's own listeners/signal fields are scanned via iter_handles_of_mut::<IncomingMessage>."
     },
     {
       "file": "crates/perry-ext-http/src/server/server.rs",
@@ -2701,6 +2701,12 @@
       "name": "WINDOW_ROOTS",
       "verdict": "not_a_gc_pointer",
       "why": "Window-root registry maps numeric window handles to numeric root-widget handles; neither value is a JavaScript heap pointer."
+    },
+    {
+      "file": "crates/perry-ext-http/src/server/turnloop_serve/conn.rs",
+      "name": "CLOSED",
+      "verdict": "not_a_gc_pointer",
+      "why": "Vec<i64> of connection-socket handle ids (alloc_connection_socket) whose TCP connection fully closed, queued by note_closed_socket/on_closed and drained by the pump — same shape and same file as the pre-existing ABORTED/aborted() request-handle queue this crate already uses for 'aborted'. A registry id, not a NaN-boxed value: the socket's own listeners/signal fields are scanned via iter_handles_of_mut::<IncomingMessage> in server/mod.rs's scan_http_server_roots."
     }
   ],
   "_FRONTIER_README": "Identity-pinned debt ratchet over new perry-ui* candidates and otherwise-unclassified core raw/Perry TLS declarations (see the census docstring, \u201cThe identity-pinned frontier\u201d). A new uncovered holder fails until it is scanned, receives a researched holders verdict, or is deliberately pinned as debt. Moving a researched false positive to holders graduates it from this list. A fixed or classified holder makes its old frontier pin stale, so the receipt must be deleted.",