usockets: do not act on a peer reset while a node:net TCP socket does not read

packages/bun-usockets/src/context.c

@@ -364,6 +364,7 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls,
     s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
     s->unclassified_send_failures = 0;
     s->read_eof = 0;
+    s->defer_error_until_read = 0;
     s->hangup_closes_unsent = 0;
     s->next = 0;
     s->prev = 0;
@@ -545,6 +546,7 @@ static inline void us_internal_init_connect_socket(struct us_socket_t *s,
     s->flags.last_write_failed = 0;
     s->unclassified_send_failures = 0;
     s->read_eof = 0;
+    s->defer_error_until_read = 0;
     s->hangup_closes_unsent = 0;
     s->connect_state = NULL;
     s->connect_next = NULL;

packages/bun-usockets/src/internal/internal.h

@@ -349,6 +349,8 @@ struct us_socket_t {
   unsigned char ssl_has_pending_events : 1;
   /* Peer FIN was dispatched as on_end on a half-open socket; readable interest is never re-added and on_end never re-fires. */
   unsigned char read_eof : 1;
+  /* us_socket_defer_error_until_read. */
+  unsigned char defer_error_until_read : 1;
   /* A hangup that leaves bytes unsent closes the socket even while it is paused (loop.c defers it otherwise).
    * For an owner whose pause can wait for those bytes to drain: node:http's pipelining. */
   unsigned char hangup_closes_unsent : 1;

packages/bun-usockets/src/libusockets.h

@@ -668,6 +668,8 @@ struct us_socket_group_t *us_socket_group(us_socket_r s) nonnull_fn_decl __attri
 unsigned char us_socket_kind(us_socket_r s) nonnull_fn_decl;
 void us_socket_set_kind(us_socket_r s, unsigned char kind) nonnull_fn_decl;
 void us_socket_set_ssl_raw_tap(us_socket_r s, int enabled) nonnull_fn_decl;
+/* Like libuv (for node:net): while the socket neither reads nor has a write pending, a peer reset waits in the kernel for the resume or write that meets it. Not for a TLS socket, and not on Windows. */
+void us_socket_defer_error_until_read(us_socket_r s, int enabled) nonnull_fn_decl;
 
 void us_socket_flush(us_socket_r s) nonnull_fn_decl;
 void us_socket_shutdown(us_socket_r s) nonnull_fn_decl;

packages/bun-usockets/src/loop.c

@@ -475,6 +475,32 @@ void us_internal_loop_post(struct us_loop_t *loop) {
 #define us_ioctl ioctl
 #endif
 
+/* us_socket_defer_error_until_read: returns nonzero when the peer's reset is to stay in the kernel for now. */
+static int us_internal_socket_defers_error(struct us_socket_t *s, struct us_loop_t *loop) {
+#ifdef _WIN32
+    /* Windows discards the receive queue on a reset: nothing is there to deliver ahead of the error. */
+    (void) s;
+    (void) loop;
+    return 0;
+#else
+    /* The reads are over once the peer's FIN was delivered as on_end. */
+    const int reads = !s->flags.is_paused && !s->read_eof;
+    /* Not a TLS socket: its write does not report a rejected send() yet, so nothing would report the reset. */
+    if (!s->defer_error_until_read || s->ssl || reads || s->flags.last_write_failed) {
+        return 0;
+    }
+#ifdef LIBUS_USE_EPOLL
+    /* EPOLLERR and EPOLLHUP cannot be masked: leave epoll. A resume or a write re-adds the fd (us_poll_change). */
+    us_poll_stop(&s->p, loop);
+    s->p.state.poll_type = us_internal_poll_type(&s->p);
+#else
+    /* Only the EV_CLEAR read filter stays (kqueue_change), and it does not fire again. A resume or a write registers the others. */
+    us_poll_change(&s->p, loop, 0);
+#endif
+    return 1;
+#endif
+}
+
 void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, int events) {
     switch (us_internal_poll_type(p)) {
     case POLL_TYPE_CALLBACK: {
@@ -548,6 +574,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                         s->long_timeout = 255;
                         s->flags.low_prio_state = 0;
                         s->flags.allow_half_open = listen_socket->s.flags.allow_half_open;
+                        s->defer_error_until_read = 0;
                         s->flags.is_paused = listen_socket->accept_paused;
                         s->flags.is_ipc = 0;
                         s->flags.is_closed = 0;
@@ -603,6 +630,9 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
             /* Set once recv() returns 0 below: the only proof that the peer's stream ended with a FIN. The
              * eof hint this dispatch was called with (EPOLLHUP, kqueue's EV_EOF) also rides on a reset. */
             int read_fin = 0;
+            if (error && us_internal_socket_defers_error(s, loop)) {
+                break;
+            }
             if (events & LIBUS_SOCKET_WRITABLE && !error) {
                 s->flags.last_write_failed = 0;
                 #ifdef LIBUS_USE_KQUEUE
@@ -788,6 +818,10 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                          * buffer. This is what the comment above always described; it
                          * was keyed on the error flag, which kqueue does not set for
                          * a peer FIN. */
+                        if (s && !us_socket_is_closed(s) && error && us_internal_socket_defers_error(s, loop)) {
+                            /* on_data stopped the reads: the rest of the queue and the error wait for the resume. */
+                            return;
+                        }
                         if (s && !us_socket_is_closed(s) && (error || (!s->flags.is_paused && eof))) {
                             continue;
                         }
@@ -969,6 +1003,10 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                  * Windows does not reliably latch a received RST in SO_ERROR
                  * (see us_internal_libuv_peer_reset_probe), so it is taken
                  * there, and the CRT's ECONNRESET reached JS as ESHUTDOWN. */
+                if (us_internal_socket_defers_error(s, loop)) {
+                    /* The FIN ahead of the reset was just read: the owner reads nothing after it. */
+                    break;
+                }
                 int socket_error = us_socket_get_error(s);
                 s = us_internal_socket_close_raw(s, socket_error > 2 ? socket_error : LIBUS_ECONNRESET, NULL);
                 return;

packages/bun-usockets/src/socket.c

@@ -89,6 +89,10 @@ void us_socket_set_ssl_raw_tap(struct us_socket_t *s, int enabled) {
     s->ssl_raw_tap = !!enabled;
 }
 
+void us_socket_defer_error_until_read(struct us_socket_t *s, int enabled) {
+    s->defer_error_until_read = !!enabled;
+}
+
 struct us_socket_group_t *us_connecting_socket_group(struct us_connecting_socket_t *c) {
     return c->group;
 }
@@ -470,6 +474,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned
     s->flags.last_write_failed = 0;
     s->unclassified_send_failures = 0;
     s->read_eof = 0;
+    s->defer_error_until_read = 0;
     s->hangup_closes_unsent = 0;
     s->connect_state = NULL;
 

src/js/node/net.ts

@@ -175,6 +175,7 @@ interface TLSConnectOptions {
 interface InternalSocketHandler<Data> extends SocketHandler<Data> {
   session?(socket: Socket<Data>, session: Buffer): void;
   keylog?(socket: Socket<Data>, line: Buffer): void;
+  deferErrorUntilRead?: boolean;
 }
 
 interface InternalServerHandler<Data> extends InternalSocketHandler<Data> {
@@ -781,6 +782,7 @@ const SocketHandlers = {
     self.emit("timeout", self);
   },
   binaryType: "buffer",
+  deferErrorUntilRead: true,
 } as const satisfies InternalSocketHandler<SocketInstance>;
 
 // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L191-L198; a stopped handle does not hold the loop, a pending write still does.
@@ -1303,6 +1305,7 @@ const ServerHandlers = {
     SocketHandlers.drain(socket);
   },
   binaryType: "buffer",
+  deferErrorUntilRead: true,
 } as const satisfies InternalServerHandler<SocketInstance>;
 
 function applyRejectUnauthorized(self, tls, rejectUnauthorized) {
@@ -1630,6 +1633,7 @@ const SocketHandlers2 = {
     }
     req.oncomplete(error.errno, self._handle, req, true, true);
   },
+  deferErrorUntilRead: true,
 } satisfies InternalSocketHandler<ConnectData>;
 
 // The same table minus the per-connection callback members: a listener whose
@@ -1964,6 +1968,7 @@ function Socket(options?): void {
         const { self } = socket.data;
         if (!self) return;
         self._unrefTimer();
+        self.bytesRead += buffer.length;
         if (socket[kAdoptedTLSRaw]) return;
         const tail = self[kOnreadTail];
         if (tail !== undefined) {

src/jsc/generated.rs

@@ -208,6 +208,7 @@ pub struct SocketConfigHandlers {
     pub on_server_name: JSValue,
     pub on_alpn_callback: JSValue,
     pub binary_type: SocketConfigHandlersBinaryType,
+    pub defer_error_until_read: bool,
 }
 
 #[repr(C)]
@@ -227,6 +228,7 @@ struct ExternSocketConfigHandlers {
     onServerName: JSValue,
     onALPNCallback: JSValue,
     binary_type: SocketConfigHandlersBinaryType,
+    defer_error_until_read: bool,
 }
 
 // safe: `JSGlobalObject` is an opaque `UnsafeCell`-backed ZST handle (`&` is
@@ -258,6 +260,7 @@ impl SocketConfigHandlers {
             on_server_name: ext.onServerName,
             on_alpn_callback: ext.onALPNCallback,
             binary_type: ext.binary_type,
+            defer_error_until_read: ext.defer_error_until_read,
         }
     }
 

src/runtime/socket/Handlers.rs

@@ -46,6 +46,9 @@ pub(crate) struct Handlers {
 
     pub(crate) binary_type: Cell<BinaryType>,
 
+    /// node:net: see `us_socket_defer_error_until_read`. Applied to a socket in `on_open` and in its `reload`.
+    pub(crate) defer_error_until_read: Cell<bool>,
+
     pub(crate) vm: &'static VirtualMachine,
     pub(crate) global_object: GlobalRef,
     /// The context of the script that gave these handlers: a socket event is dispatched inside it.
@@ -73,6 +76,7 @@ pub(crate) struct Handlers {
 pub(crate) struct ReloadedHandlers {
     callbacks: [JSValue; CALLBACK_COUNT],
     pub(crate) binary_type: BinaryType,
+    defer_error_until_read: bool,
 }
 
 fn binary_type_from_generated(binary_type: GeneratedBinaryType) -> BinaryType {
@@ -327,6 +331,7 @@ impl Handlers {
         Ok(Rc::new(Handlers {
             cell: JSSocketHandlers::create(global_object, &wrapped),
             binary_type: Cell::new(binary_type_from_generated(generated.binary_type)),
+            defer_error_until_read: Cell::new(generated.defer_error_until_read),
             // SAFETY: `bun_vm()` never returns null for a Bun-owned global; the
             // VM outlives every `Handlers` (process-lifetime singleton).
             vm: global_object.bun_vm(),
@@ -411,6 +416,7 @@ impl Handlers {
         Ok(ReloadedHandlers {
             callbacks,
             binary_type: binary_type_from_generated(generated.binary_type),
+            defer_error_until_read: generated.defer_error_until_read,
         })
     }
 
@@ -420,6 +426,8 @@ impl Handlers {
         let wrapped = Self::wrap_with_context(global_object, &reloaded.callbacks);
         self.cell.set_callbacks(global_object, &wrapped);
         self.binary_type.set(reloaded.binary_type);
+        self.defer_error_until_read
+            .set(reloaded.defer_error_until_read);
     }
 }
 

src/runtime/socket/SocketConfig.bindv2.ts

@@ -32,6 +32,12 @@ export const Handlers = b.dictionary(
       default: "buffer",
       internalName: "binary_type",
     },
+    // node:net: see us_socket_defer_error_until_read.
+    deferErrorUntilRead: {
+      type: b.bool,
+      default: false,
+      internalName: "defer_error_until_read",
+    },
   },
 );
 

src/runtime/socket/socket_body.rs

@@ -1457,6 +1457,9 @@ impl<const SSL: bool> NewSocket<SSL> {
         // update the internal socket instance to the one that was just connected
         // This socket must be replaced because the previous one is a connecting socket not a uSockets socket
         this.socket.set(socket);
+        if this.get_handlers().defer_error_until_read.get() {
+            socket.defer_error_until_read(true);
+        }
         // Stale if node:net reconnected through this wrapper while it was paused.
         this.update_flags(|f| f.remove(Flags::IS_PAUSED));
         if !this.ref_pollref_on_connect.replace(true) {
@@ -3398,6 +3401,9 @@ impl<const SSL: bool> NewSocket<SSL> {
         // and every socket sharing it observe them; nothing else about the
         // shared `Handlers` (mode, active_connections) is touched.
         handlers.apply_reload(global, &reloaded);
+        this.socket
+            .get()
+            .defer_error_until_read(handlers.defer_error_until_read.get());
 
         Ok(JSValue::UNDEFINED)
     }

src/uws_sys/socket.rs

@@ -507,6 +507,13 @@ impl<const IS_SSL: bool> NewSocketHandler<IS_SSL> {
         )
     }
 
+    /// See `us_socket_defer_error_until_read` in libusockets.h.
+    pub fn defer_error_until_read(&self, enabled: bool) {
+        if let InternalSocket::Connected(s) = self.socket {
+            sock(s).defer_error_until_read(enabled);
+        }
+    }
+
     pub fn set_no_delay(&self, enabled: bool) -> bool {
         match self.socket {
             InternalSocket::Connected(s) => {

src/uws_sys/us_socket_t.rs

@@ -339,6 +339,11 @@ impl us_socket_t {
         c::us_socket_set_ssl_raw_tap(self, enabled as c_int);
     }
 
+    /// See `us_socket_defer_error_until_read` in libusockets.h.
+    pub fn defer_error_until_read(&mut self, enabled: bool) {
+        c::us_socket_defer_error_until_read(self, enabled as c_int);
+    }
+
     pub fn write(&mut self, data: &[u8]) -> i32 {
         let rc = unsafe {
             // SAFETY: data.as_ptr() valid for data.len() bytes
@@ -521,6 +526,7 @@ mod c {
         pub(super) safe fn us_socket_kind(s: &us_socket_t) -> u8;
         pub(super) safe fn us_socket_set_kind(s: &mut us_socket_t, kind: u8);
         pub(super) safe fn us_socket_set_ssl_raw_tap(s: &mut us_socket_t, enabled: c_int);
+        pub(super) safe fn us_socket_defer_error_until_read(s: &mut us_socket_t, enabled: c_int);
 
         pub(super) fn us_socket_write(s: *mut us_socket_t, data: *const u8, length: i32) -> i32;
         #[cfg(not(windows))]

test/js/bun/net/socket.test.ts

@@ -4677,8 +4677,9 @@ it("a paused socket does not wake the event loop for every segment its peer send
 // epoll_kqueue.c). The reset is the end of the connection, so the pause no longer protects
 // anything: the data queued ahead of the reset is delivered, then the socket closes with read
 // ECONNRESET. Closing without reading discarded that data (a streamed body cut short although
-// every byte arrived, #39846). Windows discards the receive queue on a reset itself. The
-// node:net and node:tls shapes are in test/js/node/tls/node-tls-server.test.ts.
+// every byte arrived, #39846). Windows discards the receive queue on a reset itself. A net.Socket
+// opts out and meets the reset when it resumes, like node. The node:net and node:tls shapes are in
+// test/js/node/tls/node-tls-server.test.ts.
 describe.concurrent.each(["tcp", "tls"] as const)("%s socket paused when its peer resets the connection", transport => {
   it("delivers the data queued ahead of the reset, then closes with read ECONNRESET, while still paused", async () => {
     const closedWith = Promise.withResolvers<Error | undefined>();