From 21ba4a5ef7c15b5f6a03cc4b1cd7b786d214b6f4 Mon Sep 17 00:00:00 2001 From: Perry Bot Date: Fri, 18 Sep 2026 18:15:45 +0000 Subject: [PATCH 1/4] wip: net.Socket surface cluster (#10441 #10442 #10444 #10465) --- .../src/lower_call/native_table/net_events.rs | 117 ++++++- crates/perry-ext-net/src/adopt.rs | 3 + crates/perry-ext-net/src/dispatch.rs | 82 ++++- crates/perry-ext-net/src/handle_exports.rs | 12 +- crates/perry-ext-net/src/ipc.rs | 16 +- crates/perry-ext-net/src/lib.rs | 42 +++ crates/perry-ext-net/src/lifecycle.rs | 208 ++++++++++-- crates/perry-ext-net/src/pipe.rs | 296 ++++++++++++++++++ crates/perry-ext-net/src/socket_events.rs | 10 + .../test_gap_net_socket_surface_cluster.ts | 125 ++++++++ 10 files changed, 883 insertions(+), 28 deletions(-) create mode 100644 crates/perry-ext-net/src/pipe.rs create mode 100644 test-files/test_gap_net_socket_surface_cluster.ts diff --git a/crates/perry-codegen/src/lower_call/native_table/net_events.rs b/crates/perry-codegen/src/lower_call/native_table/net_events.rs index c885104899..5f77f56356 100644 --- a/crates/perry-codegen/src/lower_call/native_table/net_events.rs +++ b/crates/perry-codegen/src/lower_call/native_table/net_events.rs @@ -219,9 +219,61 @@ pub(super) const NET_EVENTS_ROWS: &[NativeModSig] = &[ has_receiver: true, method: "on", class_filter: Some("Socket"), + // #10442 — was `ret: NR_VOID`. `js_ext_net_socket_on` (the runtime + // symbol both this row and `addListener` below call) now returns the + // socket handle (see `perry-ext-net/src/handle_exports.rs`), so a + // typed `const sock: net.Socket` can chain `sock.on(...).on(...)` + // the same way the untyped/`once`/`setNoDelay` paths already did. runtime: "js_ext_net_socket_on", args: &[NA_STR, NA_PTR], - ret: NR_VOID, + ret: NR_HANDLE_ID, + }, + // #10441 — front-inserting variants of `on`. Absent entirely pre-fix: + // a typed `net.Socket` receiver fell through to a plain property read + // for `prependListener`/`prependOnceListener` and got `undefined`, + // matching the untyped-dispatch gap fixed in `dispatch.rs`'s + // `socket_method_name`. + NativeModSig { + module: "net", + has_receiver: true, + method: "prependListener", + class_filter: Some("Socket"), + runtime: "js_net_socket_prepend_listener", + args: &[NA_STR, NA_PTR], + ret: NR_HANDLE_ID, + }, + NativeModSig { + module: "net", + has_receiver: true, + method: "prependOnceListener", + class_filter: Some("Socket"), + runtime: "js_net_socket_prepend_once_listener", + args: &[NA_STR, NA_PTR], + ret: NR_HANDLE_ID, + }, + // #10444 — `net.Socket` is a `stream.Duplex`; `pipe`/`unpipe` had no + // typed-receiver row at all (nor an untyped one — see + // `dispatch.rs`'s `socket_method_name`). `js_net_socket_pipe` returns + // `dest` (an arbitrary JSValue, NOT a socket handle — hence NR_F64, the + // same return kind the generic `stream` table's own `pipe` row uses) + // for chaining, matching Node. + NativeModSig { + module: "net", + has_receiver: true, + method: "pipe", + class_filter: Some("Socket"), + runtime: "js_net_socket_pipe", + args: &[NA_F64, NA_F64], + ret: NR_F64, + }, + NativeModSig { + module: "net", + has_receiver: true, + method: "unpipe", + class_filter: Some("Socket"), + runtime: "js_net_socket_unpipe", + args: &[NA_F64], + ret: NR_HANDLE_ID, }, // Issue #1852 — chainable no-op `net.Socket` option setters. Perry's // TCP transport doesn't model Nagle/keep-alive/idle-timeout or read @@ -385,6 +437,66 @@ pub(super) const NET_EVENTS_ROWS: &[NativeModSig] = &[ args: &[], ret: NR_STR, }, + // #10465 — `writable`/`readable`/`writableEnded`/`readableEnded`/ + // `_writableState`/`_readableState` were entirely absent from this + // table (a typed `net.Socket` read `undefined` for all six; pg's + // `Connection._send` gates every protocol write on `this.stream.writable` + // being truthy, so the audit's client silently dropped its startup + // message and hung until the connection timeout). + NativeModSig { + module: "net", + has_receiver: true, + method: "writable", + class_filter: None, + runtime: "js_net_socket_get_writable", + args: &[], + ret: NR_F64, + }, + NativeModSig { + module: "net", + has_receiver: true, + method: "readable", + class_filter: None, + runtime: "js_net_socket_get_readable", + args: &[], + ret: NR_F64, + }, + NativeModSig { + module: "net", + has_receiver: true, + method: "writableEnded", + class_filter: None, + runtime: "js_net_socket_get_writable_ended", + args: &[], + ret: NR_F64, + }, + NativeModSig { + module: "net", + has_receiver: true, + method: "readableEnded", + class_filter: None, + runtime: "js_net_socket_get_readable_ended", + args: &[], + ret: NR_F64, + }, + NativeModSig { + module: "net", + has_receiver: true, + method: "_writableState", + class_filter: None, + runtime: "js_net_socket_get_writable_state", + args: &[], + ret: NR_OBJ_FROM_JSON_STR, + }, + NativeModSig { + module: "net", + has_receiver: true, + method: "_readableState", + class_filter: None, + runtime: "js_net_socket_get_readable_state", + args: &[], + ret: NR_OBJ_FROM_JSON_STR, + }, NativeModSig { module: "net", has_receiver: true, @@ -511,9 +623,10 @@ pub(super) const NET_EVENTS_ROWS: &[NativeModSig] = &[ has_receiver: true, method: "addListener", class_filter: Some("Socket"), + // #10442 — same fix as the `on` row above (same runtime symbol). runtime: "js_ext_net_socket_on", args: &[NA_STR, NA_PTR], - ret: NR_VOID, + ret: NR_HANDLE_ID, }, NativeModSig { module: "net", diff --git a/crates/perry-ext-net/src/adopt.rs b/crates/perry-ext-net/src/adopt.rs index a100e75136..d439995445 100644 --- a/crates/perry-ext-net/src/adopt.rs +++ b/crates/perry-ext-net/src/adopt.rs @@ -62,6 +62,9 @@ pub fn adopt_upgraded_tcp_stream(stream: tokio::net::TcpStream) -> i64 { remote_addr: remote, raw: None, destroyed: false, + connecting: false, + writable_ended: false, + readable_ended: false, bytes_read: 0, bytes_written: 0, bytes_queued: 0, diff --git a/crates/perry-ext-net/src/dispatch.rs b/crates/perry-ext-net/src/dispatch.rs index dc63586ca4..a4aa357dba 100644 --- a/crates/perry-ext-net/src/dispatch.rs +++ b/crates/perry-ext-net/src/dispatch.rs @@ -72,7 +72,7 @@ pub(crate) fn ensure_runtime_dispatch_registered() { }); } -fn undefined() -> f64 { +pub(crate) fn undefined() -> f64 { f64::from_bits(TAG_UNDEFINED) } @@ -80,7 +80,7 @@ fn null() -> f64 { f64::from_bits(TAG_NULL) } -fn nanbox_handle(handle: i64) -> f64 { +pub(crate) fn nanbox_handle(handle: i64) -> f64 { f64::from_bits(POINTER_TAG | (handle as u64 & POINTER_MASK)) } @@ -176,6 +176,16 @@ fn socket_method_name(prop: &str) -> Option<&'static [u8]> { "on" => Some(b"on"), "addListener" => Some(b"addListener"), "once" => Some(b"once"), + // #10441 — front-inserting variants of `on`/`once`. Missing here + // meant the untyped dispatch fell through to the generic property + // read for these names, which returned `undefined`: calling it was + // a silent no-op instead of a `TypeError`. + "prependListener" => Some(b"prependListener"), + "prependOnceListener" => Some(b"prependOnceListener"), + // #10444 — `net.Socket` is a `stream.Duplex`; `pipe`/`unpipe` were + // entirely absent from this table. + "pipe" => Some(b"pipe"), + "unpipe" => Some(b"unpipe"), "off" => Some(b"off"), "removeListener" => Some(b"removeListener"), "removeAllListeners" => Some(b"removeAllListeners"), @@ -275,6 +285,38 @@ unsafe fn socket_method(handle: i64, method: &str, args: &[f64]) -> Option crate::js_net_socket_on(handle, unbox_to_i64(args[0]), unbox_to_i64(args[1])); nanbox_handle(handle) } + // #10441 — same shape as `once` below, but inserted at the FRONT of + // the listener list. + "prependListener" if args.len() >= 2 => { + crate::js_net_socket_prepend_listener( + handle, + unbox_to_i64(args[0]), + unbox_to_i64(args[1]), + ); + nanbox_handle(handle) + } + "prependOnceListener" if args.len() >= 2 => { + crate::js_net_socket_prepend_once_listener( + handle, + unbox_to_i64(args[0]), + unbox_to_i64(args[1]), + ); + nanbox_handle(handle) + } + // #10444 — forward socket data to `dest` via the same generic + // Get("write")+call duck-typed dispatch the runtime already uses to + // resolve thenables (`crate::pipe`), so `dest` can be any Writable + // representation (another handle-backed socket, a node:stream + // object, …), not just one specific one. + "pipe" if !args.is_empty() => crate::pipe::socket_pipe( + handle, + args[0], + args.get(1).copied().unwrap_or_else(undefined), + ), + "unpipe" => { + crate::pipe::socket_unpipe(handle, args.first().copied().unwrap_or_else(undefined)); + nanbox_handle(handle) + } "connect" if !args.is_empty() => { let arg2 = args.get(1).copied().unwrap_or_else(undefined); let arg3 = args.get(2).copied().unwrap_or_else(undefined); @@ -559,6 +601,42 @@ pub unsafe extern "C" fn js_ext_net_handle_property_dispatch( Some(null()) } else if prop == "destroyed" && crate::js_ext_net_is_socket_handle(handle) != 0 { Some(crate::js_net_socket_get_destroyed(handle)) + } else if crate::js_ext_net_is_socket_handle(handle) != 0 + && matches!( + prop, + "writable" + | "readable" + | "readyState" + | "connecting" + | "pending" + | "writableEnded" + | "readableEnded" + ) + { + // #10465 — the untyped (`(sock: any)`/plain-JS-driver) dispatch path + // had NO arm at all for these; every driver holds its socket through + // an untyped field (`this.stream`), so this — not the typed-receiver + // table in `net_events.rs` — is the path pg/ioredis/iovalkey/ + // @redis/client actually hit. + Some(match prop { + "writable" => crate::js_net_socket_get_writable(handle), + "readable" => crate::js_net_socket_get_readable(handle), + "connecting" => crate::js_net_socket_get_connecting(handle), + "pending" => crate::js_net_socket_get_pending(handle), + "writableEnded" => crate::js_net_socket_get_writable_ended(handle), + "readableEnded" => crate::js_net_socket_get_readable_ended(handle), + _ => f64::from_bits( + JsValue::from_string_ptr(crate::js_net_socket_get_ready_state(handle)).bits(), + ), + }) + } else if prop == "_writableState" && crate::js_ext_net_is_socket_handle(handle) != 0 { + Some(json_str_to_value(crate::js_net_socket_get_writable_state( + handle, + ))) + } else if prop == "_readableState" && crate::js_ext_net_is_socket_handle(handle) != 0 { + Some(json_str_to_value(crate::js_net_socket_get_readable_state( + handle, + ))) } else if crate::js_ext_net_is_socket_handle(handle) != 0 && matches!( prop, diff --git a/crates/perry-ext-net/src/handle_exports.rs b/crates/perry-ext-net/src/handle_exports.rs index 4c8612da70..ae2fe569a1 100644 --- a/crates/perry-ext-net/src/handle_exports.rs +++ b/crates/perry-ext-net/src/handle_exports.rs @@ -54,9 +54,17 @@ pub unsafe extern "C" fn js_net_server_on(handle: i64, event_ptr: i64, cb: i64) entry.entry(event).or_default().push(cb); } +/// #10442 — returns the socket handle so the TYPED `net.Socket` codegen +/// table (`net_events.rs`'s `on`/`addListener` rows, which call this +/// runtime symbol as their `ret: NR_HANDLE_ID` carrier) can chain +/// `sock.on(...).on(...)` instead of reading back `undefined`. The +/// underlying `js_net_socket_on` stays void — it is also the untyped +/// dynamic-dispatch path's registration call in `dispatch.rs`, which +/// already supplies its own handle return separately. #[no_mangle] -pub unsafe extern "C" fn js_ext_net_socket_on(handle: i64, event_ptr: i64, cb: i64) { - js_net_socket_on(handle, event_ptr, cb) +pub unsafe extern "C" fn js_ext_net_socket_on(handle: i64, event_ptr: i64, cb: i64) -> i64 { + js_net_socket_on(handle, event_ptr, cb); + handle } #[no_mangle] diff --git a/crates/perry-ext-net/src/ipc.rs b/crates/perry-ext-net/src/ipc.rs index 159b7230ba..f7d30f17bb 100644 --- a/crates/perry-ext-net/src/ipc.rs +++ b/crates/perry-ext-net/src/ipc.rs @@ -44,6 +44,9 @@ fn allocate_socket() -> (i64, mpsc::UnboundedReceiver) { remote_addr: None, raw: None, destroyed: false, + connecting: true, + writable_ended: false, + readable_ended: false, bytes_read: 0, bytes_written: 0, bytes_queued: 0, @@ -116,6 +119,9 @@ pub(crate) fn register_accepted_transport( remote_addr, raw: None, destroyed: false, + connecting: false, + writable_ended: false, + readable_ended: false, bytes_read: 0, bytes_written: 0, bytes_queued: 0, @@ -150,9 +156,14 @@ pub(crate) fn connect_existing(handle: i64, path: String) { let mut sockets = statics::sockets().lock().unwrap(); match sockets .get_mut(&handle) - .and_then(|socket| socket.pending_rx.take()) + .and_then(|socket| socket.pending_rx.take().map(|rx| (socket, rx))) { - Some(rx) => rx, + Some((socket, rx)) => { + // #10465 — `socket.connect(path)` on a `new net.Socket()` + // starts connecting synchronously, same as the TCP path. + socket.connecting = true; + rx + } None => { push_event(PendingNetEvent::Error( handle, @@ -187,6 +198,7 @@ fn spawn_connect(id: i64, path: String, mut rx: mpsc::UnboundedReceiver i64 { remote_addr: None, raw: None, destroyed: false, + connecting: false, + writable_ended: false, + readable_ended: false, bytes_read: 0, bytes_written: 0, bytes_queued: 0, @@ -995,6 +1028,10 @@ pub unsafe extern "C" fn js_net_socket_method_connect( let connect_async_id = init_provider_with_trigger(b"TCPCONNECTWRAP", tcp_async_id); if let Some(socket) = statics::sockets().lock().unwrap().get_mut(&handle) { socket.connect_async_id = connect_async_id; + // #10465 — `socket.connect(...)` on a `new net.Socket()` starts + // connecting synchronously from the caller's point of view, same as + // the eager `net.connect()` factory. + socket.connecting = true; } let local_server = server_state::begin_local_connect(&host, port); @@ -1020,6 +1057,7 @@ pub unsafe extern "C" fn js_net_socket_method_connect( let remote = tcp.peer_addr().ok(); if let Some(s) = statics::sockets().lock().unwrap().get_mut(&handle) { s.is_open = true; + s.connecting = false; s.local_addr = local; s.remote_addr = remote; } @@ -1084,6 +1122,9 @@ where remote_addr: None, raw: None, destroyed: false, + connecting: true, + writable_ended: false, + readable_ended: false, bytes_read: 0, bytes_written: 0, bytes_queued: 0, @@ -1145,6 +1186,7 @@ where if let Some(s) = statics::sockets().lock().unwrap().get_mut(&id) { s.is_open = true; + s.connecting = false; s.local_addr = local; s.raw_fd = raw_fd; s.remote_addr = remote; diff --git a/crates/perry-ext-net/src/lifecycle.rs b/crates/perry-ext-net/src/lifecycle.rs index b7b1ed1d97..6fddc18fb5 100644 --- a/crates/perry-ext-net/src/lifecycle.rs +++ b/crates/perry-ext-net/src/lifecycle.rs @@ -123,46 +123,152 @@ fn with_socket(handle: i64, default: T, f: impl FnOnce(&crate::SocketState) - /// `handle` must be a registered socket id (raw, NOT NaN-boxed). #[no_mangle] pub unsafe extern "C" fn js_net_socket_get_pending(handle: i64) -> f64 { - nanbox_bool(with_socket(handle, true, |s| !s.is_open && !s.destroyed)) -} - -/// `socket.connecting` — `true` only while a connection attempt is in flight. -/// Perry resolves connects synchronously inside the tokio task, so from the -/// JS side this is `false` before connect and `false` once open — matching -/// Node for the construct-then-inspect path this getter targets. + // #10465 — Node's real getter is `!this._handle || this.connecting`: once + // there is no live handle (never connected, still connecting, OR fully + // closed/destroyed) `pending` reads `true` again — it is NOT simply the + // complement of `destroyed`. A handle already reaped from the registry + // (see the `'close'` teardown in `socket_events.rs`, which removes the + // `SocketState` entry once the `'close'` event has fired) falls through + // to the `true` default below, which is what we want for that case too. + nanbox_bool(with_socket(handle, true, |s| !s.is_open)) +} + +/// `socket.connecting` — `true` from `net.connect()`/`socket.connect()` +/// until the attempt resolves (open, error, or destroy). Backed by +/// `SocketState::connecting` (#10465); pre-fix this was hardcoded `false`, +/// so `readyState` could never report `"opening"` and any caller polling +/// `connecting` during the handshake window saw the wrong value. /// /// # Safety /// /// See [`js_net_socket_get_pending`]. #[no_mangle] -pub unsafe extern "C" fn js_net_socket_get_connecting(_handle: i64) -> f64 { - nanbox_bool(false) +pub unsafe extern "C" fn js_net_socket_get_connecting(handle: i64) -> f64 { + nanbox_bool(with_socket(handle, false, |s| s.connecting)) } /// `socket.destroyed` — `true` once `.destroy()` ran or the peer closed. +/// Defaults to `true` for a handle with no live `SocketState` — the +/// `'close'` teardown removes the entry once its listeners have run, and by +/// then the socket is unambiguously destroyed (#10465; pre-fix this +/// defaulted `false`, so `destroyed` read `false` again after `'close'`). /// /// # Safety /// /// See [`js_net_socket_get_pending`]. #[no_mangle] pub unsafe extern "C" fn js_net_socket_get_destroyed(handle: i64) -> f64 { - nanbox_bool(with_socket(handle, false, |s| s.destroyed)) + nanbox_bool(with_socket(handle, true, |s| s.destroyed)) +} + +/// `socket.writable` — `true` until `.end()`/`.destroy()` flips +/// `writable_ended`. Independent of connect state, matching Node (a fresh +/// `new net.Socket()` is `writable` before it has ever connected). #10465 — +/// pre-fix this property didn't exist at all (read `undefined`). +/// +/// # Safety +/// +/// See [`js_net_socket_get_pending`]. +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_get_writable(handle: i64) -> f64 { + nanbox_bool(with_socket(handle, false, |s| { + !s.destroyed && !s.writable_ended + })) +} + +/// `socket.readable` — `true` until the peer's EOF has been observed (the +/// `'end'` event) or the socket is destroyed. #10465 companion to +/// [`js_net_socket_get_writable`]. +/// +/// # Safety +/// +/// See [`js_net_socket_get_pending`]. +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_get_readable(handle: i64) -> f64 { + nanbox_bool(with_socket(handle, false, |s| { + !s.destroyed && !s.readable_ended + })) +} + +/// `socket.writableEnded` — `true` immediately once `.end()` is called +/// (before the FIN even flushes), matching Node's documented timing. #10465. +/// +/// # Safety +/// +/// See [`js_net_socket_get_pending`]. +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_get_writable_ended(handle: i64) -> f64 { + nanbox_bool(with_socket(handle, true, |s| s.writable_ended)) +} + +/// `socket.readableEnded` — `true` once the `'end'` event has fired. +/// #10465 companion to [`js_net_socket_get_writable_ended`]. +/// +/// # Safety +/// +/// See [`js_net_socket_get_pending`]. +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_get_readable_ended(handle: i64) -> f64 { + nanbox_bool(with_socket(handle, true, |s| s.readable_ended)) +} + +/// `socket._writableState` / `socket._readableState` — Node internals expose +/// a full `WritableState`/`ReadableState` object; drivers that reach into it +/// (pg, ioredis, `@redis/client`) mostly just check `typeof … === "object"` +/// or a couple of scalar fields. #10465: this returns a minimal object +/// carrying the two fields the audited drivers actually read +/// (`ended`/`finished` mirror `writableEnded`, kept in sync with the same +/// `SocketState` bit) rather than a full internal-stream-state shape. +/// +/// # Safety +/// +/// See [`js_net_socket_get_pending`]. +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_get_writable_state(handle: i64) -> *mut StringHeader { + let ended = with_socket(handle, true, |s| s.writable_ended); + let json = format!("{{\"ended\":{ended},\"finished\":{ended}}}"); + alloc_string(&json).as_raw() +} + +/// See [`js_net_socket_get_writable_state`]. +/// +/// # Safety +/// +/// See [`js_net_socket_get_pending`]. +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_get_readable_state(handle: i64) -> *mut StringHeader { + let ended = with_socket(handle, true, |s| s.readable_ended); + let json = format!("{{\"ended\":{ended}}}"); + alloc_string(&json).as_raw() } /// `socket.readyState` — one of `"opening" | "open" | "readOnly" | -/// "writeOnly" | "closed"`. Node reports `"open"` for a freshly constructed -/// socket and `"closed"` once destroyed. +/// "writeOnly" | "closed"`. Mirrors Node's real getter (`connecting` ? +/// `"opening"` : `readable && writable` ? `"open"` : `readable` ? +/// `"readOnly"` : `writable` ? `"writeOnly"` : `"closed"`) instead of the +/// pre-#10465 two-state `destroyed ? "closed" : "open"`, which could never +/// report `"opening"` (mid-connect) or `"readOnly"` (after `.end()`, before +/// the peer's FIN). /// /// # Safety /// /// See [`js_net_socket_get_pending`]. #[no_mangle] pub unsafe extern "C" fn js_net_socket_get_ready_state(handle: i64) -> *mut StringHeader { - let state = with_socket( - handle, - "open", - |s| if s.destroyed { "closed" } else { "open" }, - ); + let state = with_socket(handle, "closed", |s| { + if s.connecting { + "opening" + } else { + let writable = !s.destroyed && !s.writable_ended; + let readable = !s.destroyed && !s.readable_ended; + match (readable, writable) { + (true, true) => "open", + (true, false) => "readOnly", + (false, true) => "writeOnly", + (false, false) => "closed", + } + } + }); alloc_string(state).as_raw() } @@ -513,6 +619,9 @@ pub unsafe extern "C" fn js_ext_net_socket_end(handle: i64, chunk_bits: i64) { } } } + // #10465 — `writableEnded` (and `writable`) flip as soon as `.end()` + // is CALLED, per Node's docs, not once the FIN actually flushes. + s.writable_ended = true; let _ = s.cmd_tx.send(crate::SocketCommand::End(0)); } } @@ -570,6 +679,8 @@ pub unsafe extern "C" fn js_ext_net_socket_end3( socket.bytes_queued = socket.bytes_queued.saturating_add(byte_len); } } + // #10465 — see the sibling note in `js_ext_net_socket_end`. + socket.writable_ended = true; if socket .cmd_tx .send(crate::SocketCommand::End(completion)) @@ -670,18 +781,31 @@ pub(crate) fn event_name_from_ptr(event_ptr: i64) -> Option { } fn register_listener_with_flag(handle: i64, event: String, cb: i64, once: bool) { + register_listener(handle, event, cb, once, false); +} + +/// #10441 — shared by `on`/`once`/`prependListener`/`prependOnceListener`. +/// `prepend` inserts at the FRONT of the listener vector instead of pushing +/// at the back, which is the only difference Node's `prependListener` has +/// from `addListener`/`on` (same once-flag bookkeeping, same pending-data +/// release for a first `'data'` listener). +fn register_listener(handle: i64, event: String, cb: i64, once: bool, prepend: bool) { if cb == 0 { return; } let releases_pending_data = event == "data"; { let mut listeners = statics::listeners().lock().unwrap(); - listeners + let vec = listeners .entry(handle) .or_default() .entry(event.clone()) - .or_default() - .push(cb); + .or_default(); + if prepend { + vec.insert(0, cb); + } else { + vec.push(cb); + } } if once { let mut flags = statics::once_flags().lock().unwrap(); @@ -885,6 +1009,50 @@ pub unsafe extern "C" fn js_net_socket_once(handle: i64, event_ptr: i64, cb: i64 handle } +/// `socket.prependListener(event, cb)` — like `.on()`/`.addListener()` but +/// inserts at the FRONT of the listener list, so this callback fires before +/// any listener already registered for `event`. #10441: pre-fix, neither the +/// dynamic (untyped-receiver) dispatch nor the typed `net.Socket` codegen +/// table had an entry for this method at all — it silently read `undefined` +/// and calling it was a no-op (ioredis/iovalkey's RESP parser attach via +/// `stream.prependListener("data", …)` never saw a byte). +/// +/// # Safety +/// +/// Same as [`js_net_socket_once`]. +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_prepend_listener( + handle: i64, + event_ptr: i64, + cb: i64, +) -> i64 { + crate::ensure_gc_scanner_registered(); + if let Some(event) = read_event(event_ptr) { + register_listener(handle, event, cb, false, true); + } + handle +} + +/// `socket.prependOnceListener(event, cb)` — the front-inserting, one-shot +/// combination of [`js_net_socket_prepend_listener`] and +/// [`js_net_socket_once`]. #10441. +/// +/// # Safety +/// +/// Same as [`js_net_socket_once`]. +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_prepend_once_listener( + handle: i64, + event_ptr: i64, + cb: i64, +) -> i64 { + crate::ensure_gc_scanner_registered(); + if let Some(event) = read_event(event_ptr) { + register_listener(handle, event, cb, true, true); + } + handle +} + /// `socket.removeListener(event, cb)` — remove the first matching cb. /// /// # Safety diff --git a/crates/perry-ext-net/src/pipe.rs b/crates/perry-ext-net/src/pipe.rs new file mode 100644 index 0000000000..48754ed975 --- /dev/null +++ b/crates/perry-ext-net/src/pipe.rs @@ -0,0 +1,296 @@ +//! #10444 — `net.Socket.prototype.pipe(dest[, options])` / `.unpipe(dest?)`. +//! +//! A `net.Socket` lives behind ext-net's own handle registry +//! (`statics::sockets()` / `statics::listeners()`), a completely different +//! representation from node:stream's own object+closure model +//! (`perry-runtime`'s `node_stream` module, which backs `Readable`/ +//! `Writable`/`Duplex`/`Transform`/`PassThrough`). Bridging those two +//! independent state machines so a socket could reuse node:stream's own +//! `pipe()` implementation (with its full backpressure/unpipe-on-error/ +//! `'pipe'`+`'unpipe'` event machinery) is a much larger undertaking than +//! this cluster fix covers. +//! +//! Instead this reuses the SAME generic `Get(dest, "write")` + call +//! duck-typed dispatch the runtime already relies on to resolve thenables +//! (`crate::promise::assimilate::assimilate_via_then_property` in +//! `perry-runtime`, which does `Get(value, "then")` then invokes it with +//! `this` bound to the thenable): fetch `dest.write` / `dest.end` by name +//! through `js_dynamic_object_get_property` and invoke whatever comes back +//! through `js_native_call_value` with `dest` as the implicit receiver. +//! That resolves correctly regardless of what representation `dest` is — +//! another handle-backed socket, a node:stream object, or a plain user +//! object that overrides `write` — the same way real Node duck-types its +//! destination. +//! +//! Scope: this forwards `'data'` to `dest.write(chunk)` and (unless +//! `{ end: false }`) calls `dest.end()` once the source's `'end'` fires, and +//! returns `dest` for chaining. It does NOT implement automatic +//! unpipe-on-error, backpressure-aware pause/resume of the source, or the +//! `'pipe'`/`'unpipe'` events on the destination that Node's real +//! `Readable.prototype.pipe` fires — those are follow-up work, not part of +//! the #10444 reproduction (a `PassThrough`/`Transform` destination reading +//! everything a socket writes). + +use std::collections::HashMap; +use std::sync::{Mutex, OnceLock}; + +use perry_ffi::{ + alloc_closure, closure_capture_f64, register_closure_arity, set_closure_capture_f64, + RawClosureHeader, +}; + +use crate::statics; + +const TAG_UNDEFINED_BITS: u64 = 0x7FFC_0000_0000_0001; +const TAG_NULL_BITS: u64 = 0x7FFC_0000_0000_0002; +const TAG_FALSE_BITS: u64 = 0x7FFC_0000_0000_0003; + +// `js_dynamic_object_get_property` / `js_implicit_this_set` / +// `js_native_call_value` aren't wrapped by perry-ffi (unlike the closure +// helpers above); declare them the same way `dispatch.rs` declares its own +// direct `perry-runtime` FFI symbols (`js_class_method_bind`, +// `js_promise_resolve`, …) — resolved at final link time, not a Rust-level +// crate dependency. +extern "C" { + fn js_dynamic_object_get_property( + obj_value: f64, + property_name_ptr: *const i8, + property_name_len: usize, + ) -> f64; + fn js_implicit_this_set(value: f64) -> f64; + fn js_native_call_value(func_value: f64, args_ptr: *const f64, args_len: usize) -> f64; +} + +fn is_nullish(v: f64) -> bool { + let bits = v.to_bits(); + bits == TAG_UNDEFINED_BITS || bits == TAG_NULL_BITS +} + +fn is_callable(v: f64) -> bool { + // Native closures / bound handle methods / class methods are all + // POINTER_TAG (0x7FFD) or the handle-method-bind shape; a non-callable + // `Get` result (missing property, a plain data field) is either + // undefined or some other tag entirely. This mirrors the coarse + // callable check `assimilate_via_then_property` uses before invoking a + // fetched `then` — good enough to avoid calling `undefined()` when + // `dest` has no `write` at all, without re-implementing full + // `IsCallable`. + !is_nullish(v) && (v.to_bits() >> 48) == 0x7FFD +} + +/// `Get(dest, "write")(chunk)` with `this` bound to `dest`. +fn generic_write(dest: f64, chunk: f64) { + unsafe { + let write_fn = js_dynamic_object_get_property(dest, c"write".as_ptr(), 5); + if !is_callable(write_fn) { + return; + } + let prev = js_implicit_this_set(dest); + let args = [chunk]; + let _ = js_native_call_value(write_fn, args.as_ptr(), args.len()); + js_implicit_this_set(prev); + } +} + +/// `Get(dest, "end")()` with `this` bound to `dest`. +fn generic_end(dest: f64) { + unsafe { + let end_fn = js_dynamic_object_get_property(dest, c"end".as_ptr(), 3); + if !is_callable(end_fn) { + return; + } + let prev = js_implicit_this_set(dest); + let _ = js_native_call_value(end_fn, std::ptr::null(), 0); + js_implicit_this_set(prev); + } +} + +extern "C" fn pipe_data_forward(closure: *const RawClosureHeader, chunk: f64) -> f64 { + if !closure.is_null() { + let dest = unsafe { closure_capture_f64(closure, 0) }; + generic_write(dest, chunk); + } + f64::from_bits(TAG_UNDEFINED_BITS) +} + +extern "C" fn pipe_end_forward(closure: *const RawClosureHeader) -> f64 { + if !closure.is_null() { + let dest = unsafe { closure_capture_f64(closure, 0) }; + let end_on_finish = unsafe { closure_capture_f64(closure, 1) }; + if end_on_finish.to_bits() != TAG_FALSE_BITS { + generic_end(dest); + } + } + f64::from_bits(TAG_UNDEFINED_BITS) +} + +static ARITY_REGISTERED: std::sync::Once = std::sync::Once::new(); + +fn ensure_pipe_closure_arities_registered() { + ARITY_REGISTERED.call_once(|| { + register_closure_arity(pipe_data_forward as *const u8, 1); + register_closure_arity(pipe_end_forward as *const u8, 0); + }); +} + +/// One socket -> destination pipe route, tracked so `unpipe` can remove +/// exactly the listener closures a matching `pipe()` call installed. +struct PipeRoute { + dest_bits: u64, + data_cb: i64, + end_cb: i64, +} + +fn pipe_routes() -> &'static Mutex>> { + static ROUTES: OnceLock>>> = OnceLock::new(); + ROUTES.get_or_init(|| Mutex::new(HashMap::new())) +} + +/// Register `data_cb`/`end_cb` as normal `'data'`/`'end'` listeners on +/// `handle`, reusing the SAME `statics::listeners()` registry every other +/// socket listener goes through — so they get the same GC-root scanning +/// (`gc_roots::scan_net_roots`) and the same dispatch path +/// (`socket_events::js_ext_net_drain_pending`) as a user's own `.on(...)`, +/// with no new plumbing. +fn install_pipe_listeners(handle: i64, data_cb: i64, end_cb: i64) { + let mut listeners = statics::listeners().lock().unwrap(); + let per_socket = listeners.entry(handle).or_default(); + per_socket + .entry("data".to_string()) + .or_default() + .push(data_cb); + per_socket + .entry("end".to_string()) + .or_default() + .push(end_cb); +} + +fn uninstall_pipe_listeners(handle: i64, data_cb: i64, end_cb: i64) { + let mut listeners = statics::listeners().lock().unwrap(); + if let Some(per_socket) = listeners.get_mut(&handle) { + if let Some(vec) = per_socket.get_mut("data") { + vec.retain(|cb| *cb != data_cb); + } + if let Some(vec) = per_socket.get_mut("end") { + vec.retain(|cb| *cb != end_cb); + } + } +} + +/// `socket.pipe(dest[, options])`. Returns `dest` unchanged (Node's +/// chaining contract), or `undefined` when `dest` is missing/nullish. +pub(crate) fn socket_pipe(handle: i64, dest: f64, options: f64) -> f64 { + if is_nullish(dest) { + return f64::from_bits(TAG_UNDEFINED_BITS); + } + crate::ensure_gc_scanner_registered(); + ensure_pipe_closure_arities_registered(); + + let end_on_finish = unsafe { + if is_nullish(options) { + f64::from_bits(0x7FFC_0000_0000_0004) // default true + } else { + let v = js_dynamic_object_get_property(options, c"end".as_ptr(), 3); + if is_nullish(v) { + f64::from_bits(0x7FFC_0000_0000_0004) + } else { + v + } + } + }; + + let data_closure = alloc_closure(pipe_data_forward as *const u8, 1); + let end_closure = alloc_closure(pipe_end_forward as *const u8, 2); + if data_closure.is_null() || end_closure.is_null() { + return f64::from_bits(TAG_UNDEFINED_BITS); + } + unsafe { + set_closure_capture_f64(data_closure, 0, dest); + set_closure_capture_f64(end_closure, 0, dest); + set_closure_capture_f64(end_closure, 1, end_on_finish); + } + let data_cb = data_closure as i64; + let end_cb = end_closure as i64; + install_pipe_listeners(handle, data_cb, end_cb); + pipe_routes() + .lock() + .unwrap() + .entry(handle) + .or_default() + .push(PipeRoute { + dest_bits: dest.to_bits(), + data_cb, + end_cb, + }); + + dest +} + +/// `socket.unpipe([dest])`. Removes the pipe route(s) installed by a prior +/// `pipe()` call — all of them when `dest` is omitted, only the ones whose +/// destination matches otherwise. Always returns the socket handle. +pub(crate) fn socket_unpipe(handle: i64, dest: f64) { + let filter_bits = (!is_nullish(dest)).then(|| dest.to_bits()); + let removed: Vec<(i64, i64)> = { + let mut routes = pipe_routes().lock().unwrap(); + let Some(list) = routes.get_mut(&handle) else { + return; + }; + let mut removed = Vec::new(); + list.retain(|route| { + let matches = filter_bits.is_none_or(|bits| bits == route.dest_bits); + if matches { + removed.push((route.data_cb, route.end_cb)); + } + !matches + }); + if list.is_empty() { + routes.remove(&handle); + } + removed + }; + for (data_cb, end_cb) in removed { + uninstall_pipe_listeners(handle, data_cb, end_cb); + } +} + +/// Drop every tracked pipe route for `handle` without touching the listener +/// registry — called from the `'close'` teardown, which already clears the +/// whole `statics::listeners()` entry for `handle` (see +/// `socket_events::js_ext_net_drain_pending`'s `Close` arm), so removing the +/// individual callbacks there would be redundant. +pub(crate) fn drop_routes(handle: i64) { + pipe_routes().lock().unwrap().remove(&handle); +} + +// ─── FFI: typed `net.Socket.prototype.pipe`/`.unpipe` ──────────────────────── +// +// The `NativeModSig` rows in +// `crates/perry-codegen/src/lower_call/native_table/net_events.rs` call +// these two symbols directly for a statically-typed `net.Socket` receiver. +// The untyped/dynamic-dispatch path (`dispatch.rs`'s `socket_method`) calls +// `socket_pipe`/`socket_unpipe` above instead of going through here, since +// it already has its own handle-nanboxing conventions. + +/// `socket.pipe(dest[, options])` for a statically-typed `net.Socket` +/// receiver. See the module doc for what this does and does not implement. +/// +/// # Safety +/// +/// `dest`/`options` must be valid NaN-boxed JS values (or `undefined`). +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_pipe(handle: i64, dest: f64, options: f64) -> f64 { + socket_pipe(handle, dest, options) +} + +/// `socket.unpipe([dest])` for a statically-typed `net.Socket` receiver. +/// Returns the socket handle for chaining, matching Node. +/// +/// # Safety +/// +/// `dest` must be a valid NaN-boxed JS value (or `undefined`). +#[no_mangle] +pub unsafe extern "C" fn js_net_socket_unpipe(handle: i64, dest: f64) -> i64 { + socket_unpipe(handle, dest); + handle +} diff --git a/crates/perry-ext-net/src/socket_events.rs b/crates/perry-ext-net/src/socket_events.rs index f644506c14..def81b921e 100644 --- a/crates/perry-ext-net/src/socket_events.rs +++ b/crates/perry-ext-net/src/socket_events.rs @@ -220,6 +220,11 @@ pub unsafe extern "C" fn js_ext_net_drain_pending() -> i32 { lifecycle::drain_once_listeners(id, "error"); } PendingNetEvent::End(id) => { + // #10465 — `readableEnded` (and `readable`) flip as part of + // emitting `'end'`, before any listener runs, matching Node. + if let Some(socket) = statics::sockets().lock().unwrap().get_mut(&id) { + socket.readable_ended = true; + } // Issue #1852 — readable side ended (peer FIN). Fire the // `'end'` listeners; the trailing `Close` event (pushed // right after `End` in `run_socket_task`) does the actual @@ -261,6 +266,11 @@ pub unsafe extern "C" fn js_ext_net_drain_pending() -> i32 { statics::http_agent_phases().lock().unwrap().remove(&id); statics::max_listeners().lock().unwrap().remove(&id); server_state::discard_pending_server_data(id); + // #10444 — the listener-map entry above just went away, so + // any pipe route's tracked callback pointers are dangling; + // drop the tracking table entry too (nothing left to + // uninstall from). + crate::pipe::drop_routes(id); } // Issue #1123 followup — server-side events. The // accept loop pushes `ServerConnection`/`ServerListening`/ diff --git a/test-files/test_gap_net_socket_surface_cluster.ts b/test-files/test_gap_net_socket_surface_cluster.ts new file mode 100644 index 0000000000..79e0afd887 --- /dev/null +++ b/test-files/test_gap_net_socket_surface_cluster.ts @@ -0,0 +1,125 @@ +// #10441/#10442/#10444/#10465 — the `net.Socket` surface cluster the package +// audit hit while compiling real socket-backed npm drivers (mysql2, pg, +// redis, ws) natively instead of through Perry's hand-written bindings: +// +// #10441 — prependListener/prependOnceListener were missing entirely +// (silently did nothing; ioredis/iovalkey's RESP parser attach +// via `stream.prependListener("data", …)` never saw a byte). +// #10442 — on()/addListener() returned `undefined` on a TYPED `net.Socket` +// receiver, breaking `sock.on(...).on(...)` chaining. +// #10444 — pipe() didn't exist on `net.Socket` at all (mongodb's +// Connection constructor does `.pipe(new SizedMessageTransform)`). +// #10465 — writable/readable/_writableState/_readableState were missing, +// and readyState/connecting/pending/destroyed didn't track the +// real connect/end/close lifecycle. +// +// One flow exercises all four against a real loopback echo server so the +// output is compared byte-for-byte against Node instead of spot-checked. + +import * as net from "node:net"; +import { PassThrough } from "node:stream"; + +const server = net.createServer((conn) => { + conn.on("data", (d) => conn.write(d)); + // Explicit half-close instead of relying on Node's default + // allowHalfOpen=false auto-end, so this test only exercises the four + // issues above, not the server's own half-open behavior. + conn.on("end", () => conn.end()); +}); + +function show(label: string, s: net.Socket) { + console.log( + label.padEnd(12), + "writable=" + s.writable, + "readable=" + s.readable, + "readyState=" + s.readyState, + "connecting=" + s.connecting, + "pending=" + s.pending, + "destroyed=" + s.destroyed, + ); +} + +server.listen(0, "127.0.0.1", () => { + const port = (server.address() as net.AddressInfo).port; + + // ── #10465: a never-connected socket ────────────────────────────────── + const fresh = new net.Socket(); + show("new Socket", fresh); + console.log( + "new Socket _writableState=" + typeof fresh._writableState, + "_readableState=" + typeof fresh._readableState, + ); + fresh.destroy(); + + // Typed receiver end to end — `net.Socket`, not `any` — since #10442's + // defect only reproduced on a statically typed receiver. + const sock: net.Socket = net.connect(port, "127.0.0.1"); + show("connecting", sock); + + // ── #10442: on()/addListener() return value + chaining ──────────────── + console.log("on() returns socket:", sock.on("noop-event", () => {}) === sock); + console.log( + "addListener() returns socket:", + sock.addListener("noop-event", () => {}) === sock, + ); + try { + sock.on("__chain_a", () => {}).on("__chain_b", () => {}); + console.log("chained on().on() ok"); + } catch (e: any) { + console.log("chained on().on() threw:", e.message); + } + + // ── #10441: prependListener/prependOnceListener ──────────────────────── + const order: string[] = []; + sock.on("data", () => order.push("normal")); + const prependRet = sock.prependListener("data", () => order.push("prepend")); + console.log("prependListener returns socket:", prependRet === sock); + const prependOnceRet = sock.prependOnceListener("data", () => order.push("prependOnce")); + console.log("prependOnceListener returns socket:", prependOnceRet === sock); + + sock.on("connect", () => { + show("connected", sock); + const anySock: any = sock; + console.log( + "untyped read:", + "writable=" + anySock.writable, + "readable=" + anySock.readable, + "readyState=" + anySock.readyState, + ); + + sock.once("data", (chunk: Buffer) => { + console.log("data order :", order.join(",")); + console.log("data payload:", JSON.stringify(chunk.toString())); + // Clear the marker listeners before wiring pipe() below so they don't + // also fire on the piped chunk. + sock.removeAllListeners("data"); + + // ── #10444: pipe() ────────────────────────────────────────────── + const dest = new PassThrough(); + const pipeRet = sock.pipe(dest); + console.log("pipe() returns dest:", pipeRet === dest); + dest.on("data", (piped: Buffer) => { + console.log("piped payload:", JSON.stringify(piped.toString())); + show("mid-stream", sock); + sock.end(); + }); + sock.write("piped-chunk"); + }); + + sock.write("first-chunk"); + }); + + sock.on("end", () => { + show("'end'", sock); + }); + + sock.on("close", () => { + show("'close'", sock); + server.close(); + }); + + sock.on("error", (e: any) => { + console.log("socket error:", e.message); + server.close(); + }); +}); From 0c0155c056e0dcd46b8e1a178f450dcbf20b6fd2 Mon Sep 17 00:00:00 2001 From: Perry Bot Date: Fri, 18 Sep 2026 18:37:28 +0000 Subject: [PATCH 2/4] fix: net.Socket pending/destroyed timing (mark_closed too-early flip) --- crates/perry-ext-net/src/adopt.rs | 1 + crates/perry-ext-net/src/gc_roots.rs | 4 ++ crates/perry-ext-net/src/ipc.rs | 3 ++ crates/perry-ext-net/src/lib.rs | 42 ++++++++++++++----- crates/perry-ext-net/src/lifecycle.rs | 28 ++++++++++--- crates/perry-ext-net/src/pipe.rs | 49 +++++++++++++++++++---- crates/perry-ext-net/src/socket_events.rs | 12 ++++++ 7 files changed, 115 insertions(+), 24 deletions(-) diff --git a/crates/perry-ext-net/src/adopt.rs b/crates/perry-ext-net/src/adopt.rs index d439995445..64be09a5e2 100644 --- a/crates/perry-ext-net/src/adopt.rs +++ b/crates/perry-ext-net/src/adopt.rs @@ -63,6 +63,7 @@ pub fn adopt_upgraded_tcp_stream(stream: tokio::net::TcpStream) -> i64 { raw: None, destroyed: false, connecting: false, + has_opened: true, writable_ended: false, readable_ended: false, bytes_read: 0, diff --git a/crates/perry-ext-net/src/gc_roots.rs b/crates/perry-ext-net/src/gc_roots.rs index af2e4c248c..334dd00d95 100644 --- a/crates/perry-ext-net/src/gc_roots.rs +++ b/crates/perry-ext-net/src/gc_roots.rs @@ -69,4 +69,8 @@ pub(crate) fn scan_net_roots(visitor: &mut GcRootVisitor<'_>) { // #8259 — the pump's in-flight dispatch frames (snapshotted callbacks + // parked payloads), which the table walks above cannot see. dispatch_custody::scan(visitor); + // #10444 — `pipe()`'s own closure-pointer bookkeeping (see the doc on + // `pipe::scan_roots` for why it needs its own visit despite the SAME + // pointers already being visited via `statics::listeners()` above). + crate::pipe::scan_roots(visitor); } diff --git a/crates/perry-ext-net/src/ipc.rs b/crates/perry-ext-net/src/ipc.rs index f7d30f17bb..1c2eee4cbf 100644 --- a/crates/perry-ext-net/src/ipc.rs +++ b/crates/perry-ext-net/src/ipc.rs @@ -45,6 +45,7 @@ fn allocate_socket() -> (i64, mpsc::UnboundedReceiver) { raw: None, destroyed: false, connecting: true, + has_opened: false, writable_ended: false, readable_ended: false, bytes_read: 0, @@ -120,6 +121,7 @@ pub(crate) fn register_accepted_transport( raw: None, destroyed: false, connecting: false, + has_opened: true, writable_ended: false, readable_ended: false, bytes_read: 0, @@ -198,6 +200,7 @@ fn spawn_connect(id: i64, path: String, mut rx: mpsc::UnboundedReceiver i64 { raw: None, destroyed: false, connecting: false, + has_opened: false, writable_ended: false, readable_ended: false, bytes_read: 0, @@ -1057,6 +1076,7 @@ pub unsafe extern "C" fn js_net_socket_method_connect( let remote = tcp.peer_addr().ok(); if let Some(s) = statics::sockets().lock().unwrap().get_mut(&handle) { s.is_open = true; + s.has_opened = true; s.connecting = false; s.local_addr = local; s.remote_addr = remote; @@ -1123,6 +1143,7 @@ where raw: None, destroyed: false, connecting: true, + has_opened: false, writable_ended: false, readable_ended: false, bytes_read: 0, @@ -1186,6 +1207,7 @@ where if let Some(s) = statics::sockets().lock().unwrap().get_mut(&id) { s.is_open = true; + s.has_opened = true; s.connecting = false; s.local_addr = local; s.raw_fd = raw_fd; diff --git a/crates/perry-ext-net/src/lifecycle.rs b/crates/perry-ext-net/src/lifecycle.rs index 6fddc18fb5..85aa85d6af 100644 --- a/crates/perry-ext-net/src/lifecycle.rs +++ b/crates/perry-ext-net/src/lifecycle.rs @@ -123,14 +123,30 @@ fn with_socket(handle: i64, default: T, f: impl FnOnce(&crate::SocketState) - /// `handle` must be a registered socket id (raw, NOT NaN-boxed). #[no_mangle] pub unsafe extern "C" fn js_net_socket_get_pending(handle: i64) -> f64 { - // #10465 — Node's real getter is `!this._handle || this.connecting`: once - // there is no live handle (never connected, still connecting, OR fully - // closed/destroyed) `pending` reads `true` again — it is NOT simply the - // complement of `destroyed`. A handle already reaped from the registry - // (see the `'close'` teardown in `socket_events.rs`, which removes the + // #10465 — Node's real getter is `!this._handle`: once there is no live + // handle (never connected, still connecting, OR fully closed/destroyed) + // `pending` reads `true` again — it is NOT simply the complement of + // `destroyed`. A handle already reaped from the registry (see the + // `'close'` teardown in `socket_events.rs`, which removes the // `SocketState` entry once the `'close'` event has fired) falls through // to the `true` default below, which is what we want for that case too. - nanbox_bool(with_socket(handle, true, |s| !s.is_open)) + // + // Deliberately keyed on `has_opened`/`destroyed`, NOT `is_open`: + // `is_open` flips false via `server_state::mark_socket_closed`, called + // from the tokio task thread as soon as teardown STARTS (before the main + // thread has processed the `'end'`/`'close'` events that same teardown + // just queued), while `destroyed` only flips at `'close'`-processing + // time — the one point that actually agrees with Node's own timing (see + // the `Close` arm in `socket_events.rs`). Once a socket has opened at + // least once, "does it have a live handle" reduces to "has it been + // destroyed yet", not to the (earlier-flipping) `is_open` flag. + nanbox_bool(with_socket(handle, true, |s| { + if s.has_opened { + s.destroyed + } else { + true + } + })) } /// `socket.connecting` — `true` from `net.connect()`/`socket.connect()` diff --git a/crates/perry-ext-net/src/pipe.rs b/crates/perry-ext-net/src/pipe.rs index 48754ed975..5fcf2bde94 100644 --- a/crates/perry-ext-net/src/pipe.rs +++ b/crates/perry-ext-net/src/pipe.rs @@ -36,7 +36,7 @@ use std::sync::{Mutex, OnceLock}; use perry_ffi::{ alloc_closure, closure_capture_f64, register_closure_arity, set_closure_capture_f64, - RawClosureHeader, + GcRootVisitor, RawClosureHeader, }; use crate::statics; @@ -135,12 +135,27 @@ fn ensure_pipe_closure_arities_registered() { /// One socket -> destination pipe route, tracked so `unpipe` can remove /// exactly the listener closures a matching `pipe()` call installed. +/// +/// Deliberately does NOT cache `dest`'s bits here: `dest` is a NaN-boxed +/// value that can be a heap pointer, and a second, un-rooted copy of it +/// would go stale the moment a GC cycle moves the object — the closure's +/// OWN capture slot 0 (scanned automatically once `data_cb` is reachable +/// via `statics::listeners()`, see `install_pipe_listeners`) is the only +/// copy this module keeps, and `matches_dest` below reads it back live at +/// comparison time instead of trusting a cache the collector cannot see. struct PipeRoute { - dest_bits: u64, data_cb: i64, end_cb: i64, } +impl PipeRoute { + fn matches_dest(&self, dest_bits: u64) -> bool { + let live_dest = + unsafe { closure_capture_f64(self.data_cb as *const RawClosureHeader, 0) }; + live_dest.to_bits() == dest_bits + } +} + fn pipe_routes() -> &'static Mutex>> { static ROUTES: OnceLock>>> = OnceLock::new(); ROUTES.get_or_init(|| Mutex::new(HashMap::new())) @@ -217,11 +232,7 @@ pub(crate) fn socket_pipe(handle: i64, dest: f64, options: f64) -> f64 { .unwrap() .entry(handle) .or_default() - .push(PipeRoute { - dest_bits: dest.to_bits(), - data_cb, - end_cb, - }); + .push(PipeRoute { data_cb, end_cb }); dest } @@ -238,7 +249,7 @@ pub(crate) fn socket_unpipe(handle: i64, dest: f64) { }; let mut removed = Vec::new(); list.retain(|route| { - let matches = filter_bits.is_none_or(|bits| bits == route.dest_bits); + let matches = filter_bits.is_none_or(|bits| route.matches_dest(bits)); if matches { removed.push((route.data_cb, route.end_cb)); } @@ -263,6 +274,28 @@ pub(crate) fn drop_routes(handle: i64) { pipe_routes().lock().unwrap().remove(&handle); } +/// GC root scanner for `pipe_routes()` — called from +/// `gc_roots::scan_net_roots` alongside the sibling `statics::listeners()` +/// scan. `data_cb`/`end_cb` are a SECOND copy of pointers already rooted via +/// `statics::listeners()` (`install_pipe_listeners` pushes the same values +/// there), but a copying GC cycle only rewrites addresses IN PLACE at +/// wherever the scanner visits them — the two copies are independent slots +/// as far as the collector is concerned, so this copy needs its own visit or +/// it keeps the pre-evacuation address after the `statics::listeners()` copy +/// has already been updated (`matches_dest`'s capture-slot read would then +/// dereference a stale/forwarded pointer — exactly the class of bug +/// `scripts/gc_runtime_root_holders.py` exists to catch). +pub(crate) fn scan_roots(visitor: &mut GcRootVisitor<'_>) { + if let Ok(mut routes) = pipe_routes().lock() { + for per_socket in routes.values_mut() { + for route in per_socket.iter_mut() { + visitor.visit_i64_slot(&mut route.data_cb); + visitor.visit_i64_slot(&mut route.end_cb); + } + } + } +} + // ─── FFI: typed `net.Socket.prototype.pipe`/`.unpipe` ──────────────────────── // // The `NativeModSig` rows in diff --git a/crates/perry-ext-net/src/socket_events.rs b/crates/perry-ext-net/src/socket_events.rs index def81b921e..f60edb82dd 100644 --- a/crates/perry-ext-net/src/socket_events.rs +++ b/crates/perry-ext-net/src/socket_events.rs @@ -250,6 +250,18 @@ pub unsafe extern "C" fn js_ext_net_drain_pending() -> i32 { fn js_tls_client_record_closed(handle: i64); } js_tls_client_record_closed(id); + // #10465 — flip the terminal state fields synchronously with + // firing `'close'`, matching Node's own timing (its + // `'close'` listeners see `destroyed: true`; earlier events + // on the SAME socket do not). This is the common teardown + // point for every path that reaches `Close`: peer EOF + + // local end, explicit `.destroy()`, connect failure, TLS + // handshake failure. + if let Some(socket) = statics::sockets().lock().unwrap().get_mut(&id) { + socket.destroyed = true; + socket.is_open = false; + socket.connecting = false; + } let had_error = f64::from_bits(JsValue::from_bool(false).bits()); let frame = dispatch_custody::DispatchFrame::park(listeners_for(id, "close")); for i in 0..frame.len() { From b9c3a1ede56cf756aca7e89f8d911356554952e0 Mon Sep 17 00:00:00 2001 From: Perry Bot Date: Fri, 18 Sep 2026 18:54:25 +0000 Subject: [PATCH 3/4] style: cargo fmt cargo fmt --- crates/perry-ext-net/src/pipe.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/crates/perry-ext-net/src/pipe.rs b/crates/perry-ext-net/src/pipe.rs index 5fcf2bde94..13518d7010 100644 --- a/crates/perry-ext-net/src/pipe.rs +++ b/crates/perry-ext-net/src/pipe.rs @@ -150,8 +150,7 @@ struct PipeRoute { impl PipeRoute { fn matches_dest(&self, dest_bits: u64) -> bool { - let live_dest = - unsafe { closure_capture_f64(self.data_cb as *const RawClosureHeader, 0) }; + let live_dest = unsafe { closure_capture_f64(self.data_cb as *const RawClosureHeader, 0) }; live_dest.to_bits() == dest_bits } } From 7a28f8ce041e4631a71c17c5ae4297861e5b1a2b Mon Sep 17 00:00:00 2001 From: Perry Bot Date: Fri, 18 Sep 2026 18:57:24 +0000 Subject: [PATCH 4/4] docs: changelog fragment for #10658 --- .../10658-net-socket-surface-cluster.md | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) create mode 100644 changelog.d/10658-net-socket-surface-cluster.md diff --git a/changelog.d/10658-net-socket-surface-cluster.md b/changelog.d/10658-net-socket-surface-cluster.md new file mode 100644 index 0000000000..a0bf5d2b11 --- /dev/null +++ b/changelog.d/10658-net-socket-surface-cluster.md @@ -0,0 +1,23 @@ +Fix a cluster of four `net.Socket` gaps the package audit hit while compiling +real socket-backed npm packages (mysql2, pg, redis, ws) natively: missing +`prependListener`/`prependOnceListener` (#10441), `on()`/`addListener()` +returning `undefined` on a typed `net.Socket` receiver instead of the socket, +breaking `.on(...).on(...)` chaining (#10442), a missing `pipe()` (#10444), +and missing/incorrect `writable`/`readable`/`readyState`/`connecting`/ +`pending`/`destroyed`/`_writableState`/`_readableState` (#10465). + +Root cause was shared shape (an incomplete dispatch table, both the untyped +dynamic-dispatch path and the typed `net.Socket` codegen table), but not a +single shared fix: #10441/#10442 were table-completion, #10444 needed a new +`pipe()` implementation (`crates/perry-ext-net/src/pipe.rs`, via the same +generic `Get("write")`+call duck-typed dispatch the runtime already uses for +thenables), and #10465 needed new lifecycle state tracking +(`SocketState::connecting`/`writable_ended`/`readable_ended`/`has_opened`). + +Validating #10465 against Node byte-for-byte surfaced two additional bugs, +fixed here: `destroyed`/`is_open` were flipped on the tokio task thread as +soon as teardown started, before the main thread had processed the `'end'` +event that same teardown queued, so a `pending`/`destroyed` read from inside +an `'end'` listener disagreed with Node; and `pipe()`'s own route-tracking +table cached a socket-destination pointer outside every GC root scanner, +which a copying GC cycle between `pipe()` and `unpipe()` could turn stale.