net: keep the handle reading while a socket is paused, like node

packages/bun-usockets/src/context.c

@@ -379,6 +379,7 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls,
     ls->on_server_name = NULL;
     ls->socket_ext_size = socket_ext_size;
     ls->deferred_accept = 0;
+    ls->accept_paused = (options & LIBUS_SOCKET_OPEN_PAUSED) && !ssl_ctx;
 
     /* Link into the group so close_all() / test-isolation can find it. */
     ls->next = group->head_listen_sockets;

packages/bun-usockets/src/eventing/epoll_kqueue.c

@@ -690,7 +690,11 @@ int us_poll_start_rc(struct us_poll_t *p, struct us_loop_t *loop, int events) {
     } while (IS_EINTR(ret));
     return ret;
 #else
-    return kqueue_change(loop->fd, p->state.fd, 0, events, p, 0);
+    /* A socket that starts without read interest (LIBUS_SOCKET_OPEN_PAUSED) takes the same
+     * transition as one that pauses, so it too keeps the read knote that reports the peer's
+     * FIN or RST while it is paused (see kqueue_change). */
+    int starts_paused = kqueue_is_socket_poll(p) && !(events & LIBUS_SOCKET_READABLE);
+    return kqueue_change(loop->fd, p->state.fd, starts_paused ? LIBUS_SOCKET_READABLE : 0, events, p, starts_paused);
 #endif
 }
 

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

@@ -475,6 +475,8 @@ struct us_listen_socket_t {
   unsigned char accept_kind;
   /* Set when TCP_DEFER_ACCEPT/SO_ACCEPTFILTER was successfully applied. */
   unsigned char deferred_accept;
+  /* LIBUS_SOCKET_OPEN_PAUSED: accepted sockets start without read interest. */
+  unsigned char accept_paused;
 };
 
 void us_internal_socket_group_link_connecting_socket(us_socket_group_r group, struct us_connecting_socket_t *c);

packages/bun-usockets/src/libusockets.h

@@ -160,6 +160,12 @@ enum {
      * unconnected socket it also makes the next send fail for a datagram
      * bound to a different, live peer. */
     LIBUS_UDP_LINUX_RECVERR = 128,
+    /* A socket adopted by us_socket_from_fd, or accepted by a listener created with this option,
+     * is registered as if us_socket_pause had been called on it, so no extra poll change is needed
+     * per connection (node:net's pauseOnConnect; cluster adopts every connection this way). Not
+     * for connects, and ignored for TLS sockets: the handshake needs the reads, the owner pauses
+     * those sockets itself. */
+    LIBUS_SOCKET_OPEN_PAUSED = 256,
 };
 
 /* Library types publicly available */

packages/bun-usockets/src/loop.c

@@ -509,7 +509,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                     do {
                         struct us_poll_t *accepted_p = us_create_poll(loop, 0, sizeof(struct us_socket_t) - sizeof(struct us_poll_t) + listen_socket->socket_ext_size);
                         us_poll_init(accepted_p, client_fd, POLL_TYPE_SOCKET);
-                        if (us_poll_start_rc(accepted_p, loop, LIBUS_SOCKET_READABLE) != 0) {
+                        if (us_poll_start_rc(accepted_p, loop, listen_socket->accept_paused ? 0 : LIBUS_SOCKET_READABLE) != 0) {
                             /* EPOLL_CTL_ADD failed (e.g. ENOSPC). Close the fd so the
                              * peer sees a RST instead of a connection that silently
                              * never answers. */
@@ -528,7 +528,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->flags.is_paused = 0;
+                        s->flags.is_paused = listen_socket->accept_paused;
                         s->flags.is_ipc = 0;
                         s->flags.is_closed = 0;
                         s->flags.adopted = 0;

packages/bun-usockets/src/socket.c

@@ -444,7 +444,8 @@ int us_socket_write2(struct us_socket_t *s, const char *header, int header_lengt
 struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned char kind, struct ssl_ctx_st *ssl_ctx, int socket_ext_size, LIBUS_SOCKET_DESCRIPTOR fd, int options, int ipc) {
     struct us_poll_t *p1 = us_create_poll(group->loop, 0, sizeof(struct us_socket_t) + socket_ext_size);
     us_poll_init(p1, fd, POLL_TYPE_SOCKET);
-    int rc = us_poll_start_rc(p1, group->loop, LIBUS_SOCKET_READABLE | LIBUS_SOCKET_WRITABLE);
+    int open_paused = (options & LIBUS_SOCKET_OPEN_PAUSED) && !ssl_ctx;
+    int rc = us_poll_start_rc(p1, group->loop, (open_paused ? 0 : LIBUS_SOCKET_READABLE) | LIBUS_SOCKET_WRITABLE);
     if (rc != 0) {
         us_poll_free(p1, group->loop);
         return 0;
@@ -458,7 +459,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned
     s->long_timeout = 255;
     s->flags.low_prio_state = 0;
     s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN) != 0;
-    s->flags.is_paused = 0;
+    s->flags.is_paused = open_paused;
     s->flags.is_ipc = ipc;
     s->flags.is_closed = 0;
     s->flags.adopted = 0;

src/js/node/net.ts

@@ -156,7 +156,7 @@ const kPerfHooksNetConnectContext = Symbol("kPerfHooksNetConnectContext");
 const khandshakeTimer = Symbol("khandshakeTimer");
 const kerrorEmitted = Symbol("kerrorEmitted");
 const kUserUnrefed = Symbol("kUserUnrefed");
-// Set when pause() dropped the handle's hold on the loop, so the read paths
+// Set when readStop() dropped the handle's hold on the loop, so the read paths
 // only restore a hold they actually removed - re-refing a handle that never
 // held the loop (a wrapped duplex with no fd) would pin the process.
 const kPausedUnref = Symbol("kPausedUnref");
@@ -413,7 +413,7 @@ const SocketHandlers: SocketHandler = {
     self._unrefTimer();
     self.bytesRead += buffer.length;
     if (!self.push(buffer)) {
-      pauseForBackpressure(self, socket);
+      readStop(self, socket);
     }
   },
   drain(socket) {
@@ -562,17 +562,27 @@ const SocketHandlers: SocketHandler = {
   binaryType: "buffer",
 } as const;
 
-// push() (or an onread callback) said stop. Like node's readStop, a socket that
-// is not reading does not hold the loop (see Socket.prototype.pause); a write
-// awaiting drain still does, and unrefAfterDrain drops the hold once it completes.
-function pauseForBackpressure(self, handle) {
+// Node's readStop: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L191-L198
+function readStop(self, handle) {
   handle?.pause?.();
+  releaseReadHold(self, handle);
+}
+
+// A handle that is not reading does not hold the loop; a pending write still does (see unrefAfterDrain).
+function releaseReadHold(self, handle) {
+  // A socket over a generic duplex has no fd and never held the loop.
   if (self[kupgraded] && !(self[kupgraded] instanceof Socket)) return;
   self[kPausedUnref] = true;
   if (!self[kwriteCallback]) handle?.unref?.();
 }
 
-// Reads are flowing again: give back the hold a pause dropped. Cleared even when
+// The handle opened paused (pauseOnConnect in its native config): https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L494-L498
+function pauseOnCreate(self, handle) {
+  releaseReadHold(self, handle);
+  self.readableFlowing = false;
+}
+
+// Reads are flowing again: give back the hold readStop dropped. Cleared even when
 // the user unref'd, so a later ref() is not undone by unrefAfterDrain.
 function restorePausedHold(self, handle) {
   if (!self[kPausedUnref]) return;
@@ -596,7 +606,7 @@ function finishSocketEnd(self) {
   // the loop while reading or with a write in flight, so node lets the process exit even if
   // the (half-open) writable side stays open and the readable side was never consumed. Mirror
   // that: drop this handle's hold on the loop unless a write is still waiting on drain, and
-  // forget any pause()-time unref so a later read()/resume() does not pin the loop again.
+  // forget any readStop-time unref so a later read()/resume() does not pin the loop again.
   // A subsequent buffered write re-refs (see _write) so its callback can still fire.
   const socket = self._handle;
   if (socket && !self[kwriteCallback]) {
@@ -755,7 +765,7 @@ const ServerHandlers: SocketHandler<NetSocket> = {
     self._unrefTimer();
     self.bytesRead += buffer.length;
     if (!self.push(buffer)) {
-      pauseForBackpressure(self, socket);
+      readStop(self, socket);
     }
   },
   keylog(socket, line) {
@@ -978,9 +988,9 @@ const ServerHandlers: SocketHandler<NetSocket> = {
         self.authorized = true;
       }
     }
-    const pauseOnConnect = server && (server.pauseOnConnect ?? server[bunSocketServerOptions]?.pauseOnConnect);
+    const pauseOnConnect = server?.pauseOnConnect;
     if (pauseOnConnect) {
-      self.pause();
+      pauseOnCreate(self, socket);
     }
     if (!pauseOnConnect && !self.destroyed) {
       // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/tls/wrap.js#L502-L524
@@ -1154,7 +1164,9 @@ function onconnection(err, clientHandle) {
   }
   clientHandle[kServerSocket] = handle;
   const options = self[bunSocketServerOptions];
-  const { pauseOnConnect, connectionListener, [kSocketClass]: SClass } = options;
+  const { connectionListener, [kSocketClass]: SClass } = options;
+  // Read per connection like node; the listener itself was created with the value listen() saw.
+  const pauseOnConnect = self.pauseOnConnect;
   // Propagate the server's half-open/highWaterMark settings to the accepted
   // socket so the Duplex's allowHalfOpen matches what the native layer was
   // configured with in kRealListen; without this, net.createServer({
@@ -1224,14 +1236,11 @@ function onconnection(err, clientHandle) {
   _socket._server = self;
 
   if (pauseOnConnect && !isTLS) {
-    _socket.pause();
+    pauseOnCreate(_socket, clientHandle);
   }
 
-  if (typeof connectionListener === "function") {
-    clientHandle.pauseOnConnect = pauseOnConnect;
-    if (!isTLS) {
-      self.prependOnceListener("connection", connectionListener);
-    }
+  if (typeof connectionListener === "function" && !isTLS) {
+    self.prependOnceListener("connection", connectionListener);
   }
   if (isTLS) initAcceptedTLSSocket(self, _socket);
 
@@ -1263,9 +1272,6 @@ const SocketHandlers2: SocketHandler<NonNullable<import("node:net").Socket["_han
     }
     if (!self[kupgraded]) req!.oncomplete(0, self._handle, req, true, true);
     socket.data.req = undefined;
-    if (self.pauseOnConnect) {
-      self.pause();
-    }
     if (self[kupgraded]) {
       self.connecting = false;
       SocketHandlers2.drain!(socket);
@@ -1276,7 +1282,7 @@ const SocketHandlers2: SocketHandler<NonNullable<import("node:net").Socket["_han
     const { self } = socket.data;
     self._unrefTimer();
     self.bytesRead += buffer.length;
-    if (!self.push(buffer)) pauseForBackpressure(self, socket);
+    if (!self.push(buffer)) readStop(self, socket);
   },
   drain(socket) {
     $debug("Bun.Socket drain");
@@ -1501,6 +1507,7 @@ function kConnectTcp(self, addressType, req, address, port) {
     // where libuv sockets are half-open and the stream layer decides.
     allowHalfOpen: true,
     tls: req.tls,
+    pauseOnConnect: req.pauseOnConnect,
     data: { self, req },
     socket: self[khandlers],
   });
@@ -1514,6 +1521,7 @@ function kConnectPipe(self, req, address) {
     // Always half-open natively; see kConnect.
     allowHalfOpen: true,
     tls: req.tls,
+    pauseOnConnect: req.pauseOnConnect,
     data: { self, req },
     socket: self[khandlers],
   });
@@ -1611,7 +1619,6 @@ function Socket(options?) {
   this[ksocket] = undefined;
   this.server = undefined;
   this._server = undefined;
-  this.pauseOnConnect = false;
   this._peername = null;
   this._sockname = null;
   this._closeAfterHandlingError = false;
@@ -1719,7 +1726,7 @@ function Socket(options?) {
           if (self.destroyed) return;
           if (ret === false || self.isPaused()) {
             self[kOnreadTail] = kOnreadEmptyTail;
-            pauseForBackpressure(self, self._handle);
+            readStop(self, self._handle);
           }
           return;
         }
@@ -1750,7 +1757,7 @@ function Socket(options?) {
         if (ret === false || self.isPaused()) {
           const rest = buffer.subarray(offset);
           self[kOnreadTail] = rest.length !== 0 ? rest : kOnreadEmptyTail;
-          pauseForBackpressure(self, self._handle);
+          readStop(self, self._handle);
           return;
         }
       }
@@ -1914,16 +1921,18 @@ Socket.prototype.connect = function connect(...args) {
         socket: SocketHandlers,
         // Always half-open natively; see kConnect.
         allowHalfOpen: true,
+        pauseOnConnect,
       }).catch(error => {
         if (!this.destroyed) {
           this.emit("error", error);
           this.emit("close", true);
         }
       });
     }
-    this.pauseOnConnect = pauseOnConnect;
     if (pauseOnConnect) {
-      this.pause();
+      // An fd is open already; a dial is paused when it opens, and afterConnect releases its hold.
+      if (fd != null) pauseOnCreate(this, this._handle);
+      else this.readableFlowing = false;
     } else {
       process.nextTick(() => {
         // Honor pause()/resume() calls made while connecting — only start
@@ -2345,19 +2354,11 @@ Socket.prototype.resume = function resume() {
   return ret;
 };
 
+// Only a connected onread socket stops reading here: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L817-L827
 Socket.prototype.pause = function pause() {
-  if (!this.destroyed) {
-    this._handle?.pause?.();
-    // libuv only counts a stream handle as active - and therefore as keeping
-    // the event loop alive - while it is reading. A paused socket lets the
-    // process exit; resume() re-refs it unless the user explicitly unref'd.
-    this._handle?.unref?.();
-    // Only remember the unref when this handle can actually hold the loop: a
-    // TLS socket wrapped over a generic duplex has no fd, so re-refing it
-    // later would newly pin the process.
-    if (!this[kupgraded] || this[kupgraded] instanceof Socket) {
-      this[kPausedUnref] = true;
-    }
+  const handle = this._handle;
+  if (handle && this[kOnreadBuffer] !== undefined && !this.connecting && !this.destroyed) {
+    readStop(this, handle);
   }
   return Duplex.prototype.pause.$call(this);
 };
@@ -3132,6 +3133,7 @@ function internalConnect(self, options, address, port, addressType, localAddress
     req.localPort = localPort;
     req.addressType = addressType;
     req.tls = tls;
+    req.pauseOnConnect = options.pauseOnConnect;
 
     traceConnectStart(req);
     err = kConnectTcp(self, addressType, req, address, port);
@@ -3152,6 +3154,7 @@ function internalConnect(self, options, address, port, addressType, localAddress
     req.address = address;
     req.oncomplete = afterConnect;
     req.tls = tls;
+    req.pauseOnConnect = options.pauseOnConnect;
 
     traceConnectStart(req, address);
     err = kConnectPipe(self, req, address);
@@ -3276,6 +3279,7 @@ function internalConnectMultiple(context, canceled?) {
   req.localPort = localPort;
   req.addressType = addressType;
   req.tls = tls;
+  req.pauseOnConnect = context.options.pauseOnConnect;
 
   ArrayPrototypePush.$call(self.autoSelectFamilyAttemptedAddresses, `${address}:${port}`);
 
@@ -3382,6 +3386,9 @@ function afterConnect(status, handle, req, readable, writable) {
       self._handle.setKeepAlive(true, self[kSetKeepAliveInitialDelay]);
     }
 
+    // Node only starts reading at the read(0) below, which a paused stream skips. TLS needs the reads to handshake.
+    if (self.isPaused() && !self.encrypted) readStop(self, self._handle);
+
     self.emit("connect");
     self.emit("ready");
 
@@ -3391,6 +3398,7 @@ function afterConnect(status, handle, req, readable, writable) {
 
     // Start the first read, or get an immediate EOF.
     // this doesn't actually consume any bytes, because len=0.
+    // https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L1695-L1696
     if (readable && !self.isPaused()) self.read(0);
   } else {
     let details;
@@ -3894,6 +3902,7 @@ Server.prototype[kRealListen] = function (
       exclusive: exclusive || this[bunSocketServerOptions]?.exclusive || false,
       socket: serverHandlersFor(this),
       data: this,
+      pauseOnConnect: this.pauseOnConnect,
     });
     // Mirror libuv uv_pipe_chmod: readableAll/writableAll relax the unix socket
     // file's group/other permission bits. Skipped on Windows and abstract
@@ -3926,6 +3935,7 @@ Server.prototype[kRealListen] = function (
       exclusive: exclusive || this[bunSocketServerOptions]?.exclusive || false,
       socket: serverHandlersFor(this),
       data: this,
+      pauseOnConnect: this.pauseOnConnect,
     });
   } else {
     this._handle = Bun.listen({
@@ -3938,6 +3948,7 @@ Server.prototype[kRealListen] = function (
       exclusive: exclusive || this[bunSocketServerOptions]?.exclusive || false,
       socket: serverHandlersFor(this),
       data: this,
+      pauseOnConnect: this.pauseOnConnect,
     });
   }