fix(ext-net): net.Socket surface cluster — prependListener, on() chaining, pipe(), stream state

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.

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",

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

@@ -62,6 +62,10 @@ pub fn adopt_upgraded_tcp_stream(stream: tokio::net::TcpStream) -> i64 {
             remote_addr: remote,
             raw: None,
             destroyed: false,
+            connecting: false,
+            has_opened: true,
+            writable_ended: false,
+            readable_ended: false,
             bytes_read: 0,
             bytes_written: 0,
             bytes_queued: 0,

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

@@ -72,15 +72,15 @@ pub(crate) fn ensure_runtime_dispatch_registered() {
     });
 }
 
-fn undefined() -> f64 {
+pub(crate) fn undefined() -> f64 {
     f64::from_bits(TAG_UNDEFINED)
 }
 
 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<f64>
             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,

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);
 }

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]

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

@@ -44,6 +44,10 @@ fn allocate_socket() -> (i64, mpsc::UnboundedReceiver<SocketCommand>) {
             remote_addr: None,
             raw: None,
             destroyed: false,
+            connecting: true,
+            has_opened: false,
+            writable_ended: false,
+            readable_ended: false,
             bytes_read: 0,
             bytes_written: 0,
             bytes_queued: 0,
@@ -116,6 +120,10 @@ pub(crate) fn register_accepted_transport(
             remote_addr,
             raw: None,
             destroyed: false,
+            connecting: false,
+            has_opened: true,
+            writable_ended: false,
+            readable_ended: false,
             bytes_read: 0,
             bytes_written: 0,
             bytes_queued: 0,
@@ -150,9 +158,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 +200,8 @@ fn spawn_connect(id: i64, path: String, mut rx: mpsc::UnboundedReceiver<SocketCo
             let raw_fd = transport.raw_fd();
             if let Some(socket) = statics::sockets().lock().unwrap().get_mut(&id) {
                 socket.is_open = true;
+                socket.has_opened = true;
+                socket.connecting = false;
                 socket.raw_fd = raw_fd;
             }
             tokio::task::yield_now().await;