From 5390ddc6d22946744748cd88a86f0a832b6de311 Mon Sep 17 00:00:00 2001 From: "Jacob \"Doogie\" Thomas" Date: Sat, 12 Sep 2026 13:15:35 +1200 Subject: [PATCH] net: fire socket.setTimeout() when the write queue stalls `_onTimeout` suppressed the idle timeout whenever the handle had any buffered bytes, and returned without refreshing the timer. `this[kTimeout]` is one-shot, so that return consumed it and no later tick ever ran: a peer that stops reading silenced the socket's timeout permanently, which is the condition the timeout exists to surface. Match node in two places: - `_onTimeout` compares the queue size with the previous tick's. A queue that moved is draining, so record it, `_unrefTimer()` and reschedule; a queue that has not moved is stalled, so fire. The recorded size defaults to 0, so an ordinary idle socket still fires on its first period. - When a queued write completes, both `drain` handlers reset the recorded size to 0 and call `_unrefTimer()`, as node's `kAfterAsyncWrite` and `kUpdateTimer` do in `onWriteComplete`. Without the refresh, a write that drained within the first period read as 0 === 0 at the next tick and fired almost immediately after completing; without the reset, a size recorded mid-drain cost an extra period. Measured against node 25.3.0: before after node stalled queue, setTimeout(400) never 803-805ms 801-802ms write completes -> timeout, setTimeout(400) never 401-404ms 400-401ms plain idle socket, setTimeout(500) 501ms 502-503ms 501ms The tests back the queue up until write() returns false and issue no write after setTimeout(), because _write() refreshes the timer. The peer stays paused until the test resumes it, placing the drain before or after the timer's first tick. Each test fails with a different part of the fix removed. --- src/js/node/net.ts | 24 +++++++-- test/js/node/net/node-net.test.ts | 85 ++++++++++++++++++++++++++++++- 2 files changed, 104 insertions(+), 5 deletions(-) diff --git a/src/js/node/net.ts b/src/js/node/net.ts index b4cf11dd51ac..a950fbef0dd8 100644 --- a/src/js/node/net.ts +++ b/src/js/node/net.ts @@ -275,6 +275,7 @@ const { owner_symbol } = require("internal/async_hooks").symbols; // native Listener reachable via accepted socket handles (see a93d2fa48e). const kServerSocket = Symbol("kServerSocket"); const kBytesWritten = Symbol("kBytesWritten"); +const kLastWriteQueueSize = Symbol("kLastWriteQueueSize"); const bunTLSConnectOptions = Symbol.for("::buntlsconnectoptions::"); // tls.Server exposes its native SecureContext constructor through this key so // the SNI dispatch (below) can recognize a raw native context the way Node's @@ -689,6 +690,8 @@ const SocketHandlers = { } else if (res) { self._pendingData = self[kwriteCallback] = null; unrefAfterDrain(self, socket); + self[kLastWriteQueueSize] = 0; + self._unrefTimer(); callback(null); } else { self._pendingData = null; @@ -1531,6 +1534,8 @@ const SocketHandlers2 = { self[kBytesWritten] = socket.bytesWritten; self._pendingData = self[kwriteCallback] = null; unrefAfterDrain(self, socket); + self[kLastWriteQueueSize] = 0; + self._unrefTimer(); callback(null); } else { self[kBytesWritten] = socket.bytesWritten; @@ -2030,10 +2035,21 @@ Socket.prototype._onTimeout = function () { } const handle = this._handle; - // if there is a handle, and it has pending data, - // we suppress the timeout because a write is in progress - if (handle && getBufferedAmount(handle) > 0) { - return; + if (handle) { + // A queue that moved since the last tick is draining, so reschedule; one + // that has not moved is stalled, which is what the timeout exists to + // surface. Draining to empty counts as movement — firing on the tick that + // first reads 0 would cut the idle period short. Defaulting to 0 keeps an + // ordinary idle socket (queue 0, unchanged) firing on its first tick. + const writeQueueSize = getBufferedAmount(handle); + const lastWriteQueueSize = this[kLastWriteQueueSize] ?? 0; + if (writeQueueSize !== lastWriteQueueSize) { + this[kLastWriteQueueSize] = writeQueueSize; + // this[kTimeout] is one-shot; without this the timer is consumed here and + // no later tick ever runs. + this._unrefTimer(); + return; + } } this.emit("timeout"); }; diff --git a/test/js/node/net/node-net.test.ts b/test/js/node/net/node-net.test.ts index 93a6fcd55677..6cdf55d4b145 100644 --- a/test/js/node/net/node-net.test.ts +++ b/test/js/node/net/node-net.test.ts @@ -1,6 +1,6 @@ import { Socket as _BunSocket, TCPSocketListener } from "bun"; import { heapStats } from "bun:jsc"; -import { describe, expect, it } from "bun:test"; +import { afterEach, describe, expect, it } from "bun:test"; import { blackholePortSource, bunEnv, @@ -21,6 +21,7 @@ import { randomUUID } from "node:crypto"; import { once } from "node:events"; import fs from "node:fs"; import { + type AddressInfo, BlockList, connect, createConnection, @@ -3542,3 +3543,85 @@ it.skipIf(isWindows || isMusl).each([ const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); expect({ stdout, stderr, exitCode }).toEqual({ stdout: expected + "\n", stderr: "", exitCode: 0 }); }); + +// A stalled write queue is the condition socket.setTimeout() exists to surface, +// and suppressing on any non-empty queue silenced it permanently because the +// timer is one-shot. node compares against the previous tick's queue size, and +// on write completion both refreshes the timer and resets the recorded size. +// +// No write is issued after setTimeout(): Socket#_write calls _unrefTimer(), so a +// pump would keep refreshing the timer. Writes continue until write() returns +// false so the queue is backed up on every platform (winsock can take a small +// send whole). The peer stays paused until the test resumes it, which places the +// drain before or after the timer's first tick deterministically. +describe("socket.setTimeout() and the write queue", () => { + const IDLE = 600; + const servers: Server[] = []; + const sockets: Socket[] = []; + + afterEach(() => { + for (const socket of sockets.splice(0)) socket.destroy(); + for (const server of servers.splice(0)) server.close(); + }); + + async function connectToPausedPeer() { + const { promise: accepted, resolve } = Promise.withResolvers(); + const server = createServer(peer => { + peer.on("error", () => {}); + peer.pause(); + sockets.push(peer); + resolve(peer); + }).listen(0, "127.0.0.1"); + servers.push(server); + await once(server, "listening"); + const socket = connect({ host: "127.0.0.1", port: (server.address() as AddressInfo).port }); + sockets.push(socket); + socket.on("error", () => {}); + await once(socket, "connect"); + const peer = await accepted; + + const chunk = Buffer.alloc(32 << 20, "x"); + while (socket.write(chunk)) {} + expect(socket.bufferSize).toBeGreaterThan(0); + return { socket, peer }; + } + + it("fires when the peer stops reading and the queue stops draining", async () => { + const { socket } = await connectToPausedPeer(); + socket.setTimeout(IDLE); + + const outcome = await Promise.race([ + once(socket, "timeout").then(() => ({ fired: "timeout", stalled: socket.bufferSize > 0 })), + Bun.sleep(IDLE * 10).then(() => ({ fired: "none", stalled: socket.bufferSize > 0 })), + ]); + expect(outcome).toEqual({ fired: "timeout", stalled: true }); + }); + + it.each([ + ["before the first tick", 0.25], + ["after one tick has sampled the queue", 1.25], + ])("fires one idle period after the queue drains %s", async (_, resumeAfterPeriods) => { + const { socket, peer } = await connectToPausedPeer(); + socket.setTimeout(IDLE); + + let timedOutAt: number | undefined; + const timedOut = new Promise(resolve => + socket.once("timeout", () => { + timedOutAt = performance.now(); + resolve(); + }), + ); + await Bun.sleep(IDLE * resumeAfterPeriods); + const drained = once(socket, "drain"); + peer.resume(); + await drained; + const drainedAt = performance.now(); + await Promise.race([timedOut, Bun.sleep(IDLE * 3)]); + + const afterDrain = timedOutAt === undefined ? undefined : timedOutAt - drainedAt; + expect({ + firedBeforeDrain: afterDrain !== undefined && afterDrain < 0, + oneIdlePeriodAfterDrain: afterDrain !== undefined && afterDrain >= IDLE * 0.9 && afterDrain < IDLE * 1.5, + }).toEqual({ firedBeforeDrain: false, oneIdlePeriodAfterDrain: true }); + }); +});