From b62b58121e55f351028fb7ba42049341644aa5a8 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 8 Oct 2026 13:58:24 +0000 Subject: [PATCH 1/3] node:http: ServerResponse keeps Node's need-drain flag, and a queued response counts only its own bytes writableNeedDrain was derived from the buffer of the connection, the armed drain callback and the per-turn count. A response that waits behind another one read the bytes of the response ahead, so pipe() waited for a 'drain' that never came. write() now stores kNeedDrain where it returns false. One emitter tests the flag, waits while the drain callback of the handle is armed, clears the flag and emits 'drain'. The getter reads the flag for a response with a handle. writableLength counts the unsent bytes of the connection only for the response that has the socket. _send() no longer replaces the armed drain callback. --- src/js/node/_http_server.ts | 89 +- .../node/http/node-http-backpressure.test.ts | 1081 ++++++++++++++++- .../node/http/node-http-syscall-fault.test.ts | 90 ++ test/regression/issue/19111.test.ts | 34 +- 4 files changed, 1252 insertions(+), 42 deletions(-) diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 6c66f7904a12..6c409185fb84 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -60,6 +60,7 @@ const { getMaxHTTPHeaderSize, fakeSocketSymbol, kOutHeaders, + kNeedDrain, onDataIncomingMessage, http1ServerPipeline, } = require("internal/http"); @@ -1524,7 +1525,7 @@ const kKeepAliveTimeoutSet = Symbol("keepAliveTimeoutSet"); const kKeepAliveIdleStart = Symbol("keepAliveIdleStart"); // HTTP/1.1 pipelining (responses queued behind an in-flight response): // - on the socket: array of queued ServerResponses, in arrival order -// - on a queued response: { ops, bytes, needDrain, ended, isAncient } while it +// - on a queued response: { ops, bytes, ended, isAncient } while it // is queued (undefined once it owns the socket) const kPipelinedResponses = Symbol("kPipelinedResponses"); const kPipelinedQueuedState = Symbol("kPipelinedQueuedState"); @@ -2829,7 +2830,6 @@ function queuePipelinedResponse(socket, res, isAncient) { ops: [], bytes: 0, headerBytes: 0, - needDrain: false, ended: false, isAncient, socket, @@ -2973,11 +2973,7 @@ function advanceResponsePipeline(server, socket) { // internal/http1_server_fallback connection (a foreign duplex): the // response's JS handle writes to the socket itself (nothing to switch // natively), and the prototype assignSocket installs the 'close' listener - // a plain stream needs. Clear `finished` first, or assignSocket's _flush - // emits 'prefinish' before the ops replay below. - if (queued.ended) { - res.finished = false; - } + // a plain stream needs. res.assignSocket(socket); } @@ -2996,6 +2992,8 @@ function advanceResponsePipeline(server, socket) { res.finished = false; } res[kReplayingPipelinedOps] = true; + // The replay leaves the flag as it is, like _flushOutput: https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L1201-L1225 + const needDrain = res[kNeedDrain]; try { for (let i = 0; i < opsLength; i++) { const op = ops[i]; @@ -3013,13 +3011,14 @@ function advanceResponsePipeline(server, socket) { } } finally { res[kReplayingPipelinedOps] = false; + res[kNeedDrain] = needDrain; } } - if (queued.needDrain && !queued.ended) { - // write() reported backpressure while the response was queued: the native drain callback emits 'drain' if the flush hit backpressure, else emit it now. - if (!hitBackpressure) { - process.nextTick(emitPipelinedDrainNT, res); - } + // A write() returned false while the response was queued. The drain callback of the handle emits its 'drain' when the replay met backpressure. + if (res[kNeedDrain] && !queued.ended && !hitBackpressure) { + // With nothing left to send it comes here, like in _flush(): https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L1194-L1197 + if (res.writableLength === 0) emitDrainIfOwed(res); + else process.nextTick(emitDrainIfOwed, res); } } @@ -3027,8 +3026,12 @@ function markResponseEndedNT(res) { res._ended = true; } -function emitPipelinedDrainNT(res) { - if (!res.destroyed && !res.finished) { +// A 'drain' answers a write() that returned false, once the socket holds none of its bytes: https://github.com/nodejs/node/blob/v26.3.0/lib/_http_server.js#L885-L888 +function emitDrainIfOwed(res) { + if (res[kNeedDrain] && !res.destroyed && !res.finished) { + // The drain callback of the handle is armed: it emits this 'drain' when the socket has taken the bytes. + if (res[kHandle].onwritable !== undefined) return; + res[kNeedDrain] = false; res.emit("drain"); } } @@ -3105,7 +3108,7 @@ function bufferPipelinedWrite(res, queued, chunk, encoding, callback) { addPipelineOutgoingData(queued, bytes); } if (queued.bytes >= res.writableHighWaterMark) { - queued.needDrain = true; + res[kNeedDrain] = true; return false; } return true; @@ -3736,7 +3739,13 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { const flags = handle.flags; if (!!(flags & NodeHTTPResponseFlags.closed_or_completed)) { - // Socket already gone: like Node's _writeRaw(), report false and drop the callback. + // Node discards a write to a response without a body before it looks at the socket: https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L983-L992 + if (!this._hasBody) { + if (callback) process.nextTick(callback); + return true; + } + // Socket already gone: like Node's _writeRaw(), report false and drop the callback: https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L400-L406 + this[kNeedDrain] = true; return false; } @@ -3777,6 +3786,7 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { // We need to defer the callback until the write actually goes through. this[kPendingCallbacks].push(callback); } + this[kNeedDrain] = true; return false; } @@ -3787,6 +3797,7 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { if (hasBody && handle.bufferedAmount > 0) { this[kPendingCallbacks].push(callback); handle.onwritable = allowWritesToContinue.bind(this); + this[kNeedDrain] = true; return false; } process.nextTick(callback); @@ -3817,7 +3828,9 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { // An empty chunk reports the count too: Node returns state.length < highWaterMark for every write. const buffered = this[kBytesBuffered]; - return !buffered || buffered < this.writableHighWaterMark; + if (!buffered || buffered < this.writableHighWaterMark) return true; + this[kNeedDrain] = true; + return false; }; // advanceResponsePipeline replays buffered ops through these; patched res.write/res.end (compression middleware) must not see them again. @@ -3830,9 +3843,7 @@ function flushWriteAccountingNT(res) { res[kAccountingFlushScheduled] = false; const needsDrain = (res[kBytesBuffered] ?? 0) >= res.writableHighWaterMark; res[kBytesBuffered] = 0; - if (needsDrain && !res.destroyed && !res.finished) { - res.emit("drain"); - } + if (needsDrain) emitDrainIfOwed(res); } function scheduleWriteAccountingFlush(res) { if (res[kAccountingFlushScheduled]) return; @@ -3886,16 +3897,8 @@ ServerResponse.prototype._implicitHeader = function () { Object.defineProperty(ServerResponse.prototype, "writableNeedDrain", { get() { - // True between a write() that returned false and the next 'drain'. An armed native drain callback means a later write flushed the bytes. - const handle = this[kHandle]; - return ( - !this.destroyed && - !this.finished && - ((handle?.bufferedAmount ?? 0) !== 0 || - handle?.onwritable !== undefined || - (this[kBytesBuffered] ?? 0) >= this.writableHighWaterMark || - (this[kPipelinedQueuedState]?.needDrain ?? false)) - ); + // Node answers true for a response without a server after a refused write(), and no 'drain' of its socket reaches it. Here it stays false, so a later pipe() into it still flows. + return this[kHandle] !== undefined && !this.destroyed && !this.finished && this[kNeedDrain]; }, }); @@ -3917,8 +3920,11 @@ Object.defineProperty(ServerResponse.prototype, "writableLength", { // Node.js's outputData accounting) plus whatever the native handle still // has buffered, plus anything buffered while queued behind a pipelined // response. + // The bytes of the handle count only while the response has the socket: https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L221-L226 return ( - (this[kBytesBuffered] ?? 0) + (this[kHandle]?.bufferedAmount ?? 0) + (this[kPipelinedQueuedState]?.bytes ?? 0) + (this[kBytesBuffered] ?? 0) + + (this[kPipelinedQueuedState]?.bytes ?? 0) + + (this[kSocket] ? (this[kHandle]?.bufferedAmount ?? 0) : 0) ); }, }); @@ -3945,6 +3951,9 @@ ServerResponse.prototype._send = function (data, encoding, callback, _byteLength } const strict = strictContentLength(this, this[headerStateSymbol], false); + // Not `callback`: it would replace the armed drain callback of a write() that returned false, and that 'drain' would never come. + const onWritable = allowWritesToContinue.bind(this); + let result = 0; if (this[headerStateSymbol] !== NodeHTTPHeaderState.sent) { handle.cork(() => { const renderedHeaders = renderNativeHeaders(this); @@ -3962,10 +3971,14 @@ ServerResponse.prototype._send = function (data, encoding, callback, _byteLength releaseRenderedHeaders(renderedHeaders); } this[headerStateSymbol] = NodeHTTPHeaderState.sent; - handle.write(data, encoding, callback, strict); + result = handle.write(data, encoding, onWritable, strict); }); } else { - handle.write(data, encoding, callback, strict); + result = handle.write(data, encoding, onWritable, strict); + } + if ($isCallable(callback)) { + if (result < 0) this[kPendingCallbacks].push(callback); + else process.nextTick(callback); } }; @@ -4018,7 +4031,8 @@ ServerResponse.prototype.assignSocket = function (socket) { this.socket = socket; this.emit("socket", socket); // Like Node.js: drain anything written before the socket was assigned. - this._flush(); + // With a handle nothing is in outputData, and advanceResponsePipeline replays the queued writes before their 'drain' and 'prefinish'. + if (!this[kHandle]) this._flush(); }; // Backed by real storage (the native response handle exposes no such flag): @@ -4081,9 +4095,12 @@ ServerResponse.prototype.flushHeaders = function () { if (this[headerStateSymbol] === NodeHTTPHeaderState.sent) return; // Should be idempotent. if (this[headerStateSymbol] !== NodeHTTPHeaderState.assigned) this._implicitHeader(); - if (this[kPipelinedQueuedState] !== undefined) { + const queued = this[kPipelinedQueuedState]; + if (queued !== undefined) { // Queued pipelined response: its headers go out when it is assigned the // socket (advanceResponsePipeline) - nothing can be flushed before then. + // They count as buffered from here, like the _send("") of Node: https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L1227-L1234 + accountQueuedHeaderBytes(this, queued); return; } @@ -4155,7 +4172,7 @@ function callWriteHeadIfObservable(self, headerState, fromEnd?) { function allowWritesToContinue() { this._callPendingCallbacks(); - this.emit("drain"); + emitDrainIfOwed(this); } function failPendingWriteCallbacks(res, error) { diff --git a/test/js/node/http/node-http-backpressure.test.ts b/test/js/node/http/node-http-backpressure.test.ts index 22c28e6a8d87..b4bbe10dba25 100644 --- a/test/js/node/http/node-http-backpressure.test.ts +++ b/test/js/node/http/node-http-backpressure.test.ts @@ -8,14 +8,14 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { spawn } from "node:child_process"; import { once } from "node:events"; -import { readFileSync } from "node:fs"; +import { createReadStream, readFileSync } from "node:fs"; import http from "node:http"; import http2 from "node:http2"; import https from "node:https"; import type { AddressInfo } from "node:net"; import net from "node:net"; import path from "node:path"; -import { Duplex, duplexPair, finished, Readable } from "node:stream"; +import { Duplex, duplexPair, finished, pipeline, Readable, Writable } from "node:stream"; import nodeTls from "node:tls"; import { Worker } from "node:worker_threads"; @@ -948,6 +948,691 @@ describe("backpressure", () => { expect(client.received).toBe(2 * (client.headLength + BODY)); }); + // Node keeps one flag for each response: a write() of that response + // returned false, and the 'drain' for it has not come yet + // (https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L881-L899). + // pipe(), pipeline() and Writable.toWeb() read it as writableNeedDrain and + // wait for a 'drain' while it is true, so a true that no write() caused is + // a response that is never written. The bytes that the socket still holds + // do not set it, and writableLength counts them only for the response that + // has the socket. + describe("writableNeedDrain and writableLength are the response's own", () => { + const FIRST = Buffer.alloc(8 * 1024 * 1024, "a"); + const CHUNK16 = Buffer.alloc(16 * 1024, "b"); + const certPath = path.join(keysDir, "agent1-cert.pem"); + const payload = tlsOptions.cert; + const source = () => Readable.from([payload]); + const toWeb = (res: http.ServerResponse) => Writable.toWeb(res as unknown as Writable); + + // Two requests in one write. The second one asks for the close, so `done` has every byte. + const pipelined = (method = "GET") => + `GET /first HTTP/1.1\r\nHost: localhost\r\n\r\n${method} /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n`; + const single = "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"; + + // The status line and the body of the response that follows a first body of `firstBody` bytes. + function secondResponse(bytes: Buffer, firstBody = FIRST.length) { + const start = bytes.indexOf("\r\n\r\n") + 4 + firstBody; + const end = bytes.indexOf("\r\n\r\n", start); + if (end < 0) return { status: "", body: Buffer.alloc(0) }; + const status = bytes.subarray(start, bytes.indexOf("\r\n", start)).toString("latin1"); + return { status, body: bytes.subarray(end + 4) }; + } + + // A first response that stays open with 8 MB written in 16 KB chunks. The + // part that the kernel does not take waits in the server, and the next + // response waits behind it on every platform. + function writeFirst(res: http.ServerResponse) { + res.setHeader("Content-Length", FIRST.length); + for (let sent = 0; sent < FIRST.length; sent += CHUNK16.length) res.write(FIRST.subarray(0, CHUNK16.length)); + return res; + } + + const ways: [string, string, (res: http.ServerResponse) => unknown][] = [ + ["stream.pipe(res)", "GET", res => source().pipe(res)], + ["stream.pipe(res) for a HEAD request", "HEAD", res => source().pipe(res)], + ["stream.pipeline(stream, res)", "GET", res => pipeline(source(), res, () => {})], + [ + "stream.pipeline(async generator, res)", + "GET", + res => + pipeline( + async function* () { + yield payload; + }, + res, + () => {}, + ), + ], + ["fs.createReadStream(path).pipe(res)", "GET", res => createReadStream(certPath).pipe(res)], + [ + "a write() through Writable.toWeb(res)", + "GET", + async res => { + const writer = toWeb(res).getWriter(); + await writer.write(payload); + await writer.close(); + }, + ], + ["pipeTo(Writable.toWeb(res))", "GET", res => Readable.toWeb(source()).pipeTo(toWeb(res))], + [ + "a write() that waits for 'drain' while writableNeedDrain is true", + "GET", + async res => { + if (res.writableNeedDrain) await once(res, "drain"); + res.end(payload); + }, + ], + [ + "a write() that waits for 'drain' while writableLength is above 0", + "GET", + async res => { + while (res.writableLength > 0) await once(res, "drain"); + res.end(payload); + }, + ], + ]; + + // The first response is res.end(8 MB) to a client that does not read yet. + // The second request arrives in the same read, and `respond` answers it. + async function answersBehindUnfinished( + tls: boolean, + method: string, + respond: (res: http.ServerResponse) => unknown, + ) { + const fresh = Promise.withResolvers<{ writableNeedDrain: boolean; writableLength: number; waits: boolean }>(); + const errors: unknown[] = []; + await using server = createServer(tls, (req, res) => { + if (req.url === "/first") return void res.end(FIRST); + res.setHeader("Content-Length", payload.length); + fresh.resolve({ + writableNeedDrain: res.writableNeedDrain, + writableLength: res.writableLength, + waits: res.socket === null, + }); + Promise.resolve(respond(res)).catch(error => errors.push(error)); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, pipelined(method), { tls }); + expect(await Promise.race([fresh.promise, client.done])).toEqual({ + writableNeedDrain: false, + writableLength: 0, + // Winsock can take the whole first response in one send(): the second one then waits behind nothing. + waits: process.platform === "win32" ? expect.any(Boolean) : true, + }); + client.resume(); + const { bytes, ended } = await client.done; + const { status, body } = secondResponse(bytes); + expect({ status, body: body.toString("latin1"), ended, errors }).toEqual({ + status: "HTTP/1.1 200 OK", + body: method === "HEAD" ? "" : payload.toString("latin1"), + ended: true, + errors: [], + }); + } + + it.each(ways)("a response that waits behind an unfinished one is answered: %s", (_way, method, respond) => + answersBehindUnfinished(false, method, respond), + ); + + it("a response that waits behind an unfinished one is answered over TLS", () => + answersBehindUnfinished(true, "GET", res => source().pipe(res))); + + it("a response that waits counts the bytes that it buffered itself", async () => { + const counted = Promise.withResolvers(); + let first: http.ServerResponse; + await using server = createServer(false, (req, res) => { + if (req.url === "/first") return void (first = writeFirst(res)); + res.flushHeaders(); + const head = res.writableLength; + const returned = res.write(Buffer.alloc(10, "b")); + const headAndChunk = res.writableLength; + counted.resolve({ + headIsCounted: head > 0, + returned, + writableNeedDrain: res.writableNeedDrain, + // A head and a 10 byte chunk, not the megabytes that the first response still holds. + ownBytesOnly: headAndChunk > head && headAndChunk < 1024, + }); + res.end(); + first.end(); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, pipelined()); + expect(await Promise.race([counted.promise, client.done])).toEqual({ + headIsCounted: true, + returned: true, + writableNeedDrain: false, + ownBytesOnly: true, + }); + }); + + it("a write() that returned false while the response waited gets one 'drain', after the response has the socket", async () => { + const events: string[] = []; + let first: http.ServerResponse; + let total = 0; + await using server = createServer(false, (req, res) => { + if (req.url === "/first") return void (first = writeFirst(res)); + // As many chunks as reach the high water mark: the last write() returns false. + const count = Math.ceil(res.writableHighWaterMark / CHUNK16.length); + total = count * CHUNK16.length; + res.setHeader("Content-Length", total + 4); + // The callback of the last write runs after the 'drain': a writer that continues there sees the flag cleared. + const lastWritten = () => { + events.push(`callback of the last write: writableNeedDrain ${res.writableNeedDrain}`); + res.end("tail"); + }; + let returned = true; + for (let i = 0; i < count; i++) returned = res.write(CHUNK16, i === count - 1 ? lastWritten : undefined); + events.push(`waits: write() returned ${returned}, writableNeedDrain ${res.writableNeedDrain}`); + res.on("socket", () => events.push(`'socket': writableNeedDrain ${res.writableNeedDrain}`)); + res.on("drain", () => + events.push(`'drain': writableNeedDrain ${res.writableNeedDrain}, writableLength ${res.writableLength}`), + ); + first.end(); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, pipelined()); + client.resume(); + const { bytes, ended } = await client.done; + const { status, body } = secondResponse(bytes); + expect({ + events, + status, + length: body.length, + inOrder: body.equals(Buffer.concat([Buffer.alloc(total, "b"), Buffer.from("tail")])), + ended, + }).toEqual({ + events: [ + "waits: write() returned false, writableNeedDrain true", + "'socket': writableNeedDrain true", + "'drain': writableNeedDrain false, writableLength 0", + "callback of the last write: writableNeedDrain false", + ], + status: "HTTP/1.1 200 OK", + length: total + 4, + inOrder: true, + ended: true, + }); + }); + + describe("8 MB buffered behind an open response, on a server whose write() returns true up to 16 MB", () => { + const QUEUED = 8 * 1024 * 1024; + + // The second response buffers 8 MB while the first one is open. At its + // turn the socket takes only part of them. No write() returned false, + // so no 'drain' is owed and writableNeedDrain stays false. + // `goOn` continues the response. `lastWritten` is the callback of its last buffered write. + type Seen = Record; + async function afterTheReplay( + goOn: (res: http.ServerResponse, lastWritten: Promise) => Promise, + ): Promise { + const seen = Promise.withResolvers(); + let first: http.ServerResponse; + await using server = http.createServer({ highWaterMark: 2 * QUEUED }, (req, res) => { + if (req.url === "/first") return void (first = res); + res.setHeader("Content-Length", QUEUED + 4); + const lastWritten = Promise.withResolvers(); + const returned = new Set(); + for (let sent = CHUNK16.length; sent < QUEUED; sent += CHUNK16.length) returned.add(res.write(CHUNK16)); + returned.add(res.write(CHUNK16, () => lastWritten.resolve())); + goOn(res, lastWritten.promise).then( + result => seen.resolve({ returned: [...returned], ...result }), + seen.reject, + ); + first.end("first"); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, pipelined()); + client.resume(); + const [result, { bytes, ended }] = await Promise.all([seen.promise, client.done]); + const { status, body } = secondResponse(bytes, "first".length); + return { + ...result, + status, + length: body.length, + inOrder: body.equals(Buffer.concat([Buffer.alloc(QUEUED, "b"), Buffer.from("tail")])), + ended, + }; + } + const whole = { status: "HTTP/1.1 200 OK", length: QUEUED + 4, inOrder: true, ended: true }; + + it("no 'drain' comes for them", async () => { + const result = await afterTheReplay(async (res, lastWritten) => { + let drains = 0; + res.on("drain", () => drains++); + await lastWritten; + const state = { drains, writableNeedDrain: res.writableNeedDrain }; + res.end("tail"); + return state; + }); + expect(result).toEqual({ returned: [true], drains: 0, writableNeedDrain: false, ...whole }); + }); + + // Writable.toWeb() does not write a chunk while writableNeedDrain is true. + it("a chunk written through Writable.toWeb(res) right after them is delivered", async () => { + const result = await afterTheReplay(async res => { + // One turn after 'socket', the buffered writes have gone to the socket. + await once(res, "socket"); + await new Promise(resolve => setImmediate(resolve)); + const state = { writableNeedDrain: res.writableNeedDrain }; + const writer = toWeb(res).getWriter(); + await writer.write("tail"); + await writer.close(); + return state; + }); + expect(result).toEqual({ returned: [true], writableNeedDrain: false, ...whole }); + }); + }); + + it("a response class that overrides assignSocket() sees one 'prefinish' for a response that ended while it waited", async () => { + class Response extends http.ServerResponse { + assignSocket(socket: net.Socket) { + super.assignSocket(socket); + } + } + let prefinish = 0; + let first: http.ServerResponse; + await using server = http.createServer({ ServerResponse: Response }, (req, res) => { + if (req.url === "/first") return void (first = res); + res.on("prefinish", () => prefinish++); + res.end("second"); + first.end("first"); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, pipelined()); + client.resume(); + const { bytes, ended } = await client.done; + const { status, body } = secondResponse(bytes, "first".length); + expect({ prefinish, status, body: body.toString(), ended }).toEqual({ + prefinish: 1, + status: "HTTP/1.1 200 OK", + body: "second", + ended: true, + }); + }); + + // One request. Megabytes that no write() of the body put there wait in + // the socket, and pipe() still starts at once. + const unsent: [string, (res: http.ServerResponse) => unknown][] = [ + [ + "a large head that flushHeaders() sent", + res => { + res.setHeader("X-Padding", FIRST.toString()); + res.flushHeaders(); + }, + ], + [ + "a large interim response that res.socket.write() sent", + res => + res.socket!.write( + Buffer.concat([Buffer.from("HTTP/1.1 102 Processing\r\nX-Padding: "), FIRST, Buffer.from("\r\n\r\n")]), + ), + ], + ]; + it.each(unsent)("a response is piped into behind %s", async (_name, prepare) => { + const fresh = Promise.withResolvers(); + await using server = createServer(false, (req, res) => { + res.setHeader("Content-Length", payload.length); + prepare(res); + fresh.resolve(res.writableNeedDrain); + source().pipe(res); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + // Not pausedClient(): it searches what it has received for the end of the head, and this head is 8 MB. + const socket = net.connect((server.address() as AddressInfo).port, "127.0.0.1"); + try { + socket.pause(); + const received: Buffer[] = []; + socket.on("data", (chunk: Buffer) => received.push(chunk)); + const closed = once(socket, "close"); + await once(socket, "connect"); + socket.write(single); + expect(await Promise.race([fresh.promise, closed])).toBe(false); + socket.resume(); + await closed; + expect(Buffer.concat(received).subarray(-payload.length).toString("latin1")).toBe(payload.toString("latin1")); + } finally { + socket.destroy(); + } + }); + + // Node emits 'drain' on a response only when the response has nothing left + // to send (https://github.com/nodejs/node/blob/v26.3.0/lib/_http_server.js#L885-L888), + // so a handler can wait for an empty buffer with the loop in endWhenEmpty(). + // With highWaterMark 0 every write() returns false and every turn reaches + // the mark. + describe.each([ + ["the default highWaterMark", {}], + ["highWaterMark 0", { highWaterMark: 0 }], + ] as [string, http.ServerOptions][])("on a server with %s", (_mark, options) => { + type Drain = { writableNeedDrain: boolean; writableLength: number }; + const record = (res: http.ServerResponse, drains: Drain[]) => + res.on("drain", () => + drains.push({ writableNeedDrain: res.writableNeedDrain, writableLength: res.writableLength }), + ); + async function endWhenEmpty(res: http.ServerResponse) { + while (res.writableLength > 0) await once(res, "drain"); + res.end("tail"); + } + + // Four 16 KB writes reach the default mark, and the socket then takes + // only part of a large one, all in one turn. That is one backlog. + it("one backlog gets one 'drain', after the socket has taken all of it", async () => { + const TOTAL = 4 * CHUNK16.length + FIRST.length; + const drains: Drain[] = []; + const wrote = Promise.withResolvers(); + await using server = http.createServer(options, (req, res) => { + if (req.url === "/ping") return void res.end("pong"); + res.setHeader("Content-Length", TOTAL); + record(res, drains).on("drain", () => res.end()); + for (let i = 0; i < 4; i++) res.write(CHUNK16); + wrote.resolve(res.write(FIRST)); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const port = (server.address() as AddressInfo).port; + using client = pausedClient(port, single, { collect: false }); + expect(await Promise.race([wrote.promise, client.done])).toBe(false); + await ping(port); + // The client has not read anything yet. Winsock can take it all in one send(). + if (process.platform !== "win32") expect(drains).toEqual([]); + client.resume(); + await client.done; + expect({ drains, received: client.received - client.headLength }).toEqual({ + drains: [{ writableNeedDrain: false, writableLength: 0 }], + received: TOTAL, + }); + }); + + // The socket takes the small write and only part of the large one. + it("a wait for an empty buffer ends after a small write and 8 MB in one turn", async () => { + const TOTAL = 10 + FIRST.length + 4; + const drains: Drain[] = []; + const wrote = Promise.withResolvers(); + await using server = http.createServer(options, (req, res) => { + if (req.url === "/ping") return void res.end("pong"); + res.setHeader("Content-Length", TOTAL); + record(res, drains); + res.write(Buffer.alloc(10, "b")); + wrote.resolve(res.write(FIRST)); + endWhenEmpty(res); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const port = (server.address() as AddressInfo).port; + using client = pausedClient(port, single, { collect: false }); + expect(await Promise.race([wrote.promise, client.done])).toBe(false); + await ping(port); + client.resume(); + const { ended } = await client.done; + expect({ + overUnsentBytes: drains.filter(drain => drain.writableLength > 0), + received: client.received - client.headLength, + ended, + }).toEqual({ overUnsentBytes: [], received: TOTAL, ended: true }); + }); + + // The response waited behind another one and one of its writes returned + // false. At its turn its buffered writes go out, and 8 MB are written + // in the same tick. + const laterWrites: [string, (first: http.ServerResponse, goOn: () => void) => (() => void) | undefined][] = [ + ["the callback of its first buffered write", (_first, goOn) => goOn], + ["a 'finish' listener of the response ahead", (first, goOn) => void first.on("finish", goOn)], + ]; + it.each(laterWrites)( + "a wait for an empty buffer ends after 8 MB written from %s, once the response has the socket", + async (_from, arrange) => { + const drains: Drain[] = []; + let first: http.ServerResponse; + let total = 0; + let returned: boolean | undefined; + await using server = http.createServer(options, (req, res) => { + if (req.url === "/first") return void (first = res); + // As many chunks as reach the high water mark, and at least one: the last write() returns false. + const count = Math.max(1, Math.ceil(res.writableHighWaterMark / CHUNK16.length)); + total = count * CHUNK16.length + FIRST.length + 4; + res.setHeader("Content-Length", total); + record(res, drains); + const callback = arrange(first, () => { + res.write(FIRST); + endWhenEmpty(res); + }); + for (let i = 0; i < count; i++) returned = res.write(CHUNK16, i === 0 ? callback : undefined); + first.end("first"); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, pipelined()); + client.resume(); + const { bytes, ended } = await client.done; + const { status, body } = secondResponse(bytes, "first".length); + expect({ + returned, + overUnsentBytes: drains.filter(drain => drain.writableLength > 0), + status, + length: body.length, + tail: body.subarray(-4).toString(), + ended, + }).toEqual({ + returned: false, + overUnsentBytes: [], + status: "HTTP/1.1 200 OK", + length: total, + tail: "tail", + ended: true, + }); + }, + ); + }); + + it("a server with highWaterMark 0 answers a response that is piped into", async () => { + const fresh = Promise.withResolvers(); + await using server = http.createServer({ highWaterMark: 0 }, (req, res) => { + res.setHeader("Content-Length", payload.length); + fresh.resolve(res.writableNeedDrain); + source().pipe(res); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, single); + expect(await Promise.race([fresh.promise, client.done])).toBe(false); + client.resume(); + const { bytes, ended } = await client.done; + expect({ body: bytes.subarray(-payload.length).toString("latin1"), ended }).toEqual({ + body: payload.toString("latin1"), + ended: true, + }); + }); + + // res._send() is the raw write below write(). It leaves the 'drain' of an + // earlier write() in place, and its callback waits for the socket too. + const rawSends: [string, number, (res: http.ServerResponse, sent: () => void) => unknown, string[]][] = [ + // @ts-ignore _send() is not in the types + ["res._send(chunk)", 100 * 1024, res => res._send(Buffer.alloc(100 * 1024, "b")), []], + // @ts-ignore _send() is not in the types + ["res._send(chunk, encoding, callback)", 1, (res, sent) => res._send("b", "latin1", sent), ["sent"]], + ]; + it.each(rawSends)( + "a write() that returned false gets its 'drain' when %s follows it", + async (_call, length, send, callbacks) => { + const events: string[] = []; + const wrote = Promise.withResolvers(); + await using server = createServer(false, (req, res) => { + res.setHeader("Content-Length", FIRST.length + length); + res.on("drain", () => { + events.push(`'drain': writableNeedDrain ${res.writableNeedDrain}`); + res.end(); + }); + const returned = res.write(FIRST); + send(res, () => events.push("sent")); + wrote.resolve(returned); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const port = (server.address() as AddressInfo).port; + using client = pausedClient(port, single, { collect: false }); + expect(await Promise.race([wrote.promise, client.done])).toBe(false); + await ping(port); + // The client has not read anything yet. Winsock can take it all in one send(). + if (process.platform !== "win32") expect(events).toEqual([]); + client.resume(); + const { ended } = await client.done; + expect({ events: events.toSorted(), received: client.received - client.headLength, ended }).toEqual({ + events: ["'drain': writableNeedDrain false", ...callbacks], + received: FIRST.length + length, + ended: true, + }); + }, + ); + + // The constructor of a response class runs before the server knows that + // the response has to wait. + it("a response class whose constructor runs behind an unfinished response counts none of its bytes", async () => { + const seen = Promise.withResolvers(); + let first: http.ServerResponse; + class Response extends http.ServerResponse { + constructor(req: http.IncomingMessage, options?: unknown) { + // @ts-ignore the second argument is not in the types + super(req, options); + if (req.url === "/second") { + seen.resolve({ + earlierResponsePending: first.writableLength > 0, + writableLength: this.writableLength, + writableNeedDrain: this.writableNeedDrain, + }); + } + } + } + await using server = http.createServer({ ServerResponse: Response }, (req, res) => { + if (req.url === "/first") return void (first = res).end(FIRST); + res.end("second"); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, pipelined()); + expect(await Promise.race([seen.promise, client.done])).toEqual({ + // Winsock can take the whole first response in one send(). + earlierResponsePending: process.platform === "win32" ? expect.any(Boolean) : true, + writableLength: 0, + writableNeedDrain: false, + }); + client.resume(); + const { bytes, ended } = await client.done; + expect({ body: secondResponse(bytes).body.toString(), ended }).toEqual({ body: "second", ended: true }); + }); + + // The write() in the 'socket' listener is not one of the writes that the + // response buffered while it waited, so it gets a 'drain' of its own. + it("a write() in the 'socket' listener of a response that waited gets one 'drain', after the socket has taken its bytes", async () => { + const wrote = Promise.withResolvers(); + const drains: object[] = []; + let first: http.ServerResponse; + await using server = createServer(false, (req, res) => { + if (req.url === "/first") return void (first = res); + res.setHeader("Content-Length", FIRST.length + 4); + res.on("socket", () => wrote.resolve(res.write(FIRST))); + res.on("drain", () => { + drains.push({ writableNeedDrain: res.writableNeedDrain, writableLength: res.writableLength }); + res.end("tail"); + }); + first.end("first"); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const port = (server.address() as AddressInfo).port; + using client = pausedClient(port, pipelined(), { collect: false }); + expect(await Promise.race([wrote.promise, client.done])).toBe(false); + await ping(port); + // The client has not read anything yet. Winsock can take it all in one send(). + if (process.platform !== "win32") expect(drains).toEqual([]); + client.resume(); + const { ended } = await client.done; + expect({ drains, ended }).toEqual({ drains: [{ writableNeedDrain: false, writableLength: 0 }], ended: true }); + }); + + // The small write() of the 'socket' listener counts in writableLength + // until the end of the turn, so a handler that waits for an empty buffer + // must not get the 'drain' of the buffered writes before that. + it("the 'drain' of the writes that a response buffered while it waited comes with writableLength 0, also after a small write() in its 'socket' listener", async () => { + const drains: object[] = []; + let first: http.ServerResponse; + let total = 0; + let returned = true; + await using server = createServer(false, (req, res) => { + if (req.url === "/first") return void (first = writeFirst(res)); + // As many chunks as reach the high water mark: the last write() returns false. + const count = Math.ceil(res.writableHighWaterMark / CHUNK16.length); + total = count * CHUNK16.length + 1 + 4; + res.setHeader("Content-Length", total); + for (let i = 0; i < count; i++) returned = res.write(CHUNK16); + res.on("socket", () => res.write("x")); + res.on("drain", () => { + drains.push({ writableNeedDrain: res.writableNeedDrain, writableLength: res.writableLength }); + res.end("tail"); + }); + first.end(); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, pipelined()); + client.resume(); + const { bytes, ended } = await client.done; + expect({ returned, drains, length: secondResponse(bytes).body.length, ended }).toEqual({ + returned: false, + drains: [{ writableNeedDrain: false, writableLength: 0 }], + length: total, + ended: true, + }); + }); + + // Node reads `destroyed` and `finished` in the getter, and `finished` before it emits 'drain': + // https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L881-L886 + it.each(["end", "destroy"] as const)( + "after %s(), writableNeedDrain reads false and no 'drain' comes for the write() that returned false", + async how => { + const seen = Promise.withResolvers(); + await using server = createServer(false, (req, res) => { + let drains = 0; + res.on("drain", () => drains++); + let returned = true; + // As many chunks as reach the high water mark: the last write() returns false. + for (let size = 0; size < res.writableHighWaterMark; size += CHUNK16.length) returned = res.write(CHUNK16); + const before = res.writableNeedDrain; + res[how](); + const after = res.writableNeedDrain; + res.on("close", () => setImmediate(() => seen.resolve({ returned, before, after, drains }))); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, single, { collect: false }); + client.resume(); + expect(await seen.promise).toEqual({ returned: false, before: true, after: false, drains: 0 }); + }, + ); + + // write() records every false that it returns. For a response without a + // body, Node discards the write before it looks at the socket. No socket + // is left to drain, so no 'drain' follows. + it.each(["GET", "HEAD"])( + "a write() to the response of a %s request whose socket is destroyed records what it returned, and gets no 'drain'", + async method => { + const seen = Promise.withResolvers<{ returned: boolean; writableNeedDrain: boolean; drains: number }>(); + await using server = createServer(false, (req, res) => { + let drains = 0; + res.on("drain", () => drains++); + res.write("a"); + req.socket.destroy(); + const returned = res.write("x"); + const writableNeedDrain = res.writableNeedDrain; + res.on("close", () => setImmediate(() => seen.resolve({ returned, writableNeedDrain, drains }))); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient( + (server.address() as AddressInfo).port, + `${method} / HTTP/1.1\r\nHost: localhost\r\n\r\n`, + ); + const { returned, writableNeedDrain, drains } = await seen.promise; + expect({ returned, writableNeedDrain, drains }).toEqual({ + // false for the GET request on Node.js 26.3, true from 26.9 on. + returned: method === "HEAD" ? true : returned, + writableNeedDrain: !returned, + drains: 0, + }); + }, + ); + }); + // The keep-alive timer belongs to the idle time after a response. Started // at end(), it destroys the connection of a reader that takes longer than // keepAliveTimeout (5 seconds by default) to fetch the body. Not on @@ -1878,17 +2563,19 @@ describe("backpressure", () => { earlierResponsePending: boolean; returned: boolean[]; }>(); + let earlier: http.ServerResponse; await using server = http.createServer((req, res) => { if (req.url !== "/ignored") { + earlier = res; res.end(payload); return; } try { res.statusCode = req.method === "HEAD" ? 200 : 204; res.on("drain", () => called.push("drain")); - // Bun counts the unsent bytes of the earlier response on this - // response, Node counts them on the socket. - const earlierResponsePending = res.writableLength + req.socket.writableLength > 0; + // The unsent bytes belong to the earlier response, which still has the + // socket. A response that waits behind it counts none of them. + const earlierResponsePending = earlier.writableLength > 0; const returned = [ res.write("x", () => called.push("x")), res.write("", () => called.push("empty string")), @@ -2286,6 +2973,244 @@ describe("backpressure", () => { } }); + // A response that waited behind another one is given the socket, then the + // writes that it buffered go out, and then the 'drain' that one of them is + // owed comes. A 'drain' ahead of those writes lets its listener write, or + // end, in front of them. + it("a write() that returned false while the response waited behind another one gets its 'drain' after its bytes", async () => { + const events: string[] = []; + const chunk = Buffer.alloc(16 * 1024, "b"); + let first: http.ServerResponse; + let total = 0; + const connection = await pair((req, res) => { + if (req.url === "/first") return void (first = res); + // As many chunks as reach the high water mark: the last write() returns false. + const count = Math.ceil(res.writableHighWaterMark / chunk.length); + total = count * chunk.length; + res.setHeader("Content-Length", total + 4); + let returned = true; + for (let i = 0; i < count; i++) returned = res.write(chunk); + events.push(`waits: write() returned ${returned}, writableNeedDrain ${res.writableNeedDrain}`); + res.on("socket", () => events.push(`'socket': writableNeedDrain ${res.writableNeedDrain}`)); + res.on("drain", () => { + events.push(`'drain': writableNeedDrain ${res.writableNeedDrain}, writableLength ${res.writableLength}`); + res.end("tail"); + }); + first.end("first"); + }); + const client = await connection.connect(); + try { + const received: Buffer[] = []; + client.on("data", (data: Buffer) => received.push(data)); + client.write( + "GET /first HTTP/1.1\r\nHost: localhost\r\n\r\nGET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ); + client.resume(); + await once(client, "end"); + const bytes = Buffer.concat(received); + const body = bytes.subarray(bytes.lastIndexOf("\r\n\r\n") + 4); + expect({ + events, + length: body.length, + inOrder: body.equals(Buffer.concat([Buffer.alloc(total, "b"), Buffer.from("tail")])), + }).toEqual({ + events: [ + "waits: write() returned false, writableNeedDrain true", + "'socket': writableNeedDrain true", + "'drain': writableNeedDrain false, writableLength 0", + ], + length: total + 4, + inOrder: true, + }); + } finally { + client.destroy(); + } + }); + + const pipelinedPair = + "GET /first HTTP/1.1\r\nHost: localhost\r\n\r\nGET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"; + const singlePair = "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"; + + // A duplexPair() connection whose sides take `highWaterMark` bytes. The client reads nothing until resume(). + function smallPair(listener: http.RequestListener, highWaterMark: number) { + const [clientSide, serverSide] = duplexPair({ highWaterMark }); + clientSide.pause(); + http.createServer(listener).emit("connection", serverSide); + return clientSide; + } + + // Four 16 KB writes reach the high water mark, and the socket then takes + // only part of a large one, all in one turn. That is one backlog. + it("one backlog gets one 'drain', after the socket has taken all of it", async () => { + const small = Buffer.alloc(16 * 1024, "b"); + const drains: object[] = []; + const wrote = Promise.withResolvers(); + const connection = await pair((req, res) => { + res.setHeader("Content-Length", 4 * small.length + CHUNK.length); + res.on("drain", () => { + drains.push({ writableNeedDrain: res.writableNeedDrain, writableLength: res.writableLength }); + res.end(); + }); + for (let i = 0; i < 4; i++) res.write(small); + wrote.resolve(res.write(CHUNK)); + }); + const client = await connection.connect(); + try { + client.write(singlePair); + expect(await wrote.promise).toBe(false); + await connection.turn(); + // The client has not read anything yet. + expect(drains).toEqual([]); + client.resume(); + await once(client, "end"); + expect(drains).toEqual([{ writableNeedDrain: false, writableLength: 0 }]); + } finally { + client.destroy(); + } + }); + + it("a response that ended while it waited is still ended in its 'socket' listener, and gets one 'prefinish'", async () => { + const events: string[] = []; + let first: http.ServerResponse; + const connection = await pair((req, res) => { + if (req.url === "/first") return void (first = res); + res.on("prefinish", () => events.push("prefinish")); + res.on("socket", () => events.push(`socket: finished ${res.finished}, writableEnded ${res.writableEnded}`)); + res.end("second"); + first.end("first"); + }); + const client = await connection.connect(); + try { + const received: Buffer[] = []; + client.on("data", (data: Buffer) => received.push(data)); + client.write(pipelinedPair); + client.resume(); + await once(client, "end"); + expect({ events, tail: Buffer.concat(received).subarray(-6).toString() }).toEqual({ + events: ["socket: finished true, writableEnded true", "prefinish"], + tail: "second", + }); + } finally { + client.destroy(); + } + }); + + it("a write() to a response whose socket is destroyed records what it returned", async () => { + const seen = Promise.withResolvers<{ returned: boolean; writableNeedDrain: boolean }>(); + const connection = await pair((req, res) => { + req.socket.destroy(); + seen.resolve({ returned: res.write("x"), writableNeedDrain: res.writableNeedDrain }); + }); + const client = await connection.connect(); + try { + client.write(singlePair); + const { returned, writableNeedDrain } = await seen.promise; + // write() returns false on Node.js 26.3, and true from 26.9 on. + expect(writableNeedDrain).toBe(!returned); + } finally { + client.destroy(); + } + }); + + it("a response that waits counts its head from flushHeaders() on, and none of the bytes of the response ahead", async () => { + const counted = Promise.withResolvers(); + let first: http.ServerResponse; + const connection = await pair((req, res) => { + if (req.url === "/first") { + first = res; + res.setHeader("Content-Length", CHUNK.length); + res.write(CHUNK); + return; + } + const fresh = res.writableLength; + res.flushHeaders(); + const head = res.writableLength; + counted.resolve({ fresh, headIsCounted: head > 0 && head < 1024, writableNeedDrain: res.writableNeedDrain }); + res.end(); + first.end(); + }); + const client = await connection.connect(); + try { + client.write(pipelinedPair); + expect(await counted.promise).toEqual({ fresh: 0, headIsCounted: true, writableNeedDrain: false }); + } finally { + client.destroy(); + } + }); + + // The socket takes 16 KB. The buffered writes of a response that waited back + // it up at the turn of the response. Whatever mark the response used while + // it waited, writableNeedDrain then says what its own write() calls + // returned, and nothing else. + describe("after the buffered writes of a response that waited went to a socket that backs up", () => { + const chunk = Buffer.alloc(20 * 1024, "b"); + const turn = () => new Promise(resolve => setImmediate(resolve)); + + it("writableNeedDrain is what write() returned, and no 'drain' comes while the socket holds them", async () => { + const seen = Promise.withResolvers(); + let first: http.ServerResponse; + const client = smallPair(async (req, res) => { + if (req.url === "/first") return void (first = res); + res.setHeader("Content-Length", 2 * chunk.length); + let drains = 0; + res.on("drain", () => drains++); + const returned = [res.write(chunk), res.write(chunk)]; + first.end("first"); + await once(res, "socket"); + // One turn after 'socket', the buffered writes have gone to the socket. + await turn(); + seen.resolve({ + asReturned: res.writableNeedDrain === returned.includes(false), + drains, + backedUp: res.socket!.writableNeedDrain, + }); + res.end(); + }, 16 * 1024); + try { + client.write(pipelinedPair); + // The client has not read anything. + expect(await seen.promise).toEqual({ asReturned: true, drains: 0, backedUp: true }); + } finally { + client.destroy(); + } + }); + + it("a chunk written through Writable.toWeb(res) is delivered", async () => { + let first: http.ServerResponse; + const client = smallPair(async (req, res) => { + if (req.url === "/first") return void (first = res); + res.setHeader("Content-Length", 2 * chunk.length + 4); + const returned = [res.write(chunk), res.write(chunk)]; + first.end("first"); + await once(res, "socket"); + await turn(); + // Writable.toWeb() does not write a chunk while writableNeedDrain is true: after a write() that returned false, wait. + if (returned.includes(false)) await once(res, "drain"); + const writer = Writable.toWeb(res as unknown as Writable).getWriter(); + await writer.write("tail"); + await writer.close(); + }, 16 * 1024); + try { + const received: Buffer[] = []; + client.write(pipelinedPair); + // The buffered writes are in the socket before the client reads. + await turn(); + await turn(); + client.on("data", (data: Buffer) => received.push(data)); + client.resume(); + await once(client, "end"); + const bytes = Buffer.concat(received); + const body = bytes.subarray(bytes.lastIndexOf("\r\n\r\n") + 4); + expect({ length: body.length, tail: body.subarray(-4).toString() }).toEqual({ + length: 2 * chunk.length + 4, + tail: "tail", + }); + } finally { + client.destroy(); + } + }); + }); + // Node discards a write() to a response without a body and answers true. // Nothing reaches the socket, so its backlog is not this write's concern. it("a write() to a HEAD response does not wait for a socket that still holds the previous body", async () => { @@ -2316,6 +3241,152 @@ describe("backpressure", () => { }); }); + // A response is written by a script of steps. Each step is valid in Node in + // any state, and continues where Node says it can: after write(), in a write + // callback, at 'drain', or when writableNeedDrain or writableLength allow it. + // The script runs in a response that has the socket, in one that waits + // behind megabytes of an earlier response, and in one that waits behind a + // response that is still open. + describe("every byte of a response arrives, whatever backpressure pattern writes it", () => { + const BIG = 6 * 1024 * 1024; + type Step = [name: string, size: number]; + const parse = (script: string): Step[] => + script.split(" ").map(step => { + const [name, size] = step.split(":"); + return [name, Number(size ?? 0)]; + }); + const bodyLength = (steps: Step[]) => steps.reduce((sum, [, size]) => sum + size, 0); + + function run( + res: http.ServerResponse, + steps: Step[], + byte: string, + previous: http.ServerResponse | undefined, + fail: (error: unknown) => void, + ) { + let i = 0; + const whenEmpty = (then: () => void) => + res.writableLength > 0 ? res.once("drain", () => whenEmpty(then)) : then(); + const whenNoDrainIsOwed = (then: () => void) => + res.writableNeedDrain ? res.once("drain", () => whenNoDrainIsOwed(then)) : then(); + const next = () => { + try { + while (i < steps.length) { + const [name, size] = steps[i++]; + const chunk = Buffer.alloc(size, byte); + if (name === "write") res.write(chunk); + // Node emits 'drain' with writableLength 0, so this wait ends after a write() that returned false. + else if (name === "writeThenWaitUntilEmpty") { + if (!res.write(chunk)) return void whenEmpty(next); + } else if (name === "writeThenContinueInCallback") return void res.write(chunk, next); + else if (name === "waitWhileNeedDrain") return void whenNoDrainIsOwed(next); + else if (name === "immediate") return void setImmediate(next); + else if (name === "waitForPreviousFinish") { + if (previous && !previous.writableFinished) return void previous.once("finish", next); + } else if (name === "end") res.end(chunk); + else if (name === "pipe") Readable.from([chunk]).pipe(res); + // Writable.toWeb() does not write a chunk while writableNeedDrain is true, so the script waits first. + else if (name === "toWeb") { + return void whenNoDrainIsOwed(() => { + const writer = Writable.toWeb(res as unknown as Writable).getWriter(); + writer + .write(chunk) + .then(() => writer.close()) + .catch(fail); + }); + } else throw new Error(`unknown step ${name}`); + } + } catch (error) { + fail(error); + } + }; + next(); + } + + // The requests arrive in one write, one script each. The client reads once every handler has been called. + async function deliver(scripts: string[], options: http.ServerOptions) { + const all = scripts.map(parse); + const lengths = all.map(bodyLength); + const responses: http.ServerResponse[] = []; + const errors: string[] = []; + const called = Promise.withResolvers(); + await using server = http.createServer(options, (req, res) => { + const i = Number(req.url!.slice(1)); + responses[i] = res; + res.setHeader("Content-Length", lengths[i]); + res.on("error", error => errors.push(String((error as NodeJS.ErrnoException).code ?? error))); + run(res, all[i], "abc"[i], responses[i - 1], error => errors.push(String(error))); + if (i === all.length - 1) called.resolve(); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const socket = net.connect((server.address() as AddressInfo).port, "127.0.0.1"); + try { + socket.pause(); + const received: Buffer[] = []; + socket.on("data", (chunk: Buffer) => received.push(chunk)); + const closed = once(socket, "close"); + await once(socket, "connect"); + socket.write( + all + .map( + (_, i) => `GET /${i} HTTP/1.1\r\nHost: x\r\n${i === all.length - 1 ? "Connection: close\r\n" : ""}\r\n`, + ) + .join(""), + ); + await called.promise; + await new Promise(resolve => setImmediate(resolve)); + socket.resume(); + await closed; + // Each body in order, as the run of its own byte. + const bytes = Buffer.concat(received); + const bodies: string[] = []; + let at = 0; + for (let i = 0; i < all.length; i++) { + const head = bytes.indexOf("\r\n\r\n", at); + if (head < 0) break; + const body = bytes.subarray(head + 4, head + 4 + lengths[i]); + const whole = body.equals(Buffer.alloc(lengths[i], "abc"[i])); + bodies.push(whole ? `${"abc"[i]} x ${body.length}` : `damaged (${body.length} bytes)`); + at = head + 4 + lengths[i]; + } + return { bodies, trailing: bytes.length - at, errors }; + } finally { + socket.destroy(); + } + } + + const aheads: [string, string[]][] = [ + ["that has the socket", []], + ["behind a response with megabytes unsent", [`end:${BIG}`]], + ["behind a response that is still open", ["immediate end:10"]], + ]; + const marks: [string, http.ServerOptions][] = [ + ["the default highWaterMark", {}], + ["highWaterMark 0", { highWaterMark: 0 }], + ]; + const scripts = [ + "pipe:16384", + "waitWhileNeedDrain end:16384", + "toWeb:16384", + `write:10 writeThenWaitUntilEmpty:${BIG} end`, + `writeThenContinueInCallback:65536 writeThenWaitUntilEmpty:${BIG} end`, + `write:65536 waitForPreviousFinish writeThenWaitUntilEmpty:${BIG} end`, + ]; + const cells: [string, string, string, string[], http.ServerOptions][] = []; + for (const [ahead, before] of aheads) + for (const [mark, options] of marks) + for (const script of scripts) cells.push([script, ahead, mark, [...before, script], options]); + + it.each(cells)("%s, in a response %s, with %s", async (_script, _ahead, _mark, all, options) => { + const lengths = all.map(script => bodyLength(parse(script))); + expect(await deliver(all, options)).toEqual({ + bodies: lengths.map((length, i) => `${"abc"[i]} x ${length}`), + trailing: 0, + errors: [], + }); + }); + }); + // 16 KB is the size of the uWS cork buffer. A chunk that does not fit goes out with its framing in one vectored write. describe.each(["http", "https"] as const)("chunk framing on the wire (%s)", protocol => { const keysDir = path.join(import.meta.dirname, "..", "test", "fixtures", "keys"); diff --git a/test/js/node/http/node-http-syscall-fault.test.ts b/test/js/node/http/node-http-syscall-fault.test.ts index 9138d6f8f078..cb45bbf07890 100644 --- a/test/js/node/http/node-http-syscall-fault.test.ts +++ b/test/js/node/http/node-http-syscall-fault.test.ts @@ -311,6 +311,96 @@ describe.skipIf(skip)("node:http pipelining under stalled sends", () => { }); }); +// Every send takes 1000 bytes, so a write backs the socket up at once, with no megabytes and no slow client. +describe.skipIf(skip)("node:http 'drain' under short sends", () => { + const CHUNK16 = Buffer.alloc(16 * 1024, "b"); + const shortSends = () => { + for (const syscall of ["send", "writev"] as const) fault.set({ syscall, action: "short", bytes: 1000, repeat: -1 }); + }; + async function receive(server: http.Server, request: string) { + await once(server.listen(0, "127.0.0.1"), "listening"); + const socket = net.connect((server.address() as net.AddressInfo).port, "127.0.0.1"); + const chunks: Buffer[] = []; + socket.on("data", (chunk: Buffer) => chunks.push(chunk)); + socket.on("connect", () => socket.write(request)); + await once(socket, "close"); + const bytes = Buffer.concat(chunks); + return bytes.subarray(bytes.lastIndexOf("\r\n\r\n") + 4); + } + + // The chunk and the head fit in the cork buffer, so the write is accepted + // whole. The socket then takes only part of it. + test("a first write(chunk, callback) that the socket takes only part of returns false and gets one 'drain'", async () => { + const events: string[] = []; + await using server = http.createServer((req, res) => { + res.setHeader("Content-Length", CHUNK16.length + 4); + // One turn later, the write is outside the cork of the request dispatch. + setImmediate(() => { + shortSends(); + const returned = res.write(CHUNK16, () => events.push("write callback")); + events.push(`write(chunk, callback) returned ${returned}: writableNeedDrain ${res.writableNeedDrain}`); + res.on("drain", () => { + events.push(`'drain': writableNeedDrain ${res.writableNeedDrain}, writableLength ${res.writableLength}`); + res.end("tail"); + }); + }); + }); + const body = await receive(server, "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"); + expect({ events, length: body.length, tail: body.subarray(-4).toString() }).toEqual({ + events: [ + "write(chunk, callback) returned false: writableNeedDrain true", + "'drain': writableNeedDrain false, writableLength 0", + "write callback", + ], + length: CHUNK16.length + 4, + tail: "tail", + }); + }); + + // The buffered writes of a response that waited back the socket up at its + // turn. The 'drain' that one of them is owed comes from the drain callback + // of the handle, after the bytes. + test("a write() that returned false while the response waited gets its 'drain' after the socket took the buffered writes", async () => { + const events: string[] = []; + let first: http.ServerResponse; + let total = 0; + await using server = http.createServer((req, res) => { + if (req.url === "/first") return void (first = res); + // As many chunks as reach the high water mark: the last write() returns false. + const count = Math.ceil(res.writableHighWaterMark / CHUNK16.length); + total = count * CHUNK16.length; + res.setHeader("Content-Length", total + 4); + let returned = true; + for (let i = 0; i < count; i++) returned = res.write(CHUNK16); + events.push(`waits: write() returned ${returned}, writableNeedDrain ${res.writableNeedDrain}`); + res.on("socket", () => events.push(`'socket': writableNeedDrain ${res.writableNeedDrain}`)); + res.on("drain", () => { + events.push(`'drain': writableNeedDrain ${res.writableNeedDrain}, writableLength ${res.writableLength}`); + res.end("tail"); + }); + shortSends(); + first.end("first"); + }); + const body = await receive( + server, + "GET /first HTTP/1.1\r\nHost: localhost\r\n\r\nGET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ); + expect({ + events, + length: body.length, + inOrder: body.equals(Buffer.concat([Buffer.alloc(total, "b"), Buffer.from("tail")])), + }).toEqual({ + events: [ + "waits: write() returned false, writableNeedDrain true", + "'socket': writableNeedDrain true", + "'drain': writableNeedDrain false, writableLength 0", + ], + length: total + 4, + inOrder: true, + }); + }); +}); + describe.skipIf(skip)("node:http seeded backpressure fuzz", () => { const seed = Number(process.env.BUN_SOCKET_FUZZ_SEED ?? 0x5e1d) >>> 0 || 1; function makePrng(s: number) { diff --git a/test/regression/issue/19111.test.ts b/test/regression/issue/19111.test.ts index 7406f1b19f09..08725f557e82 100644 --- a/test/regression/issue/19111.test.ts +++ b/test/regression/issue/19111.test.ts @@ -2,7 +2,7 @@ // stream.Readable's `readable` event not firing in Bun 1.2.6+ import assert from "node:assert"; import { IncomingMessage, ServerResponse } from "node:http"; -import { PassThrough, Readable } from "node:stream"; +import { PassThrough, Readable, Writable } from "node:stream"; import { test } from "node:test"; // Helper to create mock IncomingMessage @@ -23,6 +23,38 @@ test("Standalone ServerResponse.writableNeedDrain is false", () => { assert.strictEqual(res.writableNeedDrain, false); }); +// The same stall, when the socket has a high water mark of 0: nothing buffered had reached the mark. +test("Standalone ServerResponse.writableNeedDrain is false on a socket with highWaterMark 0", async () => { + const chunks: Buffer[] = []; + const socket = new Writable({ + highWaterMark: 0, + write(chunk, _encoding, callback) { + chunks.push(chunk); + callback(); + }, + }); + const res = new ServerResponse(createMockIncomingMessage("/zero-mark")); + res.assignSocket(socket as never); + assert.strictEqual(res.writableNeedDrain, false); + + const finished = new Promise(resolve => res.on("finish", resolve)); + Readable.from([Buffer.from("piped")]).pipe(res); + await finished; + assert.match(Buffer.concat(chunks).toString(), /\r\n\r\n5\r\npiped\r\n0\r\n\r\n$/); +}); + +// Node.js reads true here, and a pipe() into the response then waits for a 'drain' that +// never comes: nothing forwards the 'drain' of the socket to a response that no server made. +test("Standalone ServerResponse.writableNeedDrain stays false after a write() that its socket refused", () => { + // A socket that takes a write and never completes it. + const socket = new Writable({ highWaterMark: 16, write() {} }); + const res = new ServerResponse(createMockIncomingMessage("/refused")); + res.assignSocket(socket as never); + + assert.strictEqual(res.write(Buffer.alloc(1024, "a")), false); + assert.strictEqual(res.writableNeedDrain, false); +}); + // Helper function for connect-to-web pattern function createServerResponse(incomingMessage: IncomingMessage) { const res = new ServerResponse(incomingMessage); From 8f583cb64f757d957888fc701e92dd7b5de3b6eb Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 9 Oct 2026 19:53:57 +0000 Subject: [PATCH 2/3] node:http: arm the drain callback when the socket took only part of a replayed write At the turn of a queued response, a kept write that the cork buffer took whole reports no backpressure. When the socket then takes only part of it, the owed 'drain' now waits for the drain callback of the handle. The destroyed-socket tests pin what write() returns by the Node.js version. --- src/js/node/_http_server.ts | 2 + .../node/http/node-http-backpressure.test.ts | 18 ++++----- .../node/http/node-http-syscall-fault.test.ts | 38 +++++++++++++++++-- 3 files changed, 45 insertions(+), 13 deletions(-) diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 6c409185fb84..082a4c3a85ae 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -3018,6 +3018,8 @@ function advanceResponsePipeline(server, socket) { if (res[kNeedDrain] && !queued.ended && !hitBackpressure) { // With nothing left to send it comes here, like in _flush(): https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L1194-L1197 if (res.writableLength === 0) emitDrainIfOwed(res); + // The cork buffer took a replayed write whole and the socket took only part of it: arm the drain callback. + else if (handle.bufferedAmount > 0) handle.onwritable = allowWritesToContinue.bind(res); else process.nextTick(emitDrainIfOwed, res); } } diff --git a/test/js/node/http/node-http-backpressure.test.ts b/test/js/node/http/node-http-backpressure.test.ts index b4bbe10dba25..3412bb5dd994 100644 --- a/test/js/node/http/node-http-backpressure.test.ts +++ b/test/js/node/http/node-http-backpressure.test.ts @@ -1622,13 +1622,10 @@ describe("backpressure", () => { (server.address() as AddressInfo).port, `${method} / HTTP/1.1\r\nHost: localhost\r\n\r\n`, ); - const { returned, writableNeedDrain, drains } = await seen.promise; - expect({ returned, writableNeedDrain, drains }).toEqual({ - // false for the GET request on Node.js 26.3, true from 26.9 on. - returned: method === "HEAD" ? true : returned, - writableNeedDrain: !returned, - drains: 0, - }); + // Node.js returns false for the GET request up to 26.8, and true from 26.9 on. + const [major, minor] = process.versions.node.split(".").map(Number); + const returned = method === "HEAD" || major > 26 || (major === 26 && minor >= 9); + expect(await seen.promise).toEqual({ returned, writableNeedDrain: !returned, drains: 0 }); }, ); }); @@ -3104,9 +3101,10 @@ describe("backpressure", () => { const client = await connection.connect(); try { client.write(singlePair); - const { returned, writableNeedDrain } = await seen.promise; - // write() returns false on Node.js 26.3, and true from 26.9 on. - expect(writableNeedDrain).toBe(!returned); + // Node.js returns false up to 26.8, and true from 26.9 on. + const [major, minor] = process.versions.node.split(".").map(Number); + const returned = major > 26 || (major === 26 && minor >= 9); + expect(await seen.promise).toEqual({ returned, writableNeedDrain: !returned }); } finally { client.destroy(); } diff --git a/test/js/node/http/node-http-syscall-fault.test.ts b/test/js/node/http/node-http-syscall-fault.test.ts index cb45bbf07890..948a276d09e8 100644 --- a/test/js/node/http/node-http-syscall-fault.test.ts +++ b/test/js/node/http/node-http-syscall-fault.test.ts @@ -311,11 +311,11 @@ describe.skipIf(skip)("node:http pipelining under stalled sends", () => { }); }); -// Every send takes 1000 bytes, so a write backs the socket up at once, with no megabytes and no slow client. +// Every send takes a few bytes, so a write backs the socket up at once, with no megabytes and no slow client. describe.skipIf(skip)("node:http 'drain' under short sends", () => { const CHUNK16 = Buffer.alloc(16 * 1024, "b"); - const shortSends = () => { - for (const syscall of ["send", "writev"] as const) fault.set({ syscall, action: "short", bytes: 1000, repeat: -1 }); + const shortSends = (bytes = 1000) => { + for (const syscall of ["send", "writev"] as const) fault.set({ syscall, action: "short", bytes, repeat: -1 }); }; async function receive(server: http.Server, request: string) { await once(server.listen(0, "127.0.0.1"), "listening"); @@ -399,6 +399,38 @@ describe.skipIf(skip)("node:http 'drain' under short sends", () => { inOrder: true, }); }); + + // With highWaterMark 0 a small write() returns false while the response + // waits. At its turn that one write fits the cork buffer, so the handle + // accepts it whole and reports no backpressure. The socket then takes only + // part of it. + test("a small write() that returned false while the response waited gets its 'drain' after the socket took it", async () => { + const events: string[] = []; + let first: http.ServerResponse; + await using server = http.createServer({ highWaterMark: 0 }, (req, res) => { + if (req.url === "/first") return void (first = res); + res.setHeader("Content-Length", 100 + 4); + const returned = res.write(Buffer.alloc(100, "b")); + events.push(`waits: write() returned ${returned}, writableNeedDrain ${res.writableNeedDrain}`); + res.on("drain", () => { + events.push(`'drain': writableNeedDrain ${res.writableNeedDrain}, writableLength ${res.writableLength}`); + res.end("tail"); + }); + shortSends(50); + first.end("first"); + }); + const body = await receive( + server, + "GET /first HTTP/1.1\r\nHost: localhost\r\n\r\nGET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ); + expect({ events, body: body.toString() }).toEqual({ + events: [ + "waits: write() returned false, writableNeedDrain true", + "'drain': writableNeedDrain false, writableLength 0", + ], + body: Buffer.alloc(100, "b") + "tail", + }); + }); }); describe.skipIf(skip)("node:http seeded backpressure fuzz", () => { From fdbc86dc9179b3bf43c6ba6dcb31565125540a1f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 16 Oct 2026 10:56:10 +0000 Subject: [PATCH 3/3] node:http: res._send() drops its callback when the socket is gone The native write takes nothing for a closed socket and returns 0, so the callback ran as if the bytes were sent. Node's _writeRaw() drops it. The destroyed-socket tests state the value of Bun, and apply the Node.js version switch only under Node.js. --- src/js/node/_http_server.ts | 3 ++- .../node/http/node-http-backpressure.test.ts | 25 ++++++++++++++++--- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 082a4c3a85ae..f258aadf00dc 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -3980,7 +3980,8 @@ ServerResponse.prototype._send = function (data, encoding, callback, _byteLength } if ($isCallable(callback)) { if (result < 0) this[kPendingCallbacks].push(callback); - else process.nextTick(callback); + // Socket already gone: like Node's _writeRaw(), the write is dropped and its callback never runs: https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L400-L406 + else if (!(handle.flags & NodeHTTPResponseFlags.closed_or_completed)) process.nextTick(callback); } }; diff --git a/test/js/node/http/node-http-backpressure.test.ts b/test/js/node/http/node-http-backpressure.test.ts index 3412bb5dd994..1411bbbdd14f 100644 --- a/test/js/node/http/node-http-backpressure.test.ts +++ b/test/js/node/http/node-http-backpressure.test.ts @@ -1480,6 +1480,22 @@ describe("backpressure", () => { }, ); + // Node drops a raw write to a destroyed socket, and its callback with it: + // https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L400-L406 + it("the callback of res._send() to a response whose socket is destroyed never runs", async () => { + const seen = Promise.withResolvers(); + await using server = createServer(false, (req, res) => { + let called = 0; + req.socket.destroy(); + // @ts-ignore _send() is not in the types + res._send("x", "latin1", () => called++); + res.on("close", () => setImmediate(() => seen.resolve(called))); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + using client = pausedClient((server.address() as AddressInfo).port, single); + expect(await seen.promise).toBe(0); + }); + // The constructor of a response class runs before the server knows that // the response has to wait. it("a response class whose constructor runs behind an unfinished response counts none of its bytes", async () => { @@ -1622,9 +1638,10 @@ describe("backpressure", () => { (server.address() as AddressInfo).port, `${method} / HTTP/1.1\r\nHost: localhost\r\n\r\n`, ); - // Node.js returns false for the GET request up to 26.8, and true from 26.9 on. + // Node.js returns false for the GET request up to 26.8, and true from 26.9 on. Bun returns false. const [major, minor] = process.versions.node.split(".").map(Number); - const returned = method === "HEAD" || major > 26 || (major === 26 && minor >= 9); + const nodeReturnsTrue = !process.versions.bun && (major > 26 || (major === 26 && minor >= 9)); + const returned = method === "HEAD" || nodeReturnsTrue; expect(await seen.promise).toEqual({ returned, writableNeedDrain: !returned, drains: 0 }); }, ); @@ -3101,9 +3118,9 @@ describe("backpressure", () => { const client = await connection.connect(); try { client.write(singlePair); - // Node.js returns false up to 26.8, and true from 26.9 on. + // Node.js returns false up to 26.8, and true from 26.9 on. Bun returns false. const [major, minor] = process.versions.node.split(".").map(Number); - const returned = major > 26 || (major === 26 && minor >= 9); + const returned = !process.versions.bun && (major > 26 || (major === 26 && minor >= 9)); expect(await seen.promise).toEqual({ returned, writableNeedDrain: !returned }); } finally { client.destroy();