node:net: keep delivering when an onread callback calls pause() then read()

src/js/node/net.ts

@@ -191,7 +191,8 @@ const kOnreadTail = Symbol("kOnreadTail");
 const kOnreadDraining = Symbol("kOnreadDraining");
 const kOnreadBuffer = Symbol("kOnreadBuffer");
 const kOnreadPendingEnd = Symbol("kOnreadPendingEnd");
-const kOnreadReadRequested = Symbol("kOnreadReadRequested");
+// Node's handle.reading: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L817-L845
+const kOnreadReading = Symbol("kOnreadReading");
 const kOnreadEmptyTail = Buffer.alloc(0);
 const kwriteCallback = Symbol("writeCallback");
 const kSocketClass = Symbol("kSocketClass");
@@ -607,13 +608,21 @@ const SocketHandlers: SocketHandler = {
 
 // 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.
 function readStop(self, handle) {
+  self[kOnreadReading] = false;
   handle?.pause?.();
   // 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?.();
 }
 
+// https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L830-L845
+function readStart(self, handle) {
+  self[kOnreadReading] = true;
+  handle?.resume?.();
+  restorePausedHold(self, handle);
+}
+
 // https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L494-L498
 function pauseOnCreate(self, handle) {
   readStop(self, handle);
@@ -1308,6 +1317,8 @@ const SocketHandlers2: SocketHandler<NonNullable<import("node:net").Socket["_han
     socket.data.req = undefined;
     if (self[kupgraded]) {
       self.connecting = false;
+      // The constructor's read(0) is skipped while the wrapped socket connects.
+      if (!self.isPaused()) self.read(0);
       SocketHandlers2.drain!(socket);
     }
   },
@@ -1734,7 +1745,7 @@ function Socket(options?) {
             reportError(e);
           }
           if (self.destroyed) return;
-          if (ret === false || self.isPaused()) {
+          if (ret === false || !self[kOnreadReading]) {
             self[kOnreadTail] = kOnreadEmptyTail;
             readStop(self, self._handle);
           }
@@ -1764,7 +1775,7 @@ function Socket(options?) {
           reportError(e);
         }
         if (self.destroyed) return;
-        if (ret === false || self.isPaused()) {
+        if (ret === false || !self[kOnreadReading]) {
           const rest = buffer.subarray(offset);
           self[kOnreadTail] = rest.length !== 0 ? rest : kOnreadEmptyTail;
           readStop(self, self._handle);
@@ -2328,9 +2339,10 @@ function hasUnflushedWrites(connection) {
   return connection.writableLength > 0 || connection[kwriteCallback] != null;
 }
 
-function drainOnreadTail(self, fromRead?) {
+// The tail is delivered on the next tick. A pause() before then clears kOnreadReading and the tick delivers nothing.
+function drainOnreadTail(self) {
   if (self[kOnreadTail] === undefined) return false;
-  if (fromRead) self[kOnreadReadRequested] = true;
+  self[kOnreadReading] = true;
   if (!self[kOnreadDraining]) {
     self[kOnreadDraining] = true;
     process.nextTick(drainOnreadTailNT, self);
@@ -2340,20 +2352,17 @@ function drainOnreadTail(self, fromRead?) {
 
 function drainOnreadTailNT(socket) {
   socket[kOnreadDraining] = false;
-  const fromRead = socket[kOnreadReadRequested];
-  socket[kOnreadReadRequested] = false;
   const tail = socket[kOnreadTail];
   if (tail === undefined || socket.destroyed) return;
-  if (!fromRead && socket.isPaused()) return;
+  if (!socket[kOnreadReading]) return;
   socket[kOnreadTail] = undefined;
   socket[kOnreadDeliver](tail);
   if (socket[kOnreadTail] !== undefined || socket.destroyed) return;
   if (socket[kOnreadPendingEnd]) {
     socket[kOnreadPendingEnd] = false;
     finishSocketEnd(socket);
-  } else if (fromRead || !socket.isPaused()) {
-    socket._handle?.resume?.();
-    restorePausedHold(socket, socket._handle);
+  } else if (socket[kOnreadReading]) {
+    readStart(socket, socket._handle);
   }
 }
 
@@ -2366,7 +2375,8 @@ Socket.prototype.resume = function resume() {
   // An onread socket is the exception: https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L830-L845
   if (this.readableEnded && this[kOnreadBuffer] === undefined) return ret;
   if (!this.connecting && !drainOnreadTail(this)) {
-    this._handle?.resume?.();
+    readStart(this, this._handle);
+    return ret;
   }
   // Even while still connecting, so pause-then-resume stays symmetric.
   restorePausedHold(this, this._handle);
@@ -2468,9 +2478,8 @@ Socket.prototype[Symbol.for("::bunUpgradeServerTLS::")] = function (connection,
 
 Socket.prototype.read = function read(size) {
   // See resume(): an ended readable side never restarts the handle.
-  if ((!this.readableEnded || this[kOnreadBuffer] !== undefined) && !this.connecting && !drainOnreadTail(this, true)) {
-    this._handle?.resume?.();
-    restorePausedHold(this, this._handle);
+  if ((!this.readableEnded || this[kOnreadBuffer] !== undefined) && !this.connecting && !drainOnreadTail(this)) {
+    readStart(this, this._handle);
   }
   return Duplex.prototype.read.$call(this, size);
 };
@@ -2479,9 +2488,8 @@ Socket.prototype._read = function _read(size) {
   const socket = this._handle;
   if (this.connecting || !socket) {
     this.once("connect", () => this._read(size));
-  } else if (!drainOnreadTail(this, true)) {
-    socket?.resume?.();
-    restorePausedHold(this, socket);
+  } else if (!drainOnreadTail(this)) {
+    readStart(this, socket);
   }
 };
 

test/js/node/net/node-net.test.ts

@@ -2587,6 +2587,104 @@ it("onread: read() after a redundant pause() still redelivers the declined tail"
   }
 });
 
+it("onread: pause() then read() inside the callback keeps delivering", async () => {
+  // https://github.com/oven-sh/bun/issues/42418 - node's read() restarts the
+  // handle (tryReadStart) right after pause() stopped it, so the rest of the
+  // chunk still reaches the callback without a later resume().
+  const received: string[] = [];
+  const done = Promise.withResolvers<void>();
+  const server = createServer(c => {
+    c.on("error", () => {});
+    c.end("abcdefgh");
+  });
+  let client: Socket | undefined;
+  try {
+    const listening = Promise.withResolvers<void>();
+    server.once("error", listening.reject);
+    server.listen(0, "127.0.0.1", () => {
+      server.off("error", listening.reject);
+      listening.resolve();
+    });
+    await listening.promise;
+    client = createConnection({
+      port: (server.address() as import("node:net").AddressInfo).port,
+      host: "127.0.0.1",
+      onread: {
+        buffer: Buffer.alloc(4),
+        callback(n: number, buf: Buffer) {
+          received.push(buf.toString("latin1", 0, n));
+          client!.pause();
+          client!.read();
+          if (received.length === 2) done.resolve();
+        },
+      },
+    });
+    client.on("error", done.reject);
+    client.on("close", () => done.reject(new Error(`closed before all data was delivered: ${received.join("|")}`)));
+    await done.promise;
+    expect(received).toEqual(["abcd", "efgh"]);
+  } finally {
+    client?.destroy();
+    server.close();
+  }
+});
+
+it("onread: read() then pause() inside the callback stops the delivery", async () => {
+  // The opposite order: pause() is the last word, so the handle stays stopped
+  // until resume() (node's level-triggered handle.reading).
+  const serverSockets: Socket[] = [];
+  const server = createServer(c => {
+    serverSockets.push(c);
+    c.on("error", () => {});
+    c.write("abcdefgh");
+  });
+  const received: string[] = [];
+  const firstDelivery = Promise.withResolvers<void>();
+  const done = Promise.withResolvers<void>();
+  let client: Socket | undefined;
+  try {
+    const listening = Promise.withResolvers<void>();
+    server.once("error", listening.reject);
+    server.listen(0, "127.0.0.1", () => listening.resolve());
+    await listening.promise;
+    client = createConnection({
+      port: (server.address() as import("node:net").AddressInfo).port,
+      host: "127.0.0.1",
+      onread: {
+        buffer: Buffer.alloc(4),
+        callback(n: number, buf: Buffer) {
+          received.push(buf.toString("latin1", 0, n));
+          if (received.length === 1) {
+            client!.read();
+            client!.pause();
+            firstDelivery.resolve();
+          }
+          if (received.length === 3) done.resolve();
+        },
+      },
+    });
+    client.on("error", e => {
+      firstDelivery.reject(e);
+      done.reject(e);
+    });
+    client.on("close", () => {
+      const e = new Error(`closed before all data was delivered: ${received.join("|")}`);
+      firstDelivery.reject(e);
+      done.reject(e);
+    });
+    await firstDelivery.promise;
+    await new Promise<void>(resolve => serverSockets[0].end("wxyz", () => resolve()));
+    for (let i = 0; i < 4; i++) await new Promise(resolve => setImmediate(resolve));
+    expect(received).toEqual(["abcd"]);
+    client.resume();
+    await done.promise;
+    expect(received).toEqual(["abcd", "efgh", "wxyz"]);
+  } finally {
+    client?.destroy();
+    server.close();
+  }
+});
+
 it("onread: a peer FIN does not redeliver the declined tail before resume()", async () => {
   // The EOF path's read(0) must not restart a flow the callback paused: Node's
   // readStop leaves both the tail and the FIN unread until resume().

test/js/node/tls/node-tls-upgrade.test.ts

@@ -109,6 +109,42 @@ test.each([
   },
 );
 
+test("tls.connect({ socket, onread }) over a still connecting net.Socket reads until the peer ends", async () => {
+  // The wrapped socket has not connected yet, so the constructor does not start
+  // the reads. The open of the TLS handle has to start them instead, or the
+  // onread callback stops after its first buffer and the FIN never arrives.
+  const server = tls.createServer(certs, socket => {
+    socket.on("error", () => {});
+    socket.end("abcdefgh");
+  });
+  await once(server.listen(0, "127.0.0.1"), "listening");
+  try {
+    const raw = net.connect({ port: (server.address() as net.AddressInfo).port, host: "127.0.0.1" });
+    raw.on("error", () => {});
+    let bytes = 0;
+    const tlsSocket = tls.connect({
+      socket: raw,
+      ca: certs.cert,
+      servername: "localhost",
+      onread: {
+        buffer: Buffer.alloc(4),
+        callback(n: number) {
+          bytes += n;
+        },
+      },
+    });
+    tlsSocket.on("data", data => (bytes += data.length));
+    const { promise, resolve, reject } = Promise.withResolvers<void>();
+    tlsSocket.on("error", reject);
+    tlsSocket.on("close", () => reject(new Error(`closed before end after ${bytes} bytes`)));
+    tlsSocket.on("end", resolve);
+    await promise;
+    expect(bytes).toBe(8);
+  } finally {
+    server.close();
+  }
+});
+
 // Both peers keep their plaintext 'data' listener across the upgrade.
 test("a STARTTLS exchange hands no TLS bytes to the 'data' listeners of the wrapped sockets (#32239)", async () => {
   const saw: string[] = [];