diff --git a/changelog.d/11328-http-connection-event-socket.md b/changelog.d/11328-http-connection-event-socket.md new file mode 100644 index 00000000000..e28241ae7f3 --- /dev/null +++ b/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). diff --git a/crates/perry-ext-http/src/server/request.rs b/crates/perry-ext-http/src/server/request.rs index d374ad56fc4..5e03f142b53 100644 --- a/crates/perry-ext-http/src/server/request.rs +++ b/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. diff --git a/crates/perry-ext-http/src/server/server.rs b/crates/perry-ext-http/src/server/server.rs index 65fffaf5f3d..ac3a79eb63f 100644 --- a/crates/perry-ext-http/src/server/server.rs +++ b/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, } -/// 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> = 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> = 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 = 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::(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; } diff --git a/crates/perry-ext-http/src/server/server/turnloop_listen.rs b/crates/perry-ext-http/src/server/server/turnloop_listen.rs index f981290c826..d60565ee5ea 100644 --- a/crates/perry-ext-http/src/server/server/turnloop_listen.rs +++ b/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)); } } diff --git a/crates/perry-ext-http/src/server/turnloop_h2/conn.rs b/crates/perry-ext-http/src/server/turnloop_h2/conn.rs index 0f1fe3931e4..77ccbe6edee 100644 --- a/crates/perry-ext-http/src/server/turnloop_h2/conn.rs +++ b/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; diff --git a/crates/perry-ext-http/src/server/turnloop_serve/conn.rs b/crates/perry-ext-http/src/server/turnloop_serve/conn.rs index a25ab0236cf..645e2f0e9d4 100644 --- a/crates/perry-ext-http/src/server/turnloop_serve/conn.rs +++ b/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, building: Option, @@ -143,6 +151,28 @@ pub(crate) fn take_aborted() -> Vec { 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> { + static CLOSED: OnceLock>> = 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 { + 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,6 +361,11 @@ 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 { @@ -334,6 +373,7 @@ pub(crate) fn adopt_alpn_http1( 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` diff --git a/crates/perry-ext-http/src/server/turnloop_serve/mod.rs b/crates/perry-ext-http/src/server/turnloop_serve/mod.rs index 7e0cd6b1ad4..239e864fdee 100644 --- a/crates/perry-ext-http/src/server/turnloop_serve/mod.rs +++ b/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. diff --git a/scripts/gc_runtime_root_holders.json b/scripts/gc_runtime_root_holders.json index 52d3a43623e..89a93256427 100644 --- a/scripts/gc_runtime_root_holders.json +++ b/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::." }, { "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 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:: 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.", diff --git a/test-parity/node-suite/http/server/connection-event-socket.ts b/test-parity/node-suite/http/server/connection-event-socket.ts new file mode 100644 index 00000000000..6ea861157fb --- /dev/null +++ b/test-parity/node-suite/http/server/connection-event-socket.ts @@ -0,0 +1,47 @@ +// `http.Server` emits `'connection'` with the accepted socket, before any +// request on it. Libraries rely on that argument: `server-destroy` (used by +// Astro's node adapter) keys open connections by +// `conn.remoteAddress + ":" + conn.remotePort` and drops them on the +// socket's `'close'`. +import { createServer, get } from "node:http"; + +const connections: Record = {}; +const seen: any[] = []; +const events: string[] = []; + +const server = createServer((req, res) => { + events.push("request"); + console.log("req.socket is the connection socket:", req.socket === seen[0]); + res.end("ok"); +}); + +server.on("connection", (conn: any) => { + events.push("connection"); + seen.push(conn); + console.log("connection arg type:", typeof conn); + console.log("remoteAddress type:", typeof conn?.remoteAddress); + console.log("remotePort type:", typeof conn?.remotePort); + console.log("has on/destroy:", typeof conn?.on, typeof conn?.destroy); + + const key = conn.remoteAddress + ":" + conn.remotePort; + connections[key] = conn; + console.log("tracked after connect:", Object.keys(connections).length); + conn.on("close", () => { + events.push("socket-close"); + delete connections[key]; + }); +}); + +server.listen(0, "127.0.0.1", () => { + const { port } = server.address() as { port: number }; + get({ host: "127.0.0.1", port, path: "/", agent: false }, (res) => { + res.resume(); + res.on("end", () => { + server.close(() => { + console.log("connection events:", seen.length); + console.log("tracked after close:", Object.keys(connections).length); + console.log("order:", events.join(" > ")); + }); + }); + }); +}); diff --git a/test-parity/node_suite_baseline.json b/test-parity/node_suite_baseline.json index 2ea471f5ac9..e58e35b4711 100644 --- a/test-parity/node_suite_baseline.json +++ b/test-parity/node_suite_baseline.json @@ -84,8 +84,8 @@ "total": 115 }, "http": { - "pass": 22, - "total": 53 + "pass": 23, + "total": 55 }, "http2": { "pass": 32,