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;
@@ -538,7 +539,8 @@ static inline void us_internal_init_connect_socket(struct us_socket_t *s,
     s->long_timeout = 255;
     s->flags.low_prio_state = 0;
     s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN);
-    s->flags.is_paused = 0;
+    /* Honored by us_internal_socket_after_open; until then the poll only waits for the connect. */
+    s->flags.is_paused = (options & LIBUS_SOCKET_OPEN_PAUSED) != 0;
     s->flags.is_ipc = 0;
     s->flags.is_closed = 0;
     s->flags.adopted = 0;
@@ -825,7 +827,10 @@ void us_internal_socket_after_open(struct us_socket_t *s, int error) {
             // It's expected that close is called by the caller
         }
     } else {
-        us_poll_change(&s->p, s->group->loop, LIBUS_SOCKET_READABLE);
+        if (s->ssl || (c && c->ssl_ctx)) {
+            s->flags.is_paused = 0;
+        }
+        us_poll_change(&s->p, s->group->loop, s->flags.is_paused ? 0 : LIBUS_SOCKET_READABLE);
         bsd_socket_nodelay(us_poll_fd(&s->p), 1);
         us_internal_poll_set_type(&s->p, POLL_TYPE_SOCKET);
         us_socket_timeout(s, 0);

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,11 @@ enum {
      * unconnected socket it also makes the next send fail for a datagram
      * bound to a different, live peer. */
     LIBUS_UDP_LINUX_RECVERR = 128,
+    /* The socket opens with reads paused (no read interest until us_socket_resume), whether it
+     * is connected, adopted by us_socket_from_fd, or accepted by a listener that was created
+     * with the option. Ignored for TLS sockets: they have to read to complete the handshake, the
+     * owner pauses them once it is done. node:net's pauseOnConnect. */
+    LIBUS_SOCKET_OPEN_PAUSED = 256,
 };
 
 /* Library types publicly available */

packages/bun-usockets/src/loop.c

@@ -513,7 +513,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. */
@@ -532,7 +532,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

@@ -448,7 +448,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;
@@ -462,7 +463,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);
+  Duplex.prototype.pause.$call(self);
+}
+
+// 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) {
@@ -980,7 +990,7 @@ const ServerHandlers: SocketHandler<NetSocket> = {
     }
     const pauseOnConnect = server && (server.pauseOnConnect ?? server[bunSocketServerOptions]?.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
@@ -1224,7 +1234,7 @@ function onconnection(err, clientHandle) {
   _socket._server = self;
 
   if (pauseOnConnect && !isTLS) {
-    _socket.pause();
+    pauseOnCreate(_socket, clientHandle);
   }
 
   if (typeof connectionListener === "function") {
@@ -1263,9 +1273,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 +1283,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 +1508,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 +1522,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 +1620,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 +1727,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 +1758,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 +1922,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 opens paused through its req, and afterConnect releases its hold.
+      if (fd != null) pauseOnCreate(this, this._handle);
+      else Duplex.prototype.pause.$call(this);
     } else {
       process.nextTick(() => {
         // Honor pause()/resume() calls made while connecting — only start
@@ -2345,19 +2355,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 +3134,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 +3155,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 +3280,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 +3387,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 +3399,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 +3903,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 +3936,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 +3949,7 @@ Server.prototype[kRealListen] = function (
       exclusive: exclusive || this[bunSocketServerOptions]?.exclusive || false,
       socket: serverHandlersFor(this),
       data: this,
+      pauseOnConnect: this.pauseOnConnect,
     });
   }
 

src/runtime/socket/Handlers.rs

@@ -452,6 +452,8 @@ pub struct SocketConfig {
     pub(crate) allow_half_open: bool,
     pub(crate) reuse_port: bool,
     pub(crate) ipv6_only: bool,
+    /// node:net's `pauseOnConnect` (read off the options like `localAddress`): the socket, or every accepted one, opens paused.
+    pub(crate) pause_on_connect: bool,
 }
 
 impl SocketConfig {
@@ -472,6 +474,9 @@ impl SocketConfig {
         if self.ipv6_only {
             flags |= uws::LIBUS_SOCKET_IPV6_ONLY;
         }
+        if self.pause_on_connect {
+            flags |= uws::LIBUS_SOCKET_OPEN_PAUSED;
+        }
 
         flags
     }
@@ -518,6 +523,7 @@ impl SocketConfig {
                 allow_half_open: false,
                 reuse_port: false,
                 ipv6_only: false,
+                pause_on_connect: false,
             };
         };
         // On any `?` below, `result` drops and releases what it owns — no
@@ -582,7 +588,11 @@ impl SocketConfig {
         mode: SocketMode,
     ) -> JsResult<SocketConfig> {
         let generated = GeneratedSocketConfig::from_js(global_object, opts)?;
-        Self::from_generated(vm, global_object, &generated, mode)
+        let mut config = Self::from_generated(vm, global_object, &generated, mode)?;
+        config.pause_on_connect = opts
+            .get_truthy(global_object, "pauseOnConnect")?
+            .is_some_and(|v| v.to_boolean());
+        Ok(config)
     }
 }