From 81a3059dcc5ebf833d54c85d7db12e1a4eb342aa Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 23 Oct 2026 23:55:04 -0700 Subject: [PATCH 01/10] fix(node:http): retain corked responses in the socket write queue --- packages/bun-uws/src/HttpContext.h | 2 +- src/js/internal/streams/utils.ts | 2 + src/js/internal/streams/writable.ts | 14 + src/js/node/_http_outgoing.ts | 1 + src/js/node/_http_server.ts | 321 +++++++++- .../bindings/node/JSNodeHTTPServerSocket.cpp | 5 + .../bindings/node/JSNodeHTTPServerSocket.h | 2 +- .../node/JSNodeHTTPServerSocketPrototype.cpp | 4 +- .../node/http/node-http-nested-cork.test.ts | 553 +++++++++++++++++- test/js/node/http/node-http-with-ws.test.ts | 22 +- 10 files changed, 878 insertions(+), 48 deletions(-) diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index 14c58b249928..e23defebabdf 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -927,7 +927,7 @@ struct HttpContext { auto *httpContextData = getSocketContextDataS(s); - if (httpResponseData->isConnectRequest && httpResponseData->socketData && httpContextData->onSocketDrain) { + if ((IsNodeHttp || httpResponseData->isConnectRequest) && httpResponseData->socketData && httpContextData->onSocketDrain) { httpContextData->onSocketDrain(httpResponseData->socketData, SSL, (struct us_socket_t *) s); } /* Ask the developer to write data and return success (true) or failure (false), OR skip sending anything and return success (true). */ diff --git a/src/js/internal/streams/utils.ts b/src/js/internal/streams/utils.ts index 8cf94a2ffe5f..1e3343766de7 100644 --- a/src/js/internal/streams/utils.ts +++ b/src/js/internal/streams/utils.ts @@ -32,6 +32,7 @@ const kIsReadable = Symbol.for("nodejs.stream.readable"); const kIsWritable = Symbol.for("nodejs.stream.writable"); const kIsDisturbed = Symbol.for("nodejs.stream.disturbed"); +const kWriteOperation = Symbol("kWriteOperation"); const kIsClosedPromise = Symbol.for("nodejs.webstream.isClosedPromise"); const kControllerErrorFunction = Symbol.for("nodejs.webstream.controllerErrorFunction"); @@ -390,6 +391,7 @@ function isErrored(stream) { } export default { + kWriteOperation, isDestroyed, kIsDestroyed, isDisturbed, diff --git a/src/js/internal/streams/writable.ts b/src/js/internal/streams/writable.ts index 85f4ef730e84..d2e29d5ffbc7 100644 --- a/src/js/internal/streams/writable.ts +++ b/src/js/internal/streams/writable.ts @@ -9,6 +9,8 @@ const { Stream } = require("internal/streams/legacy"); const destroyImpl = require("internal/streams/destroy"); const eos = require("internal/streams/end-of-stream"); const { addAbortSignal } = require("internal/streams/add-abort-signal"); +const streamUtils = require("internal/streams/utils"); +const kWriteOperation: typeof streamUtils.kWriteOperation = streamUtils.kWriteOperation; const streamState = require("internal/streams/state"); const { WritableState, resetBuffer, isReadableState, isWritableState, flagsOf, addFlags, removeFlags, internals, nop } = streamState; @@ -230,6 +232,18 @@ function _write(stream: $Node.Writable, state: WritableState, chunk, encoding, c return writeOrBuffer(stream, state, chunk, encoding, cb, 1); } +// Internal byte-counted operations share the byte stream's queue and lifecycle. +Writable.prototype[kWriteOperation] = function (operation, callback) { + const state: WritableState = this._writableState; + const cb: WriteCallback = callback ?? nop; + if ((state[kState] & Flag.EndingOrDestroyed) !== 0) { + writeAfterEnd(this, state, cb); + return false; + } + state.pendingcb!++; + return writeOrBuffer(this, state, operation, kWriteOperation, cb, operation.length); +}; + const writeBytes = function _write(stream: $Node.Writable, state: WritableState, chunk, encoding, cb: WriteCallback) { if (!encoding) { encoding = (state[kState] & Flag.DefaultUTF8Encoding) !== 0 ? "utf8" : state[kDefaultEncodingValue]; diff --git a/src/js/node/_http_outgoing.ts b/src/js/node/_http_outgoing.ts index ef8cc79cc014..6d7f438160ab 100644 --- a/src/js/node/_http_outgoing.ts +++ b/src/js/node/_http_outgoing.ts @@ -1322,6 +1322,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, "_headerNames", { }); export default { + kCorked, kHighWaterMark, kUniqueHeaders, kOutHeaders, diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index c4513292b7c7..9a9e78e8309b 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -2,6 +2,7 @@ const EventEmitter: typeof import("node:events").EventEmitter = require("node:events"); const { Stream } = require("node:stream"); const { isIP } = require("internal/net/isIP"); +const { kWriteOperation } = require("internal/streams/utils"); const { _checkInvalidHeaderChar: checkInvalidHeaderChar, chunkExpression, @@ -72,6 +73,7 @@ const NumberIsNaN = Number.isNaN; const { IncomingMessage, kReqShouldKeepAlive } = require("node:_http_incoming"); const { OutgoingMessage, + kCorked, kErrored, kHighWaterMark, kSocket, @@ -1469,6 +1471,130 @@ function onReadableStreamEnd() {} function clearUpgradeIncoming(socket) { socket[kUpgradeIncoming] = undefined; } +const kResponseSocketWrite = Symbol("responseSocketWrite"); +const kResponseQueueInitialized = Symbol("responseQueueInitialized"); +const kResponseQueueNeedDrain = Symbol("responseQueueNeedDrain"); +const kResponseHeadersQueued = Symbol("responseHeadersQueued"); +const kResumeResponseSocketWrite = Symbol("resumeResponseSocketWrite"); +const kResponseQueuedBodyBytes = Symbol("responseQueuedBodyBytes"); + +function hasResponseSocketWork(socket) { + const state = socket._writableState; + return !!(state && (state.corked || state.length || state.writing)); +} + +function shouldQueueResponseWrite(response) { + const socket = response.socket; + return !!( + socket && + (hasResponseSocketWork(socket) || (response[kCorked] && willBeChunked(response))) && + NodeHTTPServerSocket && socket instanceof NodeHTTPServerSocket && + response[kHandle] + ); +} + +function onResponseSocketDrain() { + const response = this._httpMessage; + if (response) flushWriteAccountingNT(response); +} + +function freezeBufferedResponseHeaders(res, fromEnd) { + if (res[headerStateSymbol] !== NodeHTTPHeaderState.none) return; + if (fromEnd) res[kImplicitHeaderFromEnd] = true; + try { + res._implicitHeader(); + } finally { + if (fromEnd) res[kImplicitHeaderFromEnd] = false; + } +} + +function queueResponseSocketWrite(response, kind, chunk, encoding, callback) { + const headerState = response[headerStateSymbol]; + let strict; + let length = 0; + if (kind === "write" || (kind === "end" && chunk != null) || kind === "raw") { + if (chunk === null) throw $ERR_STREAM_NULL_VALUES(); + if (typeof chunk === "string") { + if (encoding && !Buffer.isEncoding(encoding)) throw $ERR_UNKNOWN_ENCODING(encoding); + length = Buffer.byteLength(chunk, encoding); + } else if (require("node:util/types").isUint8Array(chunk)) { + length = chunk.byteLength; + } else { + throw $ERR_INVALID_ARG_TYPE("chunk", ["string", "Buffer", "Uint8Array"], chunk); + } + } + let bodyLength = kind === "write" || kind === "end" ? length : 0; + if (kind !== "raw" && kind !== "continue") { + const implicitEnd = kind === "end" && response[headerStateSymbol] === NodeHTTPHeaderState.none; + if (implicitEnd && !response.hasHeader("content-length")) response._contentLength = length; + freezeBufferedResponseHeaders(response, kind === "end"); + if (!response[kResponseHeadersQueued] && response._contentLength == null && response.hasHeader("content-length")) { + const declared = +response.getHeader("content-length"); + if (!NumberIsNaN(declared)) response._contentLength = declared; + } + if (kind === "write" || kind === "end") strict = strictContentLength(response, headerState, kind === "end"); + if (!response._hasBody && length) { + if (response[kRejectNonStandardBodyWrites]) throw $ERR_HTTP_BODY_NOT_ALLOWED(); + chunk = undefined; + length = 0; + bodyLength = 0; + } + if (kind === "write" || kind === "end") { + const pending = response[kResponseQueuedBodyBytes] ?? 0; + if (strict !== undefined) { + const actual = response[kHandle].getBytesWritten() + pending + bodyLength; + if (kind === "end" ? actual !== strict : actual > strict) + throw $ERR_HTTP_CONTENT_LENGTH_MISMATCH(actual, strict); + } + response[kResponseQueuedBodyBytes] = pending + bodyLength; + } + const chunked = + willBeChunked(response) && + !(implicitEnd && !response.hasHeader("transfer-encoding") && !response._removedContLen); + if (length && chunked) length += length.toString(16).length + 4; + if (kind === "end" && chunked) length += 5 + (response._trailer?.length ?? 0); + if (response[headerStateSymbol] !== NodeHTTPHeaderState.sent && !response[kResponseHeadersQueued]) { + length += snapshotResponseSocketHeaders(response, kind === "end"); + } + } else if (kind === "continue") { + length = 25; + } + const socket = response.socket; + if (!socket[kResponseQueueInitialized]) { + socket[kResponseQueueInitialized] = true; + socket.on("drain", onResponseSocketDrain); + } + const responseCorked = kind === "write" && !!response[kCorked] && willBeChunked(response); + const result = socket[kWriteOperation]( + { response, kind, chunk, encoding, length, bodyLength, strict, responseCorked }, + callback, + ); + if (!result && kind === "write" && response._hasBody) response[kResponseQueueNeedDrain] = true; + return kind === "write" && !response._hasBody ? true : result; +} + +function executeResponseSocketWrite(operation, callback) { + const { response, kind, chunk, encoding } = operation; + if (kind === "write" || kind === "end") response[kResponseQueuedBodyBytes] -= operation.bodyLength; + if (kind === "write") { + ServerResponsePrototypeWrite.$call(response, chunk, encoding, callback, kResponseSocketWrite, operation); + } else if (kind === "end") { + ServerResponsePrototypeEnd.$call(response, chunk, encoding, callback, kResponseSocketWrite, operation); + } else { + const handle = response[kHandle]; + if (!handle || (handle.flags & NodeHTTPResponseFlags.closed_or_completed) !== 0) { + callback(response.errored ?? $ERR_STREAM_DESTROYED("write")); + return; + } + if (kind === "raw") handle.writeInformational(chunk, encoding); + else if (kind === "continue") handle.writeContinue(); + else ServerResponsePrototypeFlushHeaders.$call(response, kResponseSocketWrite); + if (handle.bufferedAmount > 0) { + response[kPendingCallbacks].push(callback); + handle.onwritable = allowWritesToContinue.bind(response); + } else callback(); + } +} // Node's UpgradeStream._read: resume the request only while its body is incomplete. function resumePausedUpgradeIncoming(socket) { @@ -1681,7 +1807,8 @@ function getNodeHTTPServerSocket() { // Node's connectionListener sets socket._paused when it stops reading a // connection whose pipelined responses have buffered past the high water mark. _paused = false; - #pendingCallback = null; + #pendingCallback: ((error?: Error) => void) | null = null; + #pendingResponseCallback: ((error?: Error) => void) | null = null; #pendingAbortMessage; #closeHandled = false; #peer; @@ -1774,17 +1901,22 @@ function getNodeHTTPServerSocket() { this[kRememberPeer](); } else { handle.ondata = undefined; - handle.ondrain = undefined; + if (!this.#pendingCallback) handle.ondrain = undefined; } } } + uncork() { + const result = super.uncork(); + const response = this._httpMessage; + if (!hasResponseSocketWork(this) && response) scheduleWriteAccountingFlush(response); + return result; + } #onDrain() { const callback = this.#pendingCallback; if (callback) { this.#pendingCallback = null; (callback as Function)(); } - this.emit("drain"); } #onData(chunk, last) { this._unrefTimer(); @@ -1945,6 +2077,7 @@ function getNodeHTTPServerSocket() { this.#pendingCallback = null; (pendingCallback as Function)(writeFailure); } + this.#pendingResponseCallback?.(writeFailure); } // The close of a TLS handle waits for ciphertext that write() reported and a stalled peer does not take, but only once. destroyNow() { @@ -2222,24 +2355,51 @@ function getNodeHTTPServerSocket() { } _write(_chunk, _encoding, _callback) { + if (_encoding === kWriteOperation) { + const response = _chunk.response; + const done = error => { + if (this.#pendingResponseCallback !== done) return; + this.#pendingResponseCallback = null; + response[kResumeResponseSocketWrite] = undefined; + _callback(error); + if (!hasResponseSocketWork(this)) scheduleWriteAccountingFlush(response); + }; + this.#pendingResponseCallback = done; + const resume = () => { + response[kResumeResponseSocketWrite] = undefined; + try { + executeResponseSocketWrite(_chunk, done); + } catch (error) { + done(error); + } + }; + if (_chunk.responseCorked && response[kCorked]) response[kResumeResponseSocketWrite] = resume; + else resume(); + return; + } const handle = this[kHandle]; + const done = error => { + _callback(error); + const response = this._httpMessage; + if (response && !hasResponseSocketWork(this)) scheduleWriteAccountingFlush(response); + }; let err; this._unrefTimer(); try { if (handle) { + handle.ondrain ??= this.#onDrain.bind(this); const flushed = handle.write(_chunk, _encoding); if (!flushed && handle.ondrain) { // Streaming mode (CONNECT tunnels): wait for the native drain // callback before completing the write. - this.#pendingCallback = _callback; + this.#pendingCallback = done; return false; } } } catch (e) { err = e; } - if (err) _callback(err); - else _callback(); + done(err); } pause() { @@ -2268,6 +2428,10 @@ function getNodeHTTPServerSocket() { return this[kHandle]?.hasQueuedTunnelReads === true; } } as unknown as typeof import("node:net").Socket; + // net.Socket's byte-concatenating _writev would discard typed operations. + // Writable's own serial flush keeps response and raw writes in one queue. + NodeHTTPServerSocket.prototype[kWriteOperation] = require("internal/streams/writable").prototype[kWriteOperation]; + NodeHTTPServerSocket.prototype._writev = undefined; Object.defineProperty(NodeHTTPServerSocket, "name", { value: "Socket" }); return NodeHTTPServerSocket; } @@ -2457,6 +2621,38 @@ function releaseRenderedHeaders(flat) { if (flat === scratchFlatHeaders) scratchFlatHeadersBusy = false; } +function snapshotResponseSocketHeaders(res, fromEnd) { + const flat = renderNativeHeaders(res); + try { + const headers = { flat: flat.slice(), autoHeaders: renderedAutoHeaders, keepAliveSecs: renderedKeepAliveSecs }; + res[kResponseHeadersQueued] = headers; + let bytes = + 15 + String(res[kSnapshotStatusMessage] ?? res.statusMessage ?? STATUS_CODES[res.statusCode] ?? "unknown").length; + for (let i = 0; i < flat.length; i += 2) { + if (flat[i] !== "\u0000") bytes += flat[i].length + flat[i + 1].length + 4; + } + if (headers.autoHeaders & AUTO_HEADER_DATE) bytes += 37; + if (headers.autoHeaders & AUTO_HEADER_CONN_KEEP_ALIVE) bytes += 24; + if (headers.autoHeaders & AUTO_HEADER_CONN_CLOSE) bytes += 19; + if (headers.autoHeaders & AUTO_HEADER_KEEP_ALIVE_TIMEOUT) bytes += 22 + String(headers.keepAliveSecs).length; + if (headers.autoHeaders & AUTO_HEADER_TRANSFER_ENCODING_CHUNKED) bytes += 28; + else if (!res.hasHeader("content-length") && !res.hasHeader("transfer-encoding") && res._hasBody) { + if (fromEnd && !res._removedContLen && !res._removedTE) bytes += 18 + String(res._contentLength).length; + else if (willBeChunked(res)) bytes += 28; + } + return bytes + 2; + } finally { + releaseRenderedHeaders(flat); + } +} + +function renderResponseSocketHeaders(res) { + const headers = res[kResponseHeadersQueued]; + renderedAutoHeaders = headers.autoHeaders; + renderedKeepAliveSecs = headers.keepAliveSecs; + return headers.flat; +} + // Auto-header bits (kAutoHeader* in src/jsc/bindings/NodeHTTP.cpp - keep in // sync): renderNativeHeaders reports the framework headers (Date, Connection, // Keep-Alive) through these instead of pushing strings into the flat array; @@ -3030,12 +3226,10 @@ function advanceResponsePipeline(server, socket) { const op = ops[i]; const kind = op[0]; if (kind === "raw") { - // Buffered 1xx bytes: route through the same AsyncSocket buffer the - // response's own writeHead/end use so they precede the final response. - handle.writeInformational(op[1], op[2]); - if (typeof op[3] === "function") process.nextTick(op[3]); + if (ServerResponsePrototypeWriteRaw.$call(res, op[1], op[2], op[3]) === false) hitBackpressure = true; } else if (kind === "write") { - if (ServerResponsePrototypeWrite.$call(res, op[1], op[2], op[3]) === false) hitBackpressure = true; + const chunk = op[1] === undefined && !res._hasBody ? "" : op[1]; + if (ServerResponsePrototypeWrite.$call(res, chunk, op[2], op[3]) === false) hitBackpressure = true; } else { ServerResponsePrototypeEnd.$call(res, op[1], op[2], op[3]); } @@ -3075,10 +3269,7 @@ function emitPipelinedDrainNT(res) { // writableHighWaterMark and pause reads one request earlier than Node would, // deadlocking clients that pipeline the unblocking request behind the crossing // one (test-http-pipeline-socket-parser-typeerror). -function accountQueuedHeaderBytes(res, queued) { - if (queued.headerBytes !== 0) { - return; - } +function responseHeaderByteLength(res) { // Status line: "HTTP/1.1 NNN \r\n". let bytes = 15 + String(res.statusMessage ?? STATUS_CODES[res.statusCode] ?? "unknown").length; const outHeaders = res[kOutHeaders]; @@ -3103,6 +3294,12 @@ function accountQueuedHeaderBytes(res, queued) { if (!res.hasHeader("connection")) bytes += 24; if (!res.hasHeader("content-length") && !res.hasHeader("transfer-encoding")) bytes += 28; bytes += 2; + return bytes; +} + +function accountQueuedHeaderBytes(res, queued) { + if (queued.headerBytes !== 0) return; + const bytes = responseHeaderByteLength(res); queued.headerBytes = bytes; queued.bytes += bytes; res.outputSize += bytes; @@ -3162,6 +3359,7 @@ function bufferPipelinedEnd(res, queued, chunk, encoding, callback) { addPipelineOutgoingData(queued, bytes); } queued.ended = true; + res[kCorked] = 0; res.finished = true; return res; } @@ -3281,6 +3479,10 @@ Object.defineProperty(ServerResponse.prototype, "socket", { return (this[fakeSocketSymbol] ??= new FakeSocket(this)); }, set(value) { + for (let n = 0; n < this[kCorked]; n++) { + value?.cork(); + this[kSocket]?.uncork(); + } this[fakeSocketSymbol] = value; // Keep the OutgoingMessage storage in sync: the standalone no-handle path // routes through OutgoingMessage.write/end, whose kSocket-gated checks @@ -3313,18 +3515,20 @@ Object.defineProperty(ServerResponse.prototype, "chunkedEncoding", { Object.defineProperty(ServerResponse.prototype, "writableCorked", { get() { - return this.socket?.writableCorked ?? 0; + return this[kCorked]; }, set(_value) {}, }); ServerResponse.prototype.cork = function cork() { - // socket is null while this response is queued behind a pipelined response. + this[kCorked]++; this.socket?.cork(); }; ServerResponse.prototype.uncork = function uncork() { + this[kCorked]--; this.socket?.uncork(); + if (!this[kCorked]) this[kResumeResponseSocketWrite]?.(); }; ServerResponse.prototype.setTimeout = function setTimeout(msecs, callback) { @@ -3389,6 +3593,9 @@ ServerResponse.prototype._writeRaw = function (chunk, encoding, callback) { addPipelineOutgoingData(queued, bytes); return queued.bytes < this.writableHighWaterMark; } + if (arguments[3] !== kResponseSocketWrite && shouldQueueResponseWrite(this)) { + return queueResponseSocketWrite(this, "raw", chunk, encoding, callback); + } // Write through the response handle's AsyncSocket buffer (same path as // writeHead/end) so 1xx lines share ordering with the final response bytes; // socket.write() would land in the socket handle's separate stream buffer. @@ -3396,6 +3603,7 @@ ServerResponse.prototype._writeRaw = function (chunk, encoding, callback) { if (typeof callback === "function") process.nextTick(callback); return true; }; +const ServerResponsePrototypeWriteRaw = ServerResponse.prototype._writeRaw; ServerResponse.prototype.writeEarlyHints = function (hints, cb) { validateObject(hints, "hints"); @@ -3492,6 +3700,11 @@ ServerResponse.prototype.writeContinue = function (cb) { this._sent100 = true; return; } + if (shouldQueueResponseWrite(this)) { + queueResponseSocketWrite(this, "continue", undefined, undefined, cb); + this._sent100 = true; + return; + } const native = this.socket?.[kHandle]?.response; if (native) native.writeContinue(); else this[kHandle]?.writeContinue?.(); @@ -3501,8 +3714,9 @@ ServerResponse.prototype.writeContinue = function (cb) { // This end method is actually on the OutgoingMessage prototype in Node.js // But we don't want it for the fetch() response version. -ServerResponse.prototype.end = function (chunk, encoding, callback) { +ServerResponse.prototype.end = function (chunk, encoding, callback, token = undefined, operation: any = undefined) { const handle = this[kHandle]; + const fromSocket = token === kResponseSocketWrite; if ($isCallable(chunk)) { callback = chunk; @@ -3519,7 +3733,7 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) { const droppedChunk = !!chunk && this.destroyed && !this.finished; if (droppedChunk) chunk = undefined; - if (hasServerResponseFinished(this, chunk, callback, true)) { + if (!fromSocket && hasServerResponseFinished(this, chunk, callback, true)) { return this; } @@ -3535,6 +3749,17 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) { if (!handle) { return OutgoingMessagePrototype.end.$call(this, chunk, encoding, callback); } + if (!fromSocket && shouldQueueResponseWrite(this)) { + const socket = this.socket; + queueResponseSocketWrite(this, "end", chunk, encoding, callback); + this.finished = true; + this[kCorked] = 0; + // Node's end() releases every explicit cork, including nested socket corks. + socket._writableState.corked = 1; + socket.uncork(); + this[kResumeResponseSocketWrite]?.(); + return this; + } if (this[headerStateSymbol] === NodeHTTPHeaderState.none) { // Implicit header: Node's write_() runs _implicitHeader() (which derives @@ -3591,7 +3816,7 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) { let draining = false; if (headerState !== sentState) { { - const renderedHeaders = renderNativeHeaders(this); + const renderedHeaders = fromSocket ? renderResponseSocketHeaders(this) : renderNativeHeaders(this); let contentLength; try { // One native crossing for cork + writeHead + end (writeHeadAndEnd @@ -3602,7 +3827,7 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) { renderedHeaders, chunk, encoding, - strictContentLength(this, headerState, true), + fromSocket ? operation.strict : strictContentLength(this, headerState, true), renderedAutoHeaders, renderedKeepAliveSecs, ); @@ -3639,10 +3864,17 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) { // (no native call in between can change it), so reuse its bits instead of // paying two more native getter crossings. if (!(!chunk && flags & NodeHTTPResponseFlags.ended) && !(flags & NodeHTTPResponseFlags.socket_closed)) { - draining = handle.end(chunk, encoding, undefined, strictContentLength(this, headerState, true)) < 0; + draining = + handle.end( + chunk, + encoding, + undefined, + fromSocket ? operation.strict : strictContentLength(this, headerState, true), + ) < 0; } } this._header = " "; + this[kCorked] = 0; // The socket is NOT detached here: like Node.js, res.socket stays assigned // until the response 'finish' machinery runs (the dispatcher detaches it // right after a synchronously-finished handler returns, or via its 'finish' @@ -3708,8 +3940,9 @@ Object.defineProperty(ServerResponse.prototype, "writable", { set(_value) {}, }); -ServerResponse.prototype.write = function (chunk, encoding, callback) { +ServerResponse.prototype.write = function (chunk, encoding, callback, token = undefined, operation: any = undefined) { const handle = this[kHandle]; + const fromSocket = token === kResponseSocketWrite; if ($isCallable(chunk)) { callback = chunk; @@ -3722,7 +3955,7 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { callback = undefined; } - if (hasServerResponseFinished(this, chunk, callback, false)) { + if (!fromSocket && hasServerResponseFinished(this, chunk, callback, false)) { return false; } @@ -3735,6 +3968,9 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { } } + if (!fromSocket && shouldQueueResponseWrite(this)) { + return queueResponseSocketWrite(this, "write", chunk, encoding, callback); + } let result = 0; const headerState = this[headerStateSymbol]; @@ -3775,14 +4011,14 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { return false; } - const strict = strictContentLength(this, headerState, false); + const strict = fromSocket ? operation.strict : strictContentLength(this, headerState, false); // Node returns true for every write to a response that cannot have a body (write_ in lib/_http_outgoing.js). const hasBody = this._hasBody; const onWritable = hasBody ? allowWritesToContinue.bind(this) : undefined; if (this[headerStateSymbol] !== NodeHTTPHeaderState.sent) { handle.cork(() => { - const renderedHeaders = renderNativeHeaders(this); + const renderedHeaders = fromSocket ? renderResponseSocketHeaders(this) : renderNativeHeaders(this); try { handle.writeHead( this[kSnapshotStatusCode] ?? this.statusCode, @@ -3807,6 +4043,7 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { } if (result < 0 && hasBody) { + if (!fromSocket) this[kResponseQueueNeedDrain] = true; if (callback) { // The write was buffered due to backpressure. // We need to defer the callback until the write actually goes through. @@ -3820,6 +4057,7 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { // A write callback reports that its bytes reached the socket: uncork them before the callback can destroy the response. handle.flushHeaders(); if (hasBody && handle.bufferedAmount > 0) { + if (!fromSocket) this[kResponseQueueNeedDrain] = true; this[kPendingCallbacks].push(callback); handle.onwritable = allowWritesToContinue.bind(this); return false; @@ -3845,7 +4083,7 @@ ServerResponse.prototype.write = function (chunk, encoding, callback) { // While replaying the writes a pipelined response buffered before it was // assigned the socket, skip the per-turn accounting: 'drain' for those // writes is handled by advanceResponsePipeline / the native drain callback. - if (written > 0 && !this[kReplayingPipelinedOps]) { + if (written > 0 && !fromSocket && !this[kReplayingPipelinedOps]) { this[kBytesBuffered] = (this[kBytesBuffered] ?? 0) + written; scheduleWriteAccountingFlush(this); } @@ -3863,7 +4101,12 @@ const kBytesBuffered = Symbol("kBytesBuffered"); const kAccountingFlushScheduled = Symbol("kAccountingFlushScheduled"); function flushWriteAccountingNT(res) { res[kAccountingFlushScheduled] = false; - const needsDrain = (res[kBytesBuffered] ?? 0) >= res.writableHighWaterMark; + if (res.socket && hasResponseSocketWork(res.socket)) return; + // Native backpressure owns this obligation until its writable callback fires. + if (res[kResponseQueueNeedDrain] && res[kHandle]?.onwritable !== undefined) return; + const queuedNeedDrain = res[kResponseQueueNeedDrain]; + const needsDrain = queuedNeedDrain || (res[kBytesBuffered] ?? 0) >= res.writableHighWaterMark; + if (queuedNeedDrain) res[kResponseQueueNeedDrain] = false; res[kBytesBuffered] = 0; if (needsDrain && !res.destroyed && !res.finished) { res.emit("drain"); @@ -3921,6 +4164,10 @@ ServerResponse.prototype._implicitHeader = function () { Object.defineProperty(ServerResponse.prototype, "writableNeedDrain", { get() { + const socket = this.socket; + if (socket?.[kResponseQueueInitialized] && hasResponseSocketWork(socket)) { + return !this.destroyed && !this.finished && !!this[kResponseQueueNeedDrain]; + } // 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 ( @@ -3928,6 +4175,7 @@ Object.defineProperty(ServerResponse.prototype, "writableNeedDrain", { !this.finished && ((handle?.bufferedAmount ?? 0) !== 0 || handle?.onwritable !== undefined || + this[kResponseQueueNeedDrain] || (this[kBytesBuffered] ?? 0) >= this.writableHighWaterMark || (this[kPipelinedQueuedState]?.needDrain ?? false)) ); @@ -3948,6 +4196,10 @@ Object.defineProperty(ServerResponse.prototype, "writableFinished", { Object.defineProperty(ServerResponse.prototype, "writableLength", { get() { if (this.writableFinished) return 0; + const socket = this.socket; + if (socket?.[kResponseQueueInitialized] && hasResponseSocketWork(socket)) { + return socket.writableLength + (this[kBytesBuffered] ?? 0); + } // Bytes handed off this event-loop turn (including chunked framing, like // Node.js's outputData accounting) plus whatever the native handle still // has buffered, plus anything buffered while queued behind a pipelined @@ -3978,6 +4230,7 @@ ServerResponse.prototype._send = function (data, encoding, callback, _byteLength if (!handle) { return OutgoingMessagePrototype._send.$apply(this, arguments); } + if (shouldQueueResponseWrite(this)) return queueResponseSocketWrite(this, "write", data, encoding, callback); const strict = strictContentLength(this, this[headerStateSymbol], false); if (this[headerStateSymbol] !== NodeHTTPHeaderState.sent) { @@ -4122,12 +4375,18 @@ ServerResponse.prototype.flushHeaders = function () { return; } + if (arguments[0] !== kResponseSocketWrite && shouldQueueResponseWrite(this)) { + queueResponseSocketWrite(this, "flush", undefined, undefined, undefined); + return; + } + const handle = this[kHandle]; if (handle) { if (this[headerStateSymbol] === NodeHTTPHeaderState.assigned) { this[headerStateSymbol] = NodeHTTPHeaderState.sent; - const renderedHeaders = renderNativeHeaders(this); + const renderedHeaders = + arguments[0] === kResponseSocketWrite ? renderResponseSocketHeaders(this) : renderNativeHeaders(this); try { handle.writeHead( this[kSnapshotStatusCode] ?? this.statusCode, @@ -4149,6 +4408,8 @@ ServerResponse.prototype.flushHeaders = function () { this._send(""); } }; +const ServerResponsePrototypeFlushHeaders = ServerResponse.prototype.flushHeaders; + function updateHasBody(response, statusCode) { // RFC 2616, 10.2.5: @@ -4190,7 +4451,7 @@ function callWriteHeadIfObservable(self, headerState, fromEnd?) { function allowWritesToContinue() { this._callPendingCallbacks(); - this.emit("drain"); + flushWriteAccountingNT(this); } function failPendingWriteCallbacks(res, error) { diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp index 6f09c589e38c..d422d6f3c261 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp @@ -182,6 +182,11 @@ static bool isTunnel(const JSNodeHTTPServerSocket* self) return self->is_ssl ? isInTunnelMode(socket) : isInTunnelMode(socket); } +bool JSNodeHTTPServerSocket::flushesStreamBufferOnDrain() const +{ + return isTunnel(this); +} + void JSNodeHTTPServerSocket::applyTunnelReads() { if (!isTunnel(this)) { diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h index be9eff8308ee..c7df6f550ec9 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h @@ -157,7 +157,7 @@ class JSNodeHTTPServerSocket : public JSC::JSDestructibleObject { /* Sends the response bytes that are not in the uWS buffer (the zero-copy tail of a res.write(), the cork buffer) to the kernel or into it. A raw write or a FIN then goes out behind them. */ void flushResponseBytesAhead(); /* Only a tunnel gets the drain call that flushes streamBuffer. On any other socket the uWS buffer takes what the kernel does not. */ - bool flushesStreamBufferOnDrain() const { return !!functionToCallOnDrain; } + bool flushesStreamBufferOnDrain() const; /* The WebSocket that adopted the connection reads from here on. */ void releaseTunnelReadsForUpgrade(); diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp index 1d73c52dd2e7..1224e80bb108 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp @@ -278,8 +278,8 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionNodeHTTPServerSocketWrite, (JSC::JSGlobalObje thisObject->flushResponseBytesAhead(); const bool hold = thisObject->hasUnsentResponseBytes(); auto result = us_socket_buffered_js_write(thisObject->socket, thisObject->is_ssl, thisObject->ended, hold, thisObject->flushesStreamBufferOnDrain(), &thisObject->streamBuffer, globalObject, JSValue::encode(callFrame->argument(0)), JSValue::encode(callFrame->argument(1))); - // JS parks the write callback on false only when it has an ondrain (_write in _http_server.ts). - if (hold && thisObject->functionToCallOnDrain && JSValue::decode(result).isFalse()) { + // Both held responses and a partial raw write can leave bytes in the uWS buffer. + if (thisObject->functionToCallOnDrain && JSValue::decode(result).isFalse()) { thisObject->heldWriteAwaitsDrain = true; } thisObject->updateTunnelIdle(); diff --git a/test/js/node/http/node-http-nested-cork.test.ts b/test/js/node/http/node-http-nested-cork.test.ts index 3b1fef67e884..6c12ebba449d 100644 --- a/test/js/node/http/node-http-nested-cork.test.ts +++ b/test/js/node/http/node-http-nested-cork.test.ts @@ -1,5 +1,556 @@ import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe } from "harness"; +import { bunEnv, bunExe, nodeMajorVersion, runtimesWithNode } from "harness"; + +async function responseCorkOracle(mode: string) { + const assert = require("node:assert/strict"); + const http = require("node:http"); + const net = require("node:net"); + const { once } = require("node:events"); + const { setImmediate: immediate } = require("node:timers/promises"); + const { pipeline } = require("node:stream/promises"); + if (mode.startsWith("standalone")) { + const chunks: Buffer[] = []; + const finished = Promise.withResolvers(); + const written = Promise.withResolvers(); + let callbacks = 0; + const server = http.createServer(); + server.on("upgrade", (_req, socket) => { + const req = new http.IncomingMessage(socket); + req.httpVersionMajor = 1; + req.httpVersionMinor = 1; + const res = new http.ServerResponse(req); + res.assignSocket(socket); + res.strictContentLength = mode === "standalone-strict"; + res.setHeader("content-length", 2); + const onWrite = error => { + assert.ifError(error); + if (++callbacks === 2) written.resolve(); + }; + res.on("finish", () => { + finished.resolve(); + socket.end(); + }); + res.cork(); + res.write("A", onWrite); + res.write("B", onWrite); + res.end(); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const client = net.connect(server.address().port, "127.0.0.1"); + client.on("data", chunk => chunks.push(chunk)); + try { + await once(client, "connect"); + client.write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\nUpgrade: test\r\n\r\n"); + await once(client, "close"); + await Promise.all([finished.promise, written.promise]); + assert.equal(callbacks, 2); + assert.match(Buffer.concat(chunks).toString(), /^HTTP\/1\.1 200 OK\r\n/); + assert.ok(Buffer.concat(chunks).toString().endsWith("\r\n\r\nAB")); + } finally { + client.destroy(); + server.closeAllConnections(); + await new Promise(resolve => server.close(resolve)); + } + console.log("ok"); + return; + } + const chunks: Buffer[] = []; + const discardsResponse = + mode === "destroy" || mode === "destroy-no-error" || mode === "strict-end" || mode === "strict-first-write"; + const second = Promise.withResolvers(); + const reassigned = Promise.withResolvers(); + const completed = Promise.withResolvers(); + let handled = 0; + let client; + let first; + const highWaterMark = mode === "accounting" ? 64 : mode === "native-queued" ? 64 * 1024 * 1024 : 1024; + const server = http.createServer({ highWaterMark }, (req, res) => { + if (req.url === "/barrier") return res.end("barrier"); + handled++; + if (handled === 2) { + assert.equal(res.socket, null); + res.setHeader("content-length", 2); + res.cork(); + if (mode === "reassigned-no-body") { + res.statusCode = 204; + res.write("ignored"); + res.end(); + second.resolve(); + return; + } + if (mode === "reassigned-informational") { + let callbackRan = false; + res.writeContinue(() => (callbackRan = true)); + res.once("socket", () => { + void (async () => { + await barrier(); + assert.equal(Buffer.concat(chunks).includes("100 Continue"), false); + assert.equal(callbackRan, false); + res.uncork(); + res.end("C"); + })().then(reassigned.resolve, reassigned.reject); + }); + res.write("B"); + second.resolve(); + return; + } + res.write("B"); + res.end("C"); + assert.equal(res.writableCorked, 0); + second.resolve(); + return; + } + first = res; + void handle(res).then(completed.resolve, completed.reject); + }); + const counts = res => [res.writableCorked, res.socket.writableCorked]; + const barrier = async () => { + const response = await fetch(`http://127.0.0.1:${server.address().port}/barrier`); + await response.arrayBuffer(); + await immediate(); + }; + async function handle(res) { + if (mode === "informational-pressure") { + res.setHeader("content-length", 2); + const written = Promise.withResolvers(); + let callbacks = 0; + res.cork(); + res.writeEarlyHints({ link: `; rel=preload` }, error => { + if (error) written.reject(error); + else { + callbacks++; + written.resolve(); + } + }); + res.uncork(); + res.end("AB"); + await barrier(); + assert.equal(callbacks, 0); + client.resume(); + await written.promise; + assert.equal(callbacks, 1); + return; + } + if (mode.startsWith("strict-")) { + res.strictContentLength = true; + const contentLength = ["4"]; + res.setHeader("content-length", contentLength); + res.cork(); + if (mode === "strict-no-body") { + res.statusCode = 204; + assert.throws(() => res.write(undefined), { code: "ERR_INVALID_ARG_TYPE" }); + res.end("AB"); + } else if (mode === "strict-mutable-header") { + res.write("AB"); + contentLength[0] = "6"; + res.end("CD"); + } else if (mode === "strict-end") { + assert.throws(() => res.end("AB"), { code: "ERR_HTTP_CONTENT_LENGTH_MISMATCH" }); + res.destroy(); + } else if (mode === "strict-first-write") { + assert.equal(res.write("ABCDE"), true); + assert.throws(() => res.end(), { code: "ERR_HTTP_CONTENT_LENGTH_MISMATCH" }); + res.destroy(); + } else { + assert.equal(res.write("AB"), true); + assert.throws(() => res.write("CDE"), { code: "ERR_HTTP_CONTENT_LENGTH_MISMATCH" }); + res.end("CD"); + } + return; + } + if (mode === "pipeline") { + res.setHeader("content-length", 4 * 1024 * 1024); + res.cork(); + let reads = 0; + const started = Promise.withResolvers(); + const sending = pipeline(async function* () { + for (let i = 0; i < 8; i++) { + reads++; + started.resolve(); + yield Buffer.alloc(512 * 1024, 65); + } + }, res); + // Register failure handling before any assertion can interrupt the source. + sending.catch(() => {}); + await started.promise; + for (let i = 0; i < 4; i++) await immediate(); + await barrier(); + assert.equal(chunks.length, 0); + assert.equal(reads, 1); + assert.equal(res.writableCorked, 1); + assert.equal(res.writableNeedDrain, true); + assert.ok(res.writableLength >= 512 * 1024); + res.uncork(); + await sending; + return; + } + if (mode.startsWith("reassigned")) { + res.setHeader("content-length", 1); + res.cork(); + res.write("A"); + await second.promise; + assert.equal(chunks.length, 0); + res.uncork(); + res.end(); + return; + } + if (mode === "native-raw-drain" || mode === "native-queued") { + const large = Buffer.alloc(32 * 1024 * 1024, 65); + res.setHeader("content-length", large.length + (mode === "native-raw-drain" ? 1 : 0)); + let drains = 0; + res.on("drain", () => drains++); + if (mode === "native-queued") { + res.cork(); + const written = new Promise(resolve => assert.equal(res.write(large, resolve), true)); + res.uncork(); + assert.equal(res.writableNeedDrain, false); + await written; + await immediate(); + assert.equal(drains, 0); + } else { + const drained = once(res, "drain"); + assert.equal(res.write(large), false); + const written = new Promise((resolve, reject) => + res.socket.write("B", error => (error ? reject(error) : resolve())), + ); + await Promise.all([drained, written]); + assert.equal(drains, 1); + } + assert.equal(res.writableNeedDrain, false); + res.end(); + return; + } + if (mode === "raw-drain") { + const large = Buffer.alloc(16 * 1024 * 1024, 66); + res.setHeader("content-length", large.length + 2048); + let drains = 0; + res.on("drain", () => drains++); + assert.equal(res.write(Buffer.alloc(2048, 65)), false); + await new Promise((resolve, reject) => res.socket.write(large, error => (error ? reject(error) : resolve()))); + await immediate(); + assert.equal(res.writableNeedDrain, false); + assert.equal(drains, 1); + res.end(); + return; + } + if (mode === "interleaved" || mode === "after-uncork") { + const large = Buffer.alloc(16 * 1024 * 1024, 65); + res.setHeader("content-length", large.length + (mode === "interleaved" ? 4 : 3)); + res.cork(); + assert.equal(res.write(large), false); + if (mode === "interleaved") res.socket.write("B"); + res.write("C"); + res.uncork(); + assert.equal(res.socket.writableCorked, 0); + assert.ok(res.socket.writableLength > 0); + res.write("D"); + if (mode === "interleaved") res.socket.write("E", () => res.end()); + else res.end("E"); + return; + } + if (mode !== "cross-chunked") res.setHeader("content-length", mode === "high-water" ? 2048 : 2); + if (mode === "accounting") { + res.sendDate = false; + res.cork(); + assert.equal(res.write("A"), true); + assert.equal(res.writableLength, 58); + assert.equal(res.writableNeedDrain, false); + res.end("B"); + return; + } + if (mode === "headers-frozen") { + const value = ["before"]; + res.setHeader("x-proof", value); + res.cork(); + res.write("A"); + value[0] = "after"; + res.end("B"); + return; + } + if (mode === "destroy" || mode === "destroy-no-error") { + const callbacks = []; + const writes = []; + res.cork(); + writes.push( + new Promise(resolve => + res.write("A", error => { + callbacks.push(error?.code); + resolve(); + }), + ), + ); + writes.push( + new Promise(resolve => + res.socket.write("B", error => { + callbacks.push(error?.code); + resolve(); + }), + ), + ); + res.socket.on("error", () => {}); + res.socket.destroy( + mode === "destroy" ? Object.assign(new Error("corked abort"), { code: "CORKED_ABORT" }) : undefined, + ); + await Promise.all(writes); + const code = mode === "destroy" ? "CORKED_ABORT" : "ERR_STREAM_DESTROYED"; + assert.deepEqual(callbacks, [code, code]); + assert.equal(chunks.length, 0); + return; + } + if (mode === "headers" || mode === "informational") { + res.cork(); + if (mode === "headers") { + res.flushHeaders(); + assert.equal(res.headersSent, true); + } else { + res.writeContinue(); + } + res.write("A"); + await barrier(); + assert.equal(chunks.length, 0); + res.uncork(); + } else if (mode === "callbacks") { + const callbacks = []; + res.cork(); + const responseWrite = new Promise(resolve => + res.write("A", () => { + callbacks.push("response"); + resolve(); + }), + ); + const rawWrite = new Promise(resolve => + res.socket.write("B", () => { + callbacks.push("raw"); + resolve(); + }), + ); + res.uncork(); + await Promise.all([responseWrite, rawWrite]); + assert.deepEqual(callbacks, ["response", "raw"]); + res.end(); + return; + } else if (mode === "counts") { + assert.equal(res.write.length, 3); + assert.equal(res.end.length, 3); + assert.deepEqual(counts(res), [0, 0]); + res.cork(); + assert.deepEqual(counts(res), [1, 1]); + res.cork(); + assert.deepEqual(counts(res), [2, 2]); + res.write("A"); + res.uncork(); + assert.deepEqual(counts(res), [1, 1]); + await barrier(); + assert.equal(chunks.length, 0); + res.uncork(); + assert.deepEqual(counts(res), [0, 0]); + } else if (mode === "cross-content-length" || mode === "cross-chunked") { + res.cork(); + res.write("A"); + res.socket.uncork(); + assert.deepEqual(counts(res), [1, 0]); + await barrier(); + if (mode === "cross-chunked") assert.equal(chunks.length, 0); + res.uncork(); + assert.deepEqual(counts(res), [0, 0]); + } else if (mode === "socket-cork") { + res.socket.cork(); + assert.deepEqual(counts(res), [0, 1]); + res.write("A"); + res.uncork(); + assert.deepEqual(counts(res), [-1, 0]); + } else if (mode === "end-corked") { + res.socket.cork(); + res.cork(); + res.cork(); + res.write("A"); + res.end("B"); + assert.deepEqual(counts(res), [0, 0]); + return; + } else if (mode === "high-water") { + res.cork(); + let callbacks = 0; + let drains = 0; + res.on("drain", () => drains++); + assert.equal( + res.write(Buffer.alloc(2048, 65), () => callbacks++), + false, + ); + await barrier(); + assert.equal(chunks.length, 0); + assert.equal(callbacks, 0); + assert.equal(drains, 0); + assert.equal(res.writableNeedDrain, true); + assert.ok(res.writableLength >= 2048); + const drained = once(res, "drain"); + res.uncork(); + await drained; + await immediate(); + assert.equal(res.writableNeedDrain, false); + assert.equal(callbacks, 1); + assert.equal(drains, 1); + res.end(); + return; + } else { + res.cork(); + res.write("A"); + assert.equal(chunks.length, 0); + await new Promise(resolve => (mode === "next-tick" ? process.nextTick(resolve) : setTimeout(resolve, 0))); + res.uncork(); + } + res.end("B"); + } + function parseResponses(bytes) { + const bodies = []; + while (bytes.length) { + const headerEnd = bytes.indexOf("\r\n\r\n"); + assert.ok(headerEnd >= 0); + const header = bytes.subarray(0, headerEnd).toString(); + bytes = bytes.subarray(headerEnd + 4); + if (/^HTTP\/1\.1 1\d\d /.test(header)) continue; + if (/^HTTP\/1\.1 (204|304) /.test(header)) { + bodies.push(Buffer.alloc(0)); + continue; + } + if (/transfer-encoding: chunked/i.test(header)) { + const parts = []; + while (true) { + const lineEnd = bytes.indexOf("\r\n"); + const size = parseInt(bytes.subarray(0, lineEnd).toString(), 16); + assert.ok(Number.isFinite(size)); + bytes = bytes.subarray(lineEnd + 2); + if (!size) { + bytes = bytes.subarray(2); + break; + } + parts.push(bytes.subarray(0, size)); + bytes = bytes.subarray(size + 2); + } + bodies.push(Buffer.concat(parts)); + } else { + const length = Number(/content-length: (\d+)/i.exec(header)?.[1]); + assert.ok(Number.isFinite(length)); + assert.ok(bytes.length >= length); + bodies.push(bytes.subarray(0, length)); + bytes = bytes.subarray(length); + } + } + return bodies; + } + try { + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + client = net.connect(server.address().port, "127.0.0.1"); + if (mode === "informational-pressure") client.pause(); + const closed = new Promise((resolve, reject) => { + client.on("data", chunk => chunks.push(chunk)); + client.on("error", error => { + if (!discardsResponse) reject(error); + }); + client.on("close", resolve); + }); + closed.catch(() => {}); + client.write( + mode.startsWith("reassigned") + ? "GET / HTTP/1.1\r\nHost: localhost\r\n\r\nGET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n" + : "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ); + await completed.promise; + if (mode === "reassigned-informational") await reassigned.promise; + await closed; + if (mode === "headers-frozen") assert.match(Buffer.concat(chunks).toString(), /x-proof: before\r\n/i); + if (mode === "strict-no-body") { + const bytes = Buffer.concat(chunks).toString(); + assert.match(bytes, /^HTTP\/1.1 204 /); + assert.equal(bytes.slice(bytes.indexOf("\r\n\r\n") + 4), ""); + } else if (!discardsResponse) { + const bodies = parseResponses(Buffer.concat(chunks)); + if (mode.startsWith("reassigned")) + assert.deepEqual( + bodies.map(body => body.toString()), + ["A", mode === "reassigned-no-body" ? "" : "BC"], + ); + else if (mode === "pipeline") assert.deepEqual(bodies, [Buffer.alloc(4 * 1024 * 1024, 65)]); + else if (mode === "high-water") assert.deepEqual(bodies, [Buffer.alloc(2048, 65)]); + else if (mode === "strict-sequence" || mode === "strict-mutable-header") + assert.deepEqual(bodies, [Buffer.from("ABCD")]); + else if (mode === "raw-drain") { + assert.equal(bodies[0].length, 16 * 1024 * 1024 + 2048); + assert.deepEqual(bodies[0].subarray(0, 2048), Buffer.alloc(2048, 65)); + assert.deepEqual(bodies[0].subarray(2048), Buffer.alloc(16 * 1024 * 1024, 66)); + } else if (mode === "native-raw-drain" || mode === "native-queued") { + assert.equal(bodies[0].length, 32 * 1024 * 1024 + (mode === "native-raw-drain" ? 1 : 0)); + assert.equal(bodies[0].subarray(-1).toString(), mode === "native-raw-drain" ? "B" : "A"); + } else if (mode === "interleaved" || mode === "after-uncork") { + assert.equal(bodies[0].subarray(16 * 1024 * 1024).toString(), mode === "interleaved" ? "BCDE" : "CDE"); + } else + assert.deepEqual( + bodies.map(body => body.toString()), + ["AB"], + ); + } + } finally { + client?.destroy(); + first?.destroy(); + server.closeAllConnections(); + await new Promise(resolve => server.close(resolve)); + } + console.log("ok"); +} + +// Node 26 changed cross-cork chunked header flushing; these regressions target Node 24's contract. +const corkRuntimes = runtimesWithNode(24).filter(([runtime]) => runtime === "bun" || nodeMajorVersion() === 24); +describe.each(corkRuntimes)("response cork Node oracle (%s)", (_runtime, executable) => { + for (const mode of [ + "pipeline", + "counts", + "cross-content-length", + "cross-chunked", + "socket-cork", + "end-corked", + "high-water", + "next-tick", + "timer", + "destroy", + "destroy-no-error", + "headers", + "informational", + "callbacks", + "reassigned", + "interleaved", + "after-uncork", + "accounting", + "headers-frozen", + "raw-drain", + "reassigned-informational", + "strict-end", + "strict-sequence", + "strict-first-write", + "strict-no-body", + "strict-mutable-header", + "reassigned-no-body", + "native-raw-drain", + "native-queued", + "standalone", + "standalone-strict", + "informational-pressure", + ]) { + test.concurrent(mode, async () => { + await using child = Bun.spawn({ + cmd: [ + executable, + "-e", + `(${responseCorkOracle.toString()})(${JSON.stringify(mode)}).catch(error => { console.error(error); process.exit(1); })`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([child.stdout.text(), child.stderr.text(), child.exited]); + expect({ stdout, stderr, exitCode }).toEqual({ stdout: "ok\n", stderr: "", exitCode: 0 }); + }); + } +}); // Security test: with two cork slots that can be stolen/evicted, a bug in slot // bookkeeping could cause bytes meant for socket A to land in socket B's diff --git a/test/js/node/http/node-http-with-ws.test.ts b/test/js/node/http/node-http-with-ws.test.ts index 411dc331b233..277e14617e20 100644 --- a/test/js/node/http/node-http-with-ws.test.ts +++ b/test/js/node/http/node-http-with-ws.test.ts @@ -260,27 +260,19 @@ describe.concurrent("npm ws on node:http upgrade sockets", () => { test.each([ ["a fresh caller cork", "fresh"], - ["a caller cork below the dispatcher cork", "below-dispatcher"], + ["a caller cork on a reused socket", "reused"], ["nested caller corks after a partial release", "partial-nested"], - ["a caller cork that replaced the dispatcher cork", "replaced-dispatcher"], - ["a caller cork when dispatcher corking is suppressed", "suppressed-dispatcher"], + ["a caller cork after uncorking and corking again", "recorked"], ] as const)("preserves %s and its pending writes", async (_name, mode) => { let serverSocket!: Socket; - let suppressedCork: Socket["cork"] | undefined; const server = http.createServer((_req, res) => res.end("ok")); server.on("connection", socket => { serverSocket = socket; - if (mode !== "replaced-dispatcher") socket.cork(); - if (mode === "partial-nested") socket.cork(); - if (mode === "suppressed-dispatcher") { - suppressedCork = socket.cork; - socket.cork = () => {}; - } + if (mode === "fresh") socket.cork(); }); const wss = new NpmWebSocketServer({ noServer: true }); const upgraded = Promise.withResolvers(); server.on("upgrade", (req, socket, head) => { - if (suppressedCork) socket.cork = suppressedCork; try { wss.handleUpgrade(req, socket, head, ws => { ws.send("pending"); @@ -297,12 +289,16 @@ describe.concurrent("npm ws on node:http upgrade sockets", () => { try { if (mode !== "fresh") { await finishKeepAliveRequest(port, agent); + serverSocket.cork(); } - if (mode === "replaced-dispatcher") { + if (mode === "recorked") { serverSocket.uncork(); serverSocket.cork(); } - if (mode === "partial-nested") serverSocket.uncork(); + if (mode === "partial-nested") { + serverSocket.cork(); + serverSocket.uncork(); + } client = new NpmWebSocket(`ws://127.0.0.1:${port}/upgrade`, { agent }); const opened = waitForWebSocketOpen(client); const message = waitForWebSocketMessage(client); From df5340e779cd54e0a73d44b98cffebef9cfd4315 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 24 Oct 2026 00:02:44 -0700 Subject: [PATCH 02/10] test(http): cover current Node releases in cork oracle --- src/js/node/_http_server.ts | 4 ++-- test/js/node/http/node-http-nested-cork.test.ts | 5 ++--- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 9a9e78e8309b..e53b32e8f0c7 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -1488,7 +1488,8 @@ function shouldQueueResponseWrite(response) { return !!( socket && (hasResponseSocketWork(socket) || (response[kCorked] && willBeChunked(response))) && - NodeHTTPServerSocket && socket instanceof NodeHTTPServerSocket && + NodeHTTPServerSocket && + socket instanceof NodeHTTPServerSocket && response[kHandle] ); } @@ -4410,7 +4411,6 @@ ServerResponse.prototype.flushHeaders = function () { }; const ServerResponsePrototypeFlushHeaders = ServerResponse.prototype.flushHeaders; - function updateHasBody(response, statusCode) { // RFC 2616, 10.2.5: // The 204 response MUST NOT include a message-body, and thus is always diff --git a/test/js/node/http/node-http-nested-cork.test.ts b/test/js/node/http/node-http-nested-cork.test.ts index 6c12ebba449d..1b31ded64c8c 100644 --- a/test/js/node/http/node-http-nested-cork.test.ts +++ b/test/js/node/http/node-http-nested-cork.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, nodeMajorVersion, runtimesWithNode } from "harness"; +import { bunEnv, bunExe, runtimesWithNode } from "harness"; async function responseCorkOracle(mode: string) { const assert = require("node:assert/strict"); @@ -498,8 +498,7 @@ async function responseCorkOracle(mode: string) { console.log("ok"); } -// Node 26 changed cross-cork chunked header flushing; these regressions target Node 24's contract. -const corkRuntimes = runtimesWithNode(24).filter(([runtime]) => runtime === "bun" || nodeMajorVersion() === 24); +const corkRuntimes = runtimesWithNode(24); describe.each(corkRuntimes)("response cork Node oracle (%s)", (_runtime, executable) => { for (const mode of [ "pipeline", From 9cd3575d26f693f030b27df4ffbdc9cf986d0e09 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 24 Oct 2026 00:43:16 -0700 Subject: [PATCH 03/10] test(http): use upstream runtime discovery helpers --- test/js/node/http/node-http-nested-cork.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/test/js/node/http/node-http-nested-cork.test.ts b/test/js/node/http/node-http-nested-cork.test.ts index 1b31ded64c8c..cc57dde3ce62 100644 --- a/test/js/node/http/node-http-nested-cork.test.ts +++ b/test/js/node/http/node-http-nested-cork.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, runtimesWithNode } from "harness"; +import { bunEnv, bunExe, nodeExeAtLeast } from "harness"; async function responseCorkOracle(mode: string) { const assert = require("node:assert/strict"); @@ -498,7 +498,9 @@ async function responseCorkOracle(mode: string) { console.log("ok"); } -const corkRuntimes = runtimesWithNode(24); +const corkRuntimes: Array<[string, string]> = [["bun", bunExe()]]; +const corkNode = nodeExeAtLeast("24.0.0"); +if (corkNode) corkRuntimes.push(["node", corkNode]); describe.each(corkRuntimes)("response cork Node oracle (%s)", (_runtime, executable) => { for (const mode of [ "pipeline", From 18104a5ead4eefeb868256fe8e8a6ce4b1defc19 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 27 Oct 2026 19:36:30 -0700 Subject: [PATCH 04/10] fix(http): retain AOT responses in the socket write queue --- src/js/ahead-of-time/node/_http_server.ts | 346 +++++++++- src/js/internal/streams/utils.ts | 2 + src/js/internal/streams/writable.ts | 3 +- src/js/node/_http_server.ts | 6 +- .../http/node-http-cork-oracle.fixture.ts | 518 +++++++++++++++ .../node/http/node-http-nested-cork.test.ts | 604 +++--------------- 6 files changed, 913 insertions(+), 566 deletions(-) create mode 100644 test/js/node/http/node-http-cork-oracle.fixture.ts diff --git a/src/js/ahead-of-time/node/_http_server.ts b/src/js/ahead-of-time/node/_http_server.ts index aa946e2a188e..9a9cac200d88 100644 --- a/src/js/ahead-of-time/node/_http_server.ts +++ b/src/js/ahead-of-time/node/_http_server.ts @@ -1,11 +1,12 @@ // What the compiler reads in place of src/js/node/_http_server.ts, which `bun run` evaluates. The two do the same. -// Written against that file with the SHA-256 ff5fd0cf3ffd5e9a3fb246c83dbbefe854e635a64472500199657149d7819c82. +// Written against that file with the SHA-256 e412f206fd25709a071fccffe7729804615e7a66c3cdd63e2b386052a6bb8d34. // When that file changes, bring the change here and write the new hash: test/internal/second-texts-of-modules.test.ts. // Hardcoded module "node:_http_server" const EventEmitter = require("node:events"); const { Stream } = require("node:stream"); const { Stream: StreamOfLibrary } = require("internal/streams/legacy"); +const { kWriteOperation, kWriteOperationEncoding } = require("internal/streams/utils"); const StreamPrototype = Stream.prototype; // A name read from `plainStream` is looked up from Stream.prototype on. No constructor runs: the other text makes no stream here. function PlainStream() {} @@ -28,6 +29,174 @@ const { asyncContext, } = require("internal/process/next_tick_queue"); let statusMessageLastChecked = ""; +const kResponseSocketWrite = Symbol("responseSocketWrite"); +const kResponseQueueInitialized = Symbol("responseQueueInitialized"); +const kResponseQueueNeedDrain = Symbol("responseQueueNeedDrain"); +const kResponseHeadersQueued = Symbol("responseHeadersQueued"); +const kResumeResponseSocketWrite = Symbol("resumeResponseSocketWrite"); +const kResponseQueuedBodyBytes = Symbol("responseQueuedBodyBytes"); + +function hasResponseSocketWork(socket: any) { + const state = socket._writableState; + return !!(state && (state.corked || state.length || state.writing)); +} + +function shouldQueueResponseWrite(response: any) { + const socket = response.socket; + return !!( + socket && + (hasResponseSocketWork(socket) || (response[kCorked] && willBeChunked(response))) && + NodeHTTPServerSocket && + socket instanceof NodeHTTPServerSocket && + response[kHandle] + ); +} + +function onResponseSocketDrain(this: any) { + const response = this._httpMessage; + if (response) flushWriteAccountingNT(response); +} + +function freezeBufferedResponseHeaders(res: any, fromEnd: boolean) { + if (res[headerStateSymbol] !== NodeHTTPHeaderState.none) return; + if (fromEnd) res[kImplicitHeaderFromEnd] = true; + try { + res._implicitHeader(); + } finally { + if (fromEnd) res[kImplicitHeaderFromEnd] = false; + } +} + +function queueResponseSocketWrite(response: any, kind: string, chunk: any, encoding: any, callback: any) { + const headerState = response[headerStateSymbol]; + let strict; + let length = 0; + if (kind === "write" || (kind === "end" && chunk != null) || kind === "raw") { + if (chunk === null) throw $ERR_STREAM_NULL_VALUES(); + if (typeof chunk === "string") { + if (encoding && !Buffer.isEncoding(encoding)) throw $ERR_UNKNOWN_ENCODING(encoding); + length = Buffer.byteLength(chunk, encoding); + } else if (require("node:util/types").isUint8Array(chunk)) { + length = chunk.byteLength; + } else { + throw $ERR_INVALID_ARG_TYPE("chunk", ["string", "Buffer", "Uint8Array"], chunk); + } + } + let bodyLength = kind === "write" || kind === "end" ? length : 0; + if (kind !== "raw" && kind !== "continue") { + const implicitEnd = kind === "end" && response[headerStateSymbol] === NodeHTTPHeaderState.none; + if (implicitEnd && !response.hasHeader("content-length")) response._contentLength = length; + freezeBufferedResponseHeaders(response, kind === "end"); + if (!response[kResponseHeadersQueued] && response._contentLength == null && response.hasHeader("content-length")) { + const declared = +response.getHeader("content-length"); + if (!NumberIsNaN(declared)) response._contentLength = declared; + } + if (kind === "write" || kind === "end") strict = strictContentLength(response, headerState, kind === "end"); + if (!response._hasBody && length) { + if (response[kRejectNonStandardBodyWrites]) throw $ERR_HTTP_BODY_NOT_ALLOWED(); + chunk = kind === "write" ? "" : undefined; + length = 0; + bodyLength = 0; + } + if (kind === "write" || kind === "end") { + const pending = response[kResponseQueuedBodyBytes] ?? 0; + if (strict !== undefined) { + const actual = response[kHandle].getBytesWritten() + pending + bodyLength; + if (kind === "end" ? actual !== strict : actual > strict) + throw $ERR_HTTP_CONTENT_LENGTH_MISMATCH(actual, strict); + } + response[kResponseQueuedBodyBytes] = pending + bodyLength; + } + const chunked = + willBeChunked(response) && + !(implicitEnd && !response.hasHeader("transfer-encoding") && !response._removedContLen); + if (length && chunked) length += length.toString(16).length + 4; + if (kind === "end" && chunked) length += 5 + (response._trailer?.length ?? 0); + if (response[headerStateSymbol] !== NodeHTTPHeaderState.sent && !response[kResponseHeadersQueued]) { + length += snapshotResponseSocketHeaders(response, kind === "end"); + } + } else if (kind === "continue") { + length = 25; + } + const socket = response.socket; + if (!socket[kResponseQueueInitialized]) { + socket[kResponseQueueInitialized] = true; + socket.on("drain", onResponseSocketDrain); + } + const responseCorked = kind === "write" && !!response[kCorked] && willBeChunked(response); + const result = socket[kWriteOperation]( + { response, kind, chunk, encoding, length, bodyLength, strict, responseCorked }, + callback, + ); + if (!result && kind === "write" && response._hasBody) response[kResponseQueueNeedDrain] = true; + return kind === "write" && !response._hasBody ? true : result; +} + +function executeResponseSocketWrite(operation: any, callback: any) { + const { response, kind, chunk, encoding } = operation; + if (kind === "write" || kind === "end") response[kResponseQueuedBodyBytes] -= operation.bodyLength; + if (kind === "write") { + (ServerResponsePrototypeWrite as Function).$call( + response, + chunk, + encoding, + callback, + kResponseSocketWrite, + operation, + ); + } else if (kind === "end") { + (ServerResponsePrototypeEnd as Function).$call( + response, + chunk, + encoding, + callback, + kResponseSocketWrite, + operation, + ); + } else { + const handle = response[kHandle]; + if (!handle || (handle.flags & NodeHTTPResponseFlags.closed_or_completed) !== 0) { + callback(response.errored ?? $ERR_STREAM_DESTROYED("write")); + return; + } + if (kind === "raw") handle.writeInformational(chunk, encoding); + else if (kind === "continue") handle.writeContinue(); + else (ServerResponsePrototypeFlushHeaders as Function).$call(response, kResponseSocketWrite); + if (handle.bufferedAmount > 0) { + addPendingCallback(response, callback); + handle.onwritable = allowWritesToContinue.bind(response); + } else callback(); + } +} + +function snapshotResponseSocketHeaders(res: any, fromEnd: boolean) { + const flat = renderNativeHeaders(res, res.$exchange, res.$exchange?.response === res) as string[]; + const headers = { flat: flat.slice(), autoHeaders: renderedAutoHeaders, keepAliveSecs: renderedKeepAliveSecs }; + res[kResponseHeadersQueued] = headers; + let bytes = + 15 + String(res[kSnapshotStatusMessage] ?? res.statusMessage ?? STATUS_CODES[res.statusCode] ?? "unknown").length; + for (let i = 0; i < flat.length; i += 2) { + if (flat[i] !== "\u0000") bytes += flat[i].length + flat[i + 1].length + 4; + } + if (headers.autoHeaders & AUTO_HEADER_DATE) bytes += 37; + if (headers.autoHeaders & AUTO_HEADER_CONN_KEEP_ALIVE) bytes += 24; + if (headers.autoHeaders & AUTO_HEADER_CONN_CLOSE) bytes += 19; + if (headers.autoHeaders & AUTO_HEADER_KEEP_ALIVE_TIMEOUT) bytes += 22 + String(headers.keepAliveSecs).length; + if (headers.autoHeaders & AUTO_HEADER_TRANSFER_ENCODING_CHUNKED) bytes += 28; + else if (!res.hasHeader("content-length") && !res.hasHeader("transfer-encoding") && res._hasBody) { + if (fromEnd && !res._removedContLen && !res._removedTE) bytes += 18 + String(res._contentLength).length; + else if (willBeChunked(res)) bytes += 28; + } + return bytes + 2; +} + +function renderResponseSocketHeaders(res: any) { + const headers = res[kResponseHeadersQueued]; + renderedAutoHeaders = headers.autoHeaders; + renderedKeepAliveSecs = headers.keepAliveSecs; + return headers.flat; +} + const kFlowing = 1 << 24; const kHasPaused = 1 << 25; const kPaused = 1 << 26; @@ -703,6 +872,10 @@ declare global { [kImplicitHeaderFromEnd]?: boolean; [kReplayingPipelinedOps]?: boolean; [kPendingFinish]?: HttpWriteCallback | null; + [kResponseQueuedBodyBytes]?: number; + [kResponseHeadersQueued]?: { flat: string[]; autoHeaders: number; keepAliveSecs: number }; + [kResumeResponseSocketWrite]?: () => void; + [kResponseQueueNeedDrain]?: boolean; writeHead( statusCode: number, statusMessage?: string | HttpHeadersInit | null, @@ -2442,6 +2615,7 @@ function getNodeHTTPServerSocket(): NodeHTTPServerSocketConstructor { // connection whose pipelined responses have buffered past the high water mark. _paused = false; #pendingCallback: $Node.ErrorCallback | null = null; + #pendingResponseCallback: $Node.ErrorCallback | null = null; #pendingAbortMessage: $Node.ServerResponse | null | undefined; #closeHandled = false; #peer: RememberedPeer | undefined; @@ -2552,7 +2726,6 @@ function getNodeHTTPServerSocket(): NodeHTTPServerSocketConstructor { this.#pendingCallback = null; callback(); } - this.emit("drain"); } #onData(chunk: Buffer | undefined, last: boolean) { this._unrefTimer(); @@ -2711,6 +2884,7 @@ function getNodeHTTPServerSocket(): NodeHTTPServerSocketConstructor { this.destroy(this.#closeError); } // A tunnel write that waited for a drain fails now. After destroy(), it settles the callbacks of Writable and emits no 'error'. + this.#pendingResponseCallback?.(writeFailure); const pendingCallback = this.#pendingCallback; if (pendingCallback) { this.#pendingCallback = null; @@ -2798,6 +2972,16 @@ function getNodeHTTPServerSocket(): NodeHTTPServerSocketConstructor { this.#closeHandle(handle, callback, err); } + cork() { + flushSocketHeldChunk(this); + return super.cork(); + } + uncork() { + const result = super.uncork(); + const response = this._httpMessage as $Node.ServerResponse | null; + if (!hasResponseSocketWork(this) && response) scheduleWriteAccountingFlush(response); + return result; + } _final(callback: $Node.ErrorCallback) { flushSocketHeldChunk(this); const handle = this[kHandle]; @@ -3003,26 +3187,53 @@ function getNodeHTTPServerSocket(): NodeHTTPServerSocketConstructor { return this; } - _write(_chunk: $Node.HttpChunk, _encoding: string, _callback: $Node.ErrorCallback) { + _write(_chunk: any, _encoding: any, _callback: $Node.ErrorCallback) { + if (_encoding === kWriteOperationEncoding) { + const response = _chunk.response; + const done = (error?: Error) => { + if (this.#pendingResponseCallback !== done) return; + this.#pendingResponseCallback = null; + response[kResumeResponseSocketWrite] = undefined; + _callback(error); + if (!hasResponseSocketWork(this)) scheduleWriteAccountingFlush(response); + }; + this.#pendingResponseCallback = done; + const resume = () => { + response[kResumeResponseSocketWrite] = undefined; + try { + executeResponseSocketWrite(_chunk, done); + } catch (error) { + done(error); + } + }; + if (_chunk.responseCorked && response[kCorked]) response[kResumeResponseSocketWrite] = resume; + else resume(); + return; + } flushSocketHeldChunk(this); const handle = this[kHandle]; + const done = (error?: Error) => { + _callback(error); + const response = this._httpMessage as $Node.ServerResponse | null; + if (response && !hasResponseSocketWork(this)) scheduleWriteAccountingFlush(response); + }; let err; this._unrefTimer(); try { if (handle) { + handle.ondrain ??= this.#onDrain.bind(this); const flushed = handle.write(_chunk, _encoding); if (!flushed && handle.ondrain) { // Streaming mode (CONNECT tunnels): wait for the native drain // callback before completing the write. - this.#pendingCallback = _callback; + this.#pendingCallback = done; return false; } } } catch (e) { err = e; } - if (err) _callback(err); - else _callback(); + done(err); } pause() { @@ -3051,6 +3262,9 @@ function getNodeHTTPServerSocket(): NodeHTTPServerSocketConstructor { return this[kHandle]?.hasQueuedTunnelReads === true; } }; + (NodeHTTPServerSocket.prototype as any)[kWriteOperation] = + require("internal/streams/writable").prototype[kWriteOperation]; + NodeHTTPServerSocket.prototype._writev = undefined; Object.defineProperty(NodeHTTPServerSocket, "name", { value: "Socket" }); serverSocketResume = NodeHTTPServerSocket.prototype.resume; serverSocketSetTimeout = NodeHTTPServerSocket.prototype.setTimeout; @@ -4178,12 +4392,10 @@ function advanceResponsePipeline( const op = ops[i]; const kind = op[0]; if (kind === "raw") { - // Buffered 1xx bytes: route through the same AsyncSocket buffer the - // response's own writeHead/end use so they precede the final response. - handle.writeInformational(op[1]!, op[2]); - if (typeof op[3] === "function") process.nextTick(op[3]); + if (ServerResponsePrototypeWriteRaw.$call(res, op[1]!, op[2], op[3]) === false) hitBackpressure = true; } else if (kind === "write") { - if (ServerResponsePrototypeWrite.$call(res, op[1], op[2], op[3]) === false) hitBackpressure = true; + const chunk = op[1] === undefined && !res._hasBody ? "" : op[1]; + if (ServerResponsePrototypeWrite.$call(res, chunk, op[2], op[3]) === false) hitBackpressure = true; } else { ServerResponsePrototypeEnd.$call(res, op[1], op[2], op[3]); } @@ -4320,6 +4532,7 @@ function bufferPipelinedEnd( addPipelineOutgoingData(queued, bytes); } queued.ended = true; + res[kCorked] = 0; res.finished = true; return res; } @@ -4508,6 +4721,10 @@ Object.defineProperty(ServerResponse.prototype, "socket", { return (this[fakeSocketSymbol] ??= new FakeSocket(this)); }, set(this: $Node.ServerResponse, value: $Node.HttpSocket | null) { + for (let n = 0; n < this[kCorked]; n++) { + value?.cork(); + this[kSocket]?.uncork(); + } this[fakeSocketSymbol] = value; // Keep the OutgoingMessage storage in sync: the standalone no-handle path // routes through OutgoingMessage.write/end, whose kSocket-gated checks @@ -4540,19 +4757,22 @@ Object.defineProperty(ServerResponse.prototype, "chunkedEncoding", { Object.defineProperty(ServerResponse.prototype, "writableCorked", { get(this: $Node.ServerResponse) { - return this.socket?.writableCorked ?? 0; + return this[kCorked]; }, set(_value) {}, }); ServerResponse.prototype.cork = function cork(this: $Node.ServerResponse) { sendWhatIsHeld(this); + this[kCorked]++; // socket is null while this response is queued behind a pipelined response. this.socket?.cork(); }; ServerResponse.prototype.uncork = function uncork(this: $Node.ServerResponse) { + this[kCorked]--; this.socket?.uncork(); + if (!this[kCorked]) this[kResumeResponseSocketWrite]?.(); }; ServerResponse.prototype.setTimeout = function setTimeout( @@ -4633,6 +4853,8 @@ ServerResponse.prototype._writeRaw = function ( return true; }; +const ServerResponsePrototypeWriteRaw = ServerResponse.prototype._writeRaw; + ServerResponse.prototype.writeEarlyHints = function ( this: $Node.ServerResponse, hints: Record, @@ -4737,6 +4959,11 @@ ServerResponse.prototype.writeContinue = function (this: $Node.ServerResponse, c this._sent100 = true; return; } + if (shouldQueueResponseWrite(this)) { + queueResponseSocketWrite(this, "continue", undefined, undefined, cb); + this._sent100 = true; + return; + } const native = this.socket?.[kHandle]?.response; if (native) native.writeContinue(); else this[kHandle]?.writeContinue?.(); @@ -5459,6 +5686,8 @@ ServerResponse.prototype.end = function ( chunk?: $Node.HttpChunk | null | $Node.HttpWriteCallback, encoding?: string | null | $Node.HttpWriteCallback, callback?: $Node.HttpWriteCallback | null, + token: symbol | undefined = undefined, + operation: any = undefined, ) { if ($isCallable(chunk)) { callback = chunk; @@ -5471,6 +5700,7 @@ ServerResponse.prototype.end = function ( callback = undefined; } + const fromSocket = token === kResponseSocketWrite; const exchange = this.$exchange; // The short road is for a response as its class made it: no field of it has become an accessor, so what is read off it below runs no code of the program. let flags = @@ -5638,7 +5868,13 @@ ServerResponse.prototype.end = function ( !(!chunk && handleFlags & NodeHTTPResponseFlags.ended) && !(handleFlags & NodeHTTPResponseFlags.socket_closed) ) { - draining = handle.end(chunk, encoding, undefined, strictContentLength(this, headerState, true)) < 0; + draining = + handle.end( + chunk, + encoding, + undefined, + fromSocket ? operation.strict : strictContentLength(this, headerState, true), + ) < 0; } } this._header = " "; @@ -5794,6 +6030,8 @@ ServerResponse.prototype.write = function ( chunk?: $Node.HttpChunk | null | $Node.HttpWriteCallback, encoding?: string | null | $Node.HttpWriteCallback, callback?: $Node.HttpWriteCallback | null, + token: symbol | undefined = undefined, + operation: any = undefined, ) { const exchange = this.$exchange; const isItself = @@ -6089,10 +6327,11 @@ function writeHeadAndChunk( encoding: string | null | undefined, onWritable: $Node.HttpWriteCallback | null | undefined, strict: number | undefined, + fromSocket = false, ): number { let result = 0; handle.cork(() => { - writeHeadToHandle(res, exchange, handle); + writeHeadToHandle(res, exchange, handle, fromSocket); // If handle.writeHead throws, headersSent stays false. storeHeaderState(res, exchange, NodeHTTPHeaderState.sent); result = handle.write(chunk, encoding, onWritable, strict); @@ -6106,9 +6345,11 @@ function hasSentForCallback( handle: $Node.NodeHTTPResponse | $Node.Http1ResponseHandle, hasBody: boolean, callback: $Node.HttpWriteCallback, + fromSocket = false, ): boolean { handle.flushHeaders(); if (hasBody && handle.bufferedAmount > 0) { + if (!fromSocket) res[kResponseQueueNeedDrain] = true; addPendingCallback(res, callback); handle.onwritable = allowWritesToContinue.bind(res); return false; @@ -6140,7 +6381,11 @@ const ServerResponsePrototypeEnd: $Node.ServerResponse["end"] = ServerResponse.p function flushWriteAccountingNT(res: $Node.ServerResponse) { const exchange = res.$exchange; res[kAccountingFlushScheduled] = false; - const needsDrain = (res[kBytesBuffered] ?? 0) >= res.writableHighWaterMark; + if (exchange !== undefined && exchange.response === res) exchange.isFlushScheduled = false; + if (res.socket && hasResponseSocketWork(res.socket)) return; + if (res[kResponseQueueNeedDrain] && res[kHandle]?.onwritable !== undefined) return; + const needsDrain = res[kResponseQueueNeedDrain] || (res[kBytesBuffered] ?? 0) >= res.writableHighWaterMark; + res[kResponseQueueNeedDrain] = false; res[kBytesBuffered] = 0; if (exchange !== undefined && exchange.response === res) { exchange.isFlushScheduled = false; @@ -6232,6 +6477,10 @@ const OriginalImplicitHeadFn = (ServerResponse.prototype._implicitHeader = funct Object.defineProperty(ServerResponse.prototype, "writableNeedDrain", { get(this: $Node.ServerResponse) { + const socket = this.socket; + if ((socket as any)?.[kResponseQueueInitialized] && hasResponseSocketWork(socket)) { + return !this.destroyed && !this.finished && !!this[kResponseQueueNeedDrain]; + } // 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 ( @@ -6239,6 +6488,7 @@ Object.defineProperty(ServerResponse.prototype, "writableNeedDrain", { !this.finished && ((handle?.bufferedAmount ?? 0) !== 0 || handle?.onwritable !== undefined || + this[kResponseQueueNeedDrain] || bytesBufferedOf(this) >= this.writableHighWaterMark || (this[kPipelinedQueuedState]?.needDrain ?? false)) ); @@ -6259,6 +6509,10 @@ Object.defineProperty(ServerResponse.prototype, "writableFinished", { Object.defineProperty(ServerResponse.prototype, "writableLength", { get(this: $Node.ServerResponse) { if (this.writableFinished) return 0; + const socket = this.socket; + if ((socket as any)?.[kResponseQueueInitialized] && hasResponseSocketWork(socket)) { + return socket!.writableLength + bytesBufferedOf(this); + } sendWhatIsHeld(this); // Bytes handed off this event-loop turn (including chunked framing, like // Node.js's outputData accounting) plus whatever the native handle still @@ -6298,6 +6552,8 @@ ServerResponse.prototype._send = function ( return OutgoingMessagePrototype._send.$apply(this, arguments); } + if (shouldQueueResponseWrite(this)) return queueResponseSocketWrite(this, "write", data, encoding, callback); + const strict = strictContentLength(this, this[headerStateSymbol], false); if (this[headerStateSymbol] !== NodeHTTPHeaderState.sent) { writeHeadAndChunk(this, exchange, handle, data, encoding, callback, strict); @@ -6448,7 +6704,7 @@ function emitResponseFinishedThenClose(res: $Node.ServerResponse, callback: $Nod res.emit("close"); } -ServerResponse.prototype.flushHeaders = function (this: $Node.ServerResponse) { +ServerResponse.prototype.flushHeaders = function (this: $Node.ServerResponse, token: symbol | undefined = undefined) { const exchange = this.$exchange; if (exchange === undefined || exchange.response !== this) spillStateReachedThrough(this); else sendWhatIsHeld(this); @@ -6461,11 +6717,15 @@ ServerResponse.prototype.flushHeaders = function (this: $Node.ServerResponse) { return; } + if (token !== kResponseSocketWrite && shouldQueueResponseWrite(this)) { + queueResponseSocketWrite(this, "flush", undefined, undefined, undefined); + return; + } const handle = this[kHandle]; if (handle) { if (this[headerStateSymbol] === NodeHTTPHeaderState.assigned) { storeHeaderState(this, exchange, NodeHTTPHeaderState.sent); - writeHeadToHandle(this, exchange, handle); + writeHeadToHandle(this, exchange, handle, token === kResponseSocketWrite); } handle.flushHeaders(); } else { @@ -6475,6 +6735,8 @@ ServerResponse.prototype.flushHeaders = function (this: $Node.ServerResponse) { } }; +const ServerResponsePrototypeFlushHeaders = ServerResponse.prototype.flushHeaders; + function updateHasBody(response: $Node.ServerResponse, statusCode: number) { // RFC 2616, 10.2.5: // The 204 response MUST NOT include a message-body, and thus is always @@ -6525,7 +6787,7 @@ function callObservableWriteHead(self: $Node.ServerResponse, fromEnd: boolean | function allowWritesToContinue(this: $Node.ServerResponse) { this._callPendingCallbacks(); - this.emit("drain"); + flushWriteAccountingNT(this); } function failPendingWriteCallbacks(res: $Node.ServerResponse, error: Error | null | undefined) { @@ -6717,8 +6979,11 @@ function writeHeadToHandle( res: $Node.ServerResponse, exchange: Exchange | undefined, handle: $Node.NodeHTTPResponse | $Node.Http1ResponseHandle, + fromSocket = false, ): void { - const renderedHeaders = renderNativeHeaders(res, exchange, exchange !== undefined && exchange.response === res); + const renderedHeaders = fromSocket + ? renderResponseSocketHeaders(res) + : renderNativeHeaders(res, exchange, exchange !== undefined && exchange.response === res); const statusCode = statusCodeToSend(res); const statusMessage = statusMessageToSend(res); handle.writeHead(statusCode, statusMessage, renderedHeaders as string[], renderedAutoHeaders, renderedKeepAliveSecs); @@ -6730,6 +6995,8 @@ function endInFull( chunk: $Node.HttpChunk | null | undefined, encoding: string | null | undefined, callback: $Node.HttpWriteCallback | undefined, + fromSocket = false, + operation: any = undefined, ) { keepNoOfDoor(exchange); if (exchange === undefined || exchange.response !== res) spillStateReachedThrough(res); @@ -6738,7 +7005,7 @@ function endInFull( const droppedChunk = !!chunk && res.destroyed && !res.finished; if (droppedChunk) chunk = undefined; - if (hasServerResponseFinished(res, chunk, callback, true)) { + if (!fromSocket && hasServerResponseFinished(res, chunk, callback, true)) { return res; } @@ -6751,6 +7018,16 @@ function endInFull( return OutgoingMessagePrototype.end.$call(res, chunk, encoding, callback); } + if (!fromSocket && shouldQueueResponseWrite(res)) { + const socket = res.socket!; + queueResponseSocketWrite(res, "end", chunk, encoding, callback); + res.finished = true; + res[kCorked] = 0; + socket._writableState.corked = 1; + socket.uncork(); + res[kResumeResponseSocketWrite]?.(); + return res; + } if (res[headerStateSymbol] === NodeHTTPHeaderState.none) { updateHasBody(res, (res.statusCode = validateStatusCode(res.statusCode))); } @@ -6786,10 +7063,10 @@ function endInFullBehindHead( } let draining = false; if (headerState !== NodeHTTPHeaderState.sent) { - const renderedHeaders = renderNativeHeaders( - res, - exchange, - exchange !== undefined && exchange.response === res, + const renderedHeaders = ( + fromSocket + ? renderResponseSocketHeaders(res) + : renderNativeHeaders(res, exchange, exchange !== undefined && exchange.response === res) ) as string[]; let contentLength: number; try { @@ -6799,7 +7076,7 @@ function endInFullBehindHead( renderedHeaders, chunk, encoding, - strictContentLength(res, headerState, true), + fromSocket ? operation.strict : strictContentLength(res, headerState, true), renderedAutoHeaders, renderedKeepAliveSecs, ); @@ -6814,7 +7091,13 @@ function endInFullBehindHead( res._contentLength = contentLength; } else { if (!(!chunk && flags & NodeHTTPResponseFlags.ended) && !(flags & NodeHTTPResponseFlags.socket_closed)) { - draining = handle.end(chunk, encoding, undefined, strictContentLength(res, headerState, true)) < 0; + draining = + handle.end( + chunk, + encoding, + undefined, + fromSocket ? operation.strict : strictContentLength(res, headerState, true), + ) < 0; } } res._header = " "; @@ -6828,12 +7111,14 @@ function writeInFull( chunk: $Node.HttpChunk | null | undefined, encoding: string | null | undefined, callback: $Node.HttpWriteCallback | undefined, + fromSocket = false, + operation: any = undefined, ) { keepNoOfDoor(exchange); if (exchange === undefined || exchange.response !== res) spillStateReachedThrough(res); const handle = res[kHandle]; - if (hasServerResponseFinished(res, chunk, callback, false)) { + if (!fromSocket && hasServerResponseFinished(res, chunk, callback, false)) { return false; } @@ -6878,23 +7163,24 @@ function writeInFullBehindHead( return false; } - const strict = strictContentLength(res, headerState, false); + const strict = fromSocket ? operation.strict : strictContentLength(res, headerState, false); const hasBody = res._hasBody; const onWritable = hasBody ? allowWritesToContinue.bind(res) : undefined; if (res[headerStateSymbol] !== NodeHTTPHeaderState.sent) { - result = writeHeadAndChunk(res, exchange, handle, chunk, encoding, onWritable, strict); + result = writeHeadAndChunk(res, exchange, handle, chunk, encoding, onWritable, strict, fromSocket); } else { result = handle.write(chunk, encoding, onWritable, strict); } if (result < 0 && hasBody) { + if (!fromSocket) res[kResponseQueueNeedDrain] = true; if (callback) addPendingCallback(res, callback); return false; } res._callPendingCallbacks(); - if (callback && !hasSentForCallback(res, handle, hasBody, callback)) return false; + if (callback && !hasSentForCallback(res, handle, hasBody, callback, fromSocket)) return false; let written = 0; if (chunk) { diff --git a/src/js/internal/streams/utils.ts b/src/js/internal/streams/utils.ts index 1e3343766de7..5d86bc0c7f5b 100644 --- a/src/js/internal/streams/utils.ts +++ b/src/js/internal/streams/utils.ts @@ -33,6 +33,7 @@ const kIsWritable = Symbol.for("nodejs.stream.writable"); const kIsDisturbed = Symbol.for("nodejs.stream.disturbed"); const kWriteOperation = Symbol("kWriteOperation"); +const kWriteOperationEncoding = "bun:write-operation"; const kIsClosedPromise = Symbol.for("nodejs.webstream.isClosedPromise"); const kControllerErrorFunction = Symbol.for("nodejs.webstream.controllerErrorFunction"); @@ -392,6 +393,7 @@ function isErrored(stream) { export default { kWriteOperation, + kWriteOperationEncoding, isDestroyed, kIsDestroyed, isDisturbed, diff --git a/src/js/internal/streams/writable.ts b/src/js/internal/streams/writable.ts index d2e29d5ffbc7..cedf93818ddf 100644 --- a/src/js/internal/streams/writable.ts +++ b/src/js/internal/streams/writable.ts @@ -11,6 +11,7 @@ const eos = require("internal/streams/end-of-stream"); const { addAbortSignal } = require("internal/streams/add-abort-signal"); const streamUtils = require("internal/streams/utils"); const kWriteOperation: typeof streamUtils.kWriteOperation = streamUtils.kWriteOperation; +const kWriteOperationEncoding = streamUtils.kWriteOperationEncoding; const streamState = require("internal/streams/state"); const { WritableState, resetBuffer, isReadableState, isWritableState, flagsOf, addFlags, removeFlags, internals, nop } = streamState; @@ -241,7 +242,7 @@ Writable.prototype[kWriteOperation] = function (operation, callback) { return false; } state.pendingcb!++; - return writeOrBuffer(this, state, operation, kWriteOperation, cb, operation.length); + return writeOrBuffer(this, state, operation, kWriteOperationEncoding, cb, operation.length); }; const writeBytes = function _write(stream: $Node.Writable, state: WritableState, chunk, encoding, cb: WriteCallback) { diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index e53b32e8f0c7..b63c8749a2db 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -2,7 +2,7 @@ const EventEmitter: typeof import("node:events").EventEmitter = require("node:events"); const { Stream } = require("node:stream"); const { isIP } = require("internal/net/isIP"); -const { kWriteOperation } = require("internal/streams/utils"); +const { kWriteOperation, kWriteOperationEncoding } = require("internal/streams/utils"); const { _checkInvalidHeaderChar: checkInvalidHeaderChar, chunkExpression, @@ -1536,7 +1536,7 @@ function queueResponseSocketWrite(response, kind, chunk, encoding, callback) { if (kind === "write" || kind === "end") strict = strictContentLength(response, headerState, kind === "end"); if (!response._hasBody && length) { if (response[kRejectNonStandardBodyWrites]) throw $ERR_HTTP_BODY_NOT_ALLOWED(); - chunk = undefined; + chunk = kind === "write" ? "" : undefined; length = 0; bodyLength = 0; } @@ -2356,7 +2356,7 @@ function getNodeHTTPServerSocket() { } _write(_chunk, _encoding, _callback) { - if (_encoding === kWriteOperation) { + if (_encoding === kWriteOperationEncoding) { const response = _chunk.response; const done = error => { if (this.#pendingResponseCallback !== done) return; diff --git a/test/js/node/http/node-http-cork-oracle.fixture.ts b/test/js/node/http/node-http-cork-oracle.fixture.ts new file mode 100644 index 000000000000..1d83f0f52c75 --- /dev/null +++ b/test/js/node/http/node-http-cork-oracle.fixture.ts @@ -0,0 +1,518 @@ +export async function responseCorkOracle(mode: string) { + const assert = require("node:assert/strict"); + const http = require("node:http"); + const net = require("node:net"); + const { once } = require("node:events"); + const { setImmediate: immediate } = require("node:timers/promises"); + const { pipeline } = require("node:stream/promises"); + if (mode.startsWith("standalone")) { + const chunks: Buffer[] = []; + const finished = Promise.withResolvers(); + const written = Promise.withResolvers(); + let callbacks = 0; + const server = http.createServer(); + server.on("upgrade", (_req, socket) => { + const req = new http.IncomingMessage(socket); + req.httpVersionMajor = 1; + req.httpVersionMinor = 1; + const res = new http.ServerResponse(req); + res.assignSocket(socket); + res.strictContentLength = mode === "standalone-strict"; + res.setHeader("content-length", 2); + const onWrite = error => { + assert.ifError(error); + if (++callbacks === 2) written.resolve(); + }; + res.on("finish", () => { + finished.resolve(); + socket.end(); + }); + res.cork(); + res.write("A", onWrite); + res.write("B", onWrite); + res.end(); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const client = net.connect(server.address().port, "127.0.0.1"); + client.on("data", chunk => chunks.push(chunk)); + try { + await once(client, "connect"); + client.write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\nUpgrade: test\r\n\r\n"); + await once(client, "close"); + await Promise.all([finished.promise, written.promise]); + assert.equal(callbacks, 2); + assert.match(Buffer.concat(chunks).toString(), /^HTTP\/1\.1 200 OK\r\n/); + assert.ok(Buffer.concat(chunks).toString().endsWith("\r\n\r\nAB")); + } finally { + client.destroy(); + server.closeAllConnections(); + await new Promise(resolve => server.close(resolve)); + } + console.log("ok"); + return; + } + const chunks: Buffer[] = []; + const discardsResponse = + mode === "destroy" || mode === "destroy-no-error" || mode === "strict-end" || mode === "strict-first-write"; + const second = Promise.withResolvers(); + const reassigned = Promise.withResolvers(); + const completed = Promise.withResolvers(); + let handled = 0; + let client; + let first: any; + const highWaterMark = mode === "accounting" ? 64 : mode === "native-queued" ? 64 * 1024 * 1024 : 1024; + const server = http.createServer({ highWaterMark }, (req, res) => { + if (req.url === "/barrier") return res.end("barrier"); + handled++; + if (handled === 2) { + assert.equal(res.socket, null); + res.setHeader("content-length", 2); + res.cork(); + if (mode === "reassigned-no-body") { + res.statusCode = 204; + res.write("ignored"); + res.end(); + second.resolve(); + return; + } + if (mode === "reassigned-informational") { + let callbackRan = false; + res.writeContinue(() => (callbackRan = true)); + res.once("socket", () => { + void (async () => { + await barrier(); + assert.equal(Buffer.concat(chunks).includes("100 Continue"), false); + assert.equal(callbackRan, false); + res.uncork(); + res.end("C"); + })().then(reassigned.resolve, reassigned.reject); + }); + res.write("B"); + second.resolve(); + return; + } + res.write("B"); + res.end("C"); + assert.equal(res.writableCorked, 0); + second.resolve(); + return; + } + first = res; + void handle(res).then(completed.resolve, completed.reject); + }); + const counts = res => [res.writableCorked, res.socket.writableCorked]; + const barrier = async () => { + const response = await fetch(`http://127.0.0.1:${server.address().port}/barrier`); + await response.arrayBuffer(); + await immediate(); + }; + async function handle(res) { + if (mode === "informational-pressure") { + res.setHeader("content-length", 2); + const written = Promise.withResolvers(); + let callbacks = 0; + res.cork(); + res.writeEarlyHints({ link: `; rel=preload` }, error => { + if (error) written.reject(error); + else { + callbacks++; + written.resolve(); + } + }); + res.uncork(); + res.end("AB"); + await barrier(); + assert.equal(callbacks, 0); + client.resume(); + await written.promise; + assert.equal(callbacks, 1); + return; + } + if (mode === "bodyless-write") { + res.statusCode = 204; + res.cork(); + const written = new Promise((resolve, reject) => { + assert.equal( + res.write("ignored", error => (error ? reject(error) : resolve())), + true, + ); + }); + res.end(); + await written; + return; + } + if (mode.startsWith("strict-")) { + res.strictContentLength = true; + const contentLength = ["4"]; + res.setHeader("content-length", contentLength); + res.cork(); + if (mode === "strict-no-body") { + res.statusCode = 204; + assert.throws(() => res.write(undefined), { code: "ERR_INVALID_ARG_TYPE" }); + res.end("AB"); + } else if (mode === "strict-mutable-header") { + res.write("AB"); + contentLength[0] = "6"; + res.end("CD"); + } else if (mode === "strict-end") { + assert.throws(() => res.end("AB"), { code: "ERR_HTTP_CONTENT_LENGTH_MISMATCH" }); + res.destroy(); + } else if (mode === "strict-first-write") { + assert.equal(res.write("ABCDE"), true); + assert.throws(() => res.end(), { code: "ERR_HTTP_CONTENT_LENGTH_MISMATCH" }); + res.destroy(); + } else { + assert.equal(res.write("AB"), true); + assert.throws(() => res.write("CDE"), { code: "ERR_HTTP_CONTENT_LENGTH_MISMATCH" }); + res.end("CD"); + } + return; + } + if (mode === "pipeline") { + res.setHeader("content-length", 4 * 1024 * 1024); + res.cork(); + let reads = 0; + const started = Promise.withResolvers(); + const sending = pipeline(async function* () { + for (let i = 0; i < 8; i++) { + reads++; + started.resolve(); + yield Buffer.alloc(512 * 1024, 65); + } + }, res); + // Register failure handling before any assertion can interrupt the source. + sending.catch(() => {}); + await started.promise; + for (let i = 0; i < 4; i++) await immediate(); + await barrier(); + assert.equal(chunks.length, 0); + assert.equal(reads, 1); + assert.equal(res.writableCorked, 1); + assert.equal(res.writableNeedDrain, true); + assert.ok(res.writableLength >= 512 * 1024); + res.uncork(); + await sending; + return; + } + if (mode.startsWith("reassigned")) { + res.setHeader("content-length", 1); + res.cork(); + res.write("A"); + await second.promise; + assert.equal(chunks.length, 0); + res.uncork(); + res.end(); + return; + } + if (mode === "native-raw-drain" || mode === "native-queued") { + const large = Buffer.alloc(32 * 1024 * 1024, 65); + res.setHeader("content-length", large.length + (mode === "native-raw-drain" ? 1 : 0)); + let drains = 0; + res.on("drain", () => drains++); + if (mode === "native-queued") { + res.cork(); + const written = new Promise(resolve => assert.equal(res.write(large, resolve), true)); + res.uncork(); + assert.equal(res.writableNeedDrain, false); + await written; + await immediate(); + assert.equal(drains, 0); + } else { + const drained = once(res, "drain"); + assert.equal(res.write(large), false); + const written = new Promise((resolve, reject) => + res.socket.write("B", error => (error ? reject(error) : resolve())), + ); + await Promise.all([drained, written]); + assert.equal(drains, 1); + } + assert.equal(res.writableNeedDrain, false); + res.end(); + return; + } + if (mode === "raw-drain") { + const large = Buffer.alloc(16 * 1024 * 1024, 66); + res.setHeader("content-length", large.length + 2048); + let drains = 0; + res.on("drain", () => drains++); + assert.equal(res.write(Buffer.alloc(2048, 65)), false); + await new Promise((resolve, reject) => + res.socket.write(large, error => (error ? reject(error) : resolve())), + ); + await immediate(); + assert.equal(res.writableNeedDrain, false); + assert.equal(drains, 1); + res.end(); + return; + } + if (mode === "interleaved" || mode === "after-uncork") { + const large = Buffer.alloc(16 * 1024 * 1024, 65); + res.setHeader("content-length", large.length + (mode === "interleaved" ? 4 : 3)); + res.cork(); + assert.equal(res.write(large), false); + if (mode === "interleaved") res.socket.write("B"); + res.write("C"); + res.uncork(); + assert.equal(res.socket.writableCorked, 0); + assert.ok(res.socket.writableLength > 0); + res.write("D"); + if (mode === "interleaved") res.socket.write("E", () => res.end()); + else res.end("E"); + return; + } + if (mode !== "cross-chunked") res.setHeader("content-length", mode === "high-water" ? 2048 : 2); + if (mode === "accounting") { + res.sendDate = false; + res.cork(); + assert.equal(res.write("A"), true); + assert.equal(res.writableLength, 58); + assert.equal(res.writableNeedDrain, false); + res.end("B"); + return; + } + if (mode === "headers-frozen") { + const value = ["before"]; + res.setHeader("x-proof", value); + res.cork(); + res.write("A"); + value[0] = "after"; + res.end("B"); + return; + } + if (mode === "destroy" || mode === "destroy-no-error") { + const callbacks = []; + const writes = []; + res.cork(); + writes.push( + new Promise(resolve => + res.write("A", error => { + callbacks.push(error?.code); + resolve(); + }), + ), + ); + writes.push( + new Promise(resolve => + res.socket.write("B", error => { + callbacks.push(error?.code); + resolve(); + }), + ), + ); + res.socket.on("error", () => {}); + res.socket.destroy( + mode === "destroy" ? Object.assign(new Error("corked abort"), { code: "CORKED_ABORT" }) : undefined, + ); + await Promise.all(writes); + const code = mode === "destroy" ? "CORKED_ABORT" : "ERR_STREAM_DESTROYED"; + assert.deepEqual(callbacks, [code, code]); + assert.equal(chunks.length, 0); + return; + } + if (mode === "headers" || mode === "informational") { + res.cork(); + if (mode === "headers") { + res.flushHeaders(); + assert.equal(res.headersSent, true); + } else { + res.writeContinue(); + } + res.write("A"); + await barrier(); + assert.equal(chunks.length, 0); + res.uncork(); + } else if (mode === "callbacks") { + const callbacks = []; + res.cork(); + const responseWrite = new Promise(resolve => + res.write("A", () => { + callbacks.push("response"); + resolve(); + }), + ); + const rawWrite = new Promise(resolve => + res.socket.write("B", () => { + callbacks.push("raw"); + resolve(); + }), + ); + res.uncork(); + await Promise.all([responseWrite, rawWrite]); + assert.deepEqual(callbacks, ["response", "raw"]); + res.end(); + return; + } else if (mode === "counts") { + assert.equal(res.write.length, 3); + assert.equal(res.end.length, 3); + assert.deepEqual(counts(res), [0, 0]); + res.cork(); + assert.deepEqual(counts(res), [1, 1]); + res.cork(); + assert.deepEqual(counts(res), [2, 2]); + res.write("A"); + res.uncork(); + assert.deepEqual(counts(res), [1, 1]); + await barrier(); + assert.equal(chunks.length, 0); + res.uncork(); + assert.deepEqual(counts(res), [0, 0]); + } else if (mode === "cross-content-length" || mode === "cross-chunked") { + res.cork(); + res.write("A"); + res.socket.uncork(); + assert.deepEqual(counts(res), [1, 0]); + await barrier(); + if (mode === "cross-chunked") assert.equal(chunks.length, 0); + res.uncork(); + assert.deepEqual(counts(res), [0, 0]); + } else if (mode === "socket-cork") { + res.socket.cork(); + assert.deepEqual(counts(res), [0, 1]); + res.write("A"); + res.uncork(); + assert.deepEqual(counts(res), [-1, 0]); + } else if (mode === "end-corked") { + res.socket.cork(); + res.cork(); + res.cork(); + res.write("A"); + res.end("B"); + assert.deepEqual(counts(res), [0, 0]); + return; + } else if (mode === "high-water") { + res.cork(); + let callbacks = 0; + let drains = 0; + res.on("drain", () => drains++); + assert.equal( + res.write(Buffer.alloc(2048, 65), () => callbacks++), + false, + ); + await barrier(); + assert.equal(chunks.length, 0); + assert.equal(callbacks, 0); + assert.equal(drains, 0); + assert.equal(res.writableNeedDrain, true); + assert.ok(res.writableLength >= 2048); + const drained = once(res, "drain"); + res.uncork(); + await drained; + await immediate(); + assert.equal(res.writableNeedDrain, false); + assert.equal(callbacks, 1); + assert.equal(drains, 1); + res.end(); + return; + } else { + res.cork(); + res.write("A"); + assert.equal(chunks.length, 0); + await new Promise(resolve => (mode === "next-tick" ? process.nextTick(resolve) : setTimeout(resolve, 0))); + res.uncork(); + } + res.end("B"); + } + function parseResponses(bytes) { + const bodies = []; + while (bytes.length) { + const headerEnd = bytes.indexOf("\r\n\r\n"); + assert.ok(headerEnd >= 0); + const header = bytes.subarray(0, headerEnd).toString(); + bytes = bytes.subarray(headerEnd + 4); + if (/^HTTP\/1\.1 1\d\d /.test(header)) continue; + if (/^HTTP\/1\.1 (204|304) /.test(header)) { + bodies.push(Buffer.alloc(0)); + continue; + } + if (/transfer-encoding: chunked/i.test(header)) { + const parts = []; + while (true) { + const lineEnd = bytes.indexOf("\r\n"); + const size = parseInt(bytes.subarray(0, lineEnd).toString(), 16); + assert.ok(Number.isFinite(size)); + bytes = bytes.subarray(lineEnd + 2); + if (!size) { + bytes = bytes.subarray(2); + break; + } + parts.push(bytes.subarray(0, size)); + bytes = bytes.subarray(size + 2); + } + bodies.push(Buffer.concat(parts)); + } else { + const length = Number(/content-length: (\d+)/i.exec(header)?.[1]); + assert.ok(Number.isFinite(length)); + assert.ok(bytes.length >= length); + bodies.push(bytes.subarray(0, length)); + bytes = bytes.subarray(length); + } + } + return bodies; + } + try { + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + client = net.connect(server.address().port, "127.0.0.1"); + if (mode === "informational-pressure") client.pause(); + const closed = new Promise((resolve, reject) => { + client.on("data", chunk => chunks.push(chunk)); + client.on("error", error => { + if (!discardsResponse) reject(error); + }); + client.on("close", resolve); + }); + closed.catch(() => {}); + client.write( + mode.startsWith("reassigned") + ? "GET / HTTP/1.1\r\nHost: localhost\r\n\r\nGET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n" + : "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + ); + await completed.promise; + if (mode === "reassigned-informational") await reassigned.promise; + await closed; + if (mode === "headers-frozen") assert.match(Buffer.concat(chunks).toString(), /x-proof: before\r\n/i); + if (mode === "strict-no-body" || mode === "bodyless-write") { + const bytes = Buffer.concat(chunks).toString(); + assert.match(bytes, /^HTTP\/1.1 204 /); + assert.equal(bytes.slice(bytes.indexOf("\r\n\r\n") + 4), ""); + } else if (!discardsResponse) { + const bodies = parseResponses(Buffer.concat(chunks)); + if (mode.startsWith("reassigned")) + assert.deepEqual( + bodies.map(body => body.toString()), + ["A", mode === "reassigned-no-body" ? "" : "BC"], + ); + else if (mode === "pipeline") assert.deepEqual(bodies, [Buffer.alloc(4 * 1024 * 1024, 65)]); + else if (mode === "high-water") assert.deepEqual(bodies, [Buffer.alloc(2048, 65)]); + else if (mode === "strict-sequence" || mode === "strict-mutable-header") + assert.deepEqual(bodies, [Buffer.from("ABCD")]); + else if (mode === "raw-drain") { + assert.equal(bodies[0].length, 16 * 1024 * 1024 + 2048); + assert.deepEqual(bodies[0].subarray(0, 2048), Buffer.alloc(2048, 65)); + assert.deepEqual(bodies[0].subarray(2048), Buffer.alloc(16 * 1024 * 1024, 66)); + } else if (mode === "native-raw-drain" || mode === "native-queued") { + assert.equal(bodies[0].length, 32 * 1024 * 1024 + (mode === "native-raw-drain" ? 1 : 0)); + assert.equal(bodies[0].subarray(-1).toString(), mode === "native-raw-drain" ? "B" : "A"); + } else if (mode === "interleaved" || mode === "after-uncork") { + assert.equal(bodies[0].subarray(16 * 1024 * 1024).toString(), mode === "interleaved" ? "BCDE" : "CDE"); + } else + assert.deepEqual( + bodies.map(body => body.toString()), + ["AB"], + ); + } + } finally { + client?.destroy(); + first?.destroy(); + server.closeAllConnections(); + await new Promise(resolve => server.close(resolve)); + } + console.log("ok"); +} + +if (import.meta.main) { + await responseCorkOracle(process.argv[2] ?? "counts").catch(error => { + console.error(error); + process.exitCode = 1; + }); +} diff --git a/test/js/node/http/node-http-nested-cork.test.ts b/test/js/node/http/node-http-nested-cork.test.ts index cc57dde3ce62..2cb0217b2aef 100644 --- a/test/js/node/http/node-http-nested-cork.test.ts +++ b/test/js/node/http/node-http-nested-cork.test.ts @@ -1,541 +1,49 @@ import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, nodeExeAtLeast } from "harness"; - -async function responseCorkOracle(mode: string) { - const assert = require("node:assert/strict"); - const http = require("node:http"); - const net = require("node:net"); - const { once } = require("node:events"); - const { setImmediate: immediate } = require("node:timers/promises"); - const { pipeline } = require("node:stream/promises"); - if (mode.startsWith("standalone")) { - const chunks: Buffer[] = []; - const finished = Promise.withResolvers(); - const written = Promise.withResolvers(); - let callbacks = 0; - const server = http.createServer(); - server.on("upgrade", (_req, socket) => { - const req = new http.IncomingMessage(socket); - req.httpVersionMajor = 1; - req.httpVersionMinor = 1; - const res = new http.ServerResponse(req); - res.assignSocket(socket); - res.strictContentLength = mode === "standalone-strict"; - res.setHeader("content-length", 2); - const onWrite = error => { - assert.ifError(error); - if (++callbacks === 2) written.resolve(); - }; - res.on("finish", () => { - finished.resolve(); - socket.end(); - }); - res.cork(); - res.write("A", onWrite); - res.write("B", onWrite); - res.end(); - }); - await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); - const client = net.connect(server.address().port, "127.0.0.1"); - client.on("data", chunk => chunks.push(chunk)); - try { - await once(client, "connect"); - client.write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\nUpgrade: test\r\n\r\n"); - await once(client, "close"); - await Promise.all([finished.promise, written.promise]); - assert.equal(callbacks, 2); - assert.match(Buffer.concat(chunks).toString(), /^HTTP\/1\.1 200 OK\r\n/); - assert.ok(Buffer.concat(chunks).toString().endsWith("\r\n\r\nAB")); - } finally { - client.destroy(); - server.closeAllConnections(); - await new Promise(resolve => server.close(resolve)); - } - console.log("ok"); - return; - } - const chunks: Buffer[] = []; - const discardsResponse = - mode === "destroy" || mode === "destroy-no-error" || mode === "strict-end" || mode === "strict-first-write"; - const second = Promise.withResolvers(); - const reassigned = Promise.withResolvers(); - const completed = Promise.withResolvers(); - let handled = 0; - let client; - let first; - const highWaterMark = mode === "accounting" ? 64 : mode === "native-queued" ? 64 * 1024 * 1024 : 1024; - const server = http.createServer({ highWaterMark }, (req, res) => { - if (req.url === "/barrier") return res.end("barrier"); - handled++; - if (handled === 2) { - assert.equal(res.socket, null); - res.setHeader("content-length", 2); - res.cork(); - if (mode === "reassigned-no-body") { - res.statusCode = 204; - res.write("ignored"); - res.end(); - second.resolve(); - return; - } - if (mode === "reassigned-informational") { - let callbackRan = false; - res.writeContinue(() => (callbackRan = true)); - res.once("socket", () => { - void (async () => { - await barrier(); - assert.equal(Buffer.concat(chunks).includes("100 Continue"), false); - assert.equal(callbackRan, false); - res.uncork(); - res.end("C"); - })().then(reassigned.resolve, reassigned.reject); - }); - res.write("B"); - second.resolve(); - return; - } - res.write("B"); - res.end("C"); - assert.equal(res.writableCorked, 0); - second.resolve(); - return; - } - first = res; - void handle(res).then(completed.resolve, completed.reject); - }); - const counts = res => [res.writableCorked, res.socket.writableCorked]; - const barrier = async () => { - const response = await fetch(`http://127.0.0.1:${server.address().port}/barrier`); - await response.arrayBuffer(); - await immediate(); - }; - async function handle(res) { - if (mode === "informational-pressure") { - res.setHeader("content-length", 2); - const written = Promise.withResolvers(); - let callbacks = 0; - res.cork(); - res.writeEarlyHints({ link: `; rel=preload` }, error => { - if (error) written.reject(error); - else { - callbacks++; - written.resolve(); - } - }); - res.uncork(); - res.end("AB"); - await barrier(); - assert.equal(callbacks, 0); - client.resume(); - await written.promise; - assert.equal(callbacks, 1); - return; - } - if (mode.startsWith("strict-")) { - res.strictContentLength = true; - const contentLength = ["4"]; - res.setHeader("content-length", contentLength); - res.cork(); - if (mode === "strict-no-body") { - res.statusCode = 204; - assert.throws(() => res.write(undefined), { code: "ERR_INVALID_ARG_TYPE" }); - res.end("AB"); - } else if (mode === "strict-mutable-header") { - res.write("AB"); - contentLength[0] = "6"; - res.end("CD"); - } else if (mode === "strict-end") { - assert.throws(() => res.end("AB"), { code: "ERR_HTTP_CONTENT_LENGTH_MISMATCH" }); - res.destroy(); - } else if (mode === "strict-first-write") { - assert.equal(res.write("ABCDE"), true); - assert.throws(() => res.end(), { code: "ERR_HTTP_CONTENT_LENGTH_MISMATCH" }); - res.destroy(); - } else { - assert.equal(res.write("AB"), true); - assert.throws(() => res.write("CDE"), { code: "ERR_HTTP_CONTENT_LENGTH_MISMATCH" }); - res.end("CD"); - } - return; - } - if (mode === "pipeline") { - res.setHeader("content-length", 4 * 1024 * 1024); - res.cork(); - let reads = 0; - const started = Promise.withResolvers(); - const sending = pipeline(async function* () { - for (let i = 0; i < 8; i++) { - reads++; - started.resolve(); - yield Buffer.alloc(512 * 1024, 65); - } - }, res); - // Register failure handling before any assertion can interrupt the source. - sending.catch(() => {}); - await started.promise; - for (let i = 0; i < 4; i++) await immediate(); - await barrier(); - assert.equal(chunks.length, 0); - assert.equal(reads, 1); - assert.equal(res.writableCorked, 1); - assert.equal(res.writableNeedDrain, true); - assert.ok(res.writableLength >= 512 * 1024); - res.uncork(); - await sending; - return; - } - if (mode.startsWith("reassigned")) { - res.setHeader("content-length", 1); - res.cork(); - res.write("A"); - await second.promise; - assert.equal(chunks.length, 0); - res.uncork(); - res.end(); - return; - } - if (mode === "native-raw-drain" || mode === "native-queued") { - const large = Buffer.alloc(32 * 1024 * 1024, 65); - res.setHeader("content-length", large.length + (mode === "native-raw-drain" ? 1 : 0)); - let drains = 0; - res.on("drain", () => drains++); - if (mode === "native-queued") { - res.cork(); - const written = new Promise(resolve => assert.equal(res.write(large, resolve), true)); - res.uncork(); - assert.equal(res.writableNeedDrain, false); - await written; - await immediate(); - assert.equal(drains, 0); - } else { - const drained = once(res, "drain"); - assert.equal(res.write(large), false); - const written = new Promise((resolve, reject) => - res.socket.write("B", error => (error ? reject(error) : resolve())), - ); - await Promise.all([drained, written]); - assert.equal(drains, 1); - } - assert.equal(res.writableNeedDrain, false); - res.end(); - return; - } - if (mode === "raw-drain") { - const large = Buffer.alloc(16 * 1024 * 1024, 66); - res.setHeader("content-length", large.length + 2048); - let drains = 0; - res.on("drain", () => drains++); - assert.equal(res.write(Buffer.alloc(2048, 65)), false); - await new Promise((resolve, reject) => res.socket.write(large, error => (error ? reject(error) : resolve()))); - await immediate(); - assert.equal(res.writableNeedDrain, false); - assert.equal(drains, 1); - res.end(); - return; - } - if (mode === "interleaved" || mode === "after-uncork") { - const large = Buffer.alloc(16 * 1024 * 1024, 65); - res.setHeader("content-length", large.length + (mode === "interleaved" ? 4 : 3)); - res.cork(); - assert.equal(res.write(large), false); - if (mode === "interleaved") res.socket.write("B"); - res.write("C"); - res.uncork(); - assert.equal(res.socket.writableCorked, 0); - assert.ok(res.socket.writableLength > 0); - res.write("D"); - if (mode === "interleaved") res.socket.write("E", () => res.end()); - else res.end("E"); - return; - } - if (mode !== "cross-chunked") res.setHeader("content-length", mode === "high-water" ? 2048 : 2); - if (mode === "accounting") { - res.sendDate = false; - res.cork(); - assert.equal(res.write("A"), true); - assert.equal(res.writableLength, 58); - assert.equal(res.writableNeedDrain, false); - res.end("B"); - return; - } - if (mode === "headers-frozen") { - const value = ["before"]; - res.setHeader("x-proof", value); - res.cork(); - res.write("A"); - value[0] = "after"; - res.end("B"); - return; - } - if (mode === "destroy" || mode === "destroy-no-error") { - const callbacks = []; - const writes = []; - res.cork(); - writes.push( - new Promise(resolve => - res.write("A", error => { - callbacks.push(error?.code); - resolve(); - }), - ), - ); - writes.push( - new Promise(resolve => - res.socket.write("B", error => { - callbacks.push(error?.code); - resolve(); - }), - ), - ); - res.socket.on("error", () => {}); - res.socket.destroy( - mode === "destroy" ? Object.assign(new Error("corked abort"), { code: "CORKED_ABORT" }) : undefined, - ); - await Promise.all(writes); - const code = mode === "destroy" ? "CORKED_ABORT" : "ERR_STREAM_DESTROYED"; - assert.deepEqual(callbacks, [code, code]); - assert.equal(chunks.length, 0); - return; - } - if (mode === "headers" || mode === "informational") { - res.cork(); - if (mode === "headers") { - res.flushHeaders(); - assert.equal(res.headersSent, true); - } else { - res.writeContinue(); - } - res.write("A"); - await barrier(); - assert.equal(chunks.length, 0); - res.uncork(); - } else if (mode === "callbacks") { - const callbacks = []; - res.cork(); - const responseWrite = new Promise(resolve => - res.write("A", () => { - callbacks.push("response"); - resolve(); - }), - ); - const rawWrite = new Promise(resolve => - res.socket.write("B", () => { - callbacks.push("raw"); - resolve(); - }), - ); - res.uncork(); - await Promise.all([responseWrite, rawWrite]); - assert.deepEqual(callbacks, ["response", "raw"]); - res.end(); - return; - } else if (mode === "counts") { - assert.equal(res.write.length, 3); - assert.equal(res.end.length, 3); - assert.deepEqual(counts(res), [0, 0]); - res.cork(); - assert.deepEqual(counts(res), [1, 1]); - res.cork(); - assert.deepEqual(counts(res), [2, 2]); - res.write("A"); - res.uncork(); - assert.deepEqual(counts(res), [1, 1]); - await barrier(); - assert.equal(chunks.length, 0); - res.uncork(); - assert.deepEqual(counts(res), [0, 0]); - } else if (mode === "cross-content-length" || mode === "cross-chunked") { - res.cork(); - res.write("A"); - res.socket.uncork(); - assert.deepEqual(counts(res), [1, 0]); - await barrier(); - if (mode === "cross-chunked") assert.equal(chunks.length, 0); - res.uncork(); - assert.deepEqual(counts(res), [0, 0]); - } else if (mode === "socket-cork") { - res.socket.cork(); - assert.deepEqual(counts(res), [0, 1]); - res.write("A"); - res.uncork(); - assert.deepEqual(counts(res), [-1, 0]); - } else if (mode === "end-corked") { - res.socket.cork(); - res.cork(); - res.cork(); - res.write("A"); - res.end("B"); - assert.deepEqual(counts(res), [0, 0]); - return; - } else if (mode === "high-water") { - res.cork(); - let callbacks = 0; - let drains = 0; - res.on("drain", () => drains++); - assert.equal( - res.write(Buffer.alloc(2048, 65), () => callbacks++), - false, - ); - await barrier(); - assert.equal(chunks.length, 0); - assert.equal(callbacks, 0); - assert.equal(drains, 0); - assert.equal(res.writableNeedDrain, true); - assert.ok(res.writableLength >= 2048); - const drained = once(res, "drain"); - res.uncork(); - await drained; - await immediate(); - assert.equal(res.writableNeedDrain, false); - assert.equal(callbacks, 1); - assert.equal(drains, 1); - res.end(); - return; - } else { - res.cork(); - res.write("A"); - assert.equal(chunks.length, 0); - await new Promise(resolve => (mode === "next-tick" ? process.nextTick(resolve) : setTimeout(resolve, 0))); - res.uncork(); - } - res.end("B"); - } - function parseResponses(bytes) { - const bodies = []; - while (bytes.length) { - const headerEnd = bytes.indexOf("\r\n\r\n"); - assert.ok(headerEnd >= 0); - const header = bytes.subarray(0, headerEnd).toString(); - bytes = bytes.subarray(headerEnd + 4); - if (/^HTTP\/1\.1 1\d\d /.test(header)) continue; - if (/^HTTP\/1\.1 (204|304) /.test(header)) { - bodies.push(Buffer.alloc(0)); - continue; - } - if (/transfer-encoding: chunked/i.test(header)) { - const parts = []; - while (true) { - const lineEnd = bytes.indexOf("\r\n"); - const size = parseInt(bytes.subarray(0, lineEnd).toString(), 16); - assert.ok(Number.isFinite(size)); - bytes = bytes.subarray(lineEnd + 2); - if (!size) { - bytes = bytes.subarray(2); - break; - } - parts.push(bytes.subarray(0, size)); - bytes = bytes.subarray(size + 2); - } - bodies.push(Buffer.concat(parts)); - } else { - const length = Number(/content-length: (\d+)/i.exec(header)?.[1]); - assert.ok(Number.isFinite(length)); - assert.ok(bytes.length >= length); - bodies.push(bytes.subarray(0, length)); - bytes = bytes.subarray(length); - } - } - return bodies; - } - try { - server.listen(0, "127.0.0.1"); - await once(server, "listening"); - client = net.connect(server.address().port, "127.0.0.1"); - if (mode === "informational-pressure") client.pause(); - const closed = new Promise((resolve, reject) => { - client.on("data", chunk => chunks.push(chunk)); - client.on("error", error => { - if (!discardsResponse) reject(error); - }); - client.on("close", resolve); - }); - closed.catch(() => {}); - client.write( - mode.startsWith("reassigned") - ? "GET / HTTP/1.1\r\nHost: localhost\r\n\r\nGET /second HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n" - : "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", - ); - await completed.promise; - if (mode === "reassigned-informational") await reassigned.promise; - await closed; - if (mode === "headers-frozen") assert.match(Buffer.concat(chunks).toString(), /x-proof: before\r\n/i); - if (mode === "strict-no-body") { - const bytes = Buffer.concat(chunks).toString(); - assert.match(bytes, /^HTTP\/1.1 204 /); - assert.equal(bytes.slice(bytes.indexOf("\r\n\r\n") + 4), ""); - } else if (!discardsResponse) { - const bodies = parseResponses(Buffer.concat(chunks)); - if (mode.startsWith("reassigned")) - assert.deepEqual( - bodies.map(body => body.toString()), - ["A", mode === "reassigned-no-body" ? "" : "BC"], - ); - else if (mode === "pipeline") assert.deepEqual(bodies, [Buffer.alloc(4 * 1024 * 1024, 65)]); - else if (mode === "high-water") assert.deepEqual(bodies, [Buffer.alloc(2048, 65)]); - else if (mode === "strict-sequence" || mode === "strict-mutable-header") - assert.deepEqual(bodies, [Buffer.from("ABCD")]); - else if (mode === "raw-drain") { - assert.equal(bodies[0].length, 16 * 1024 * 1024 + 2048); - assert.deepEqual(bodies[0].subarray(0, 2048), Buffer.alloc(2048, 65)); - assert.deepEqual(bodies[0].subarray(2048), Buffer.alloc(16 * 1024 * 1024, 66)); - } else if (mode === "native-raw-drain" || mode === "native-queued") { - assert.equal(bodies[0].length, 32 * 1024 * 1024 + (mode === "native-raw-drain" ? 1 : 0)); - assert.equal(bodies[0].subarray(-1).toString(), mode === "native-raw-drain" ? "B" : "A"); - } else if (mode === "interleaved" || mode === "after-uncork") { - assert.equal(bodies[0].subarray(16 * 1024 * 1024).toString(), mode === "interleaved" ? "BCDE" : "CDE"); - } else - assert.deepEqual( - bodies.map(body => body.toString()), - ["AB"], - ); - } - } finally { - client?.destroy(); - first?.destroy(); - server.closeAllConnections(); - await new Promise(resolve => server.close(resolve)); - } - console.log("ok"); -} +import { bunEnv, bunExe, nodeExeAtLeast, tempDir } from "harness"; +import { join } from "node:path"; +import { responseCorkOracle } from "./node-http-cork-oracle.fixture"; +import { unsupported as aotUnsupported } from "../../../aot/supported"; const corkRuntimes: Array<[string, string]> = [["bun", bunExe()]]; const corkNode = nodeExeAtLeast("24.0.0"); if (corkNode) corkRuntimes.push(["node", corkNode]); +const corkModes = [ + "pipeline", + "counts", + "bodyless-write", + "cross-content-length", + "cross-chunked", + "socket-cork", + "end-corked", + "high-water", + "next-tick", + "timer", + "destroy", + "destroy-no-error", + "headers", + "informational", + "callbacks", + "reassigned", + "interleaved", + "after-uncork", + "accounting", + "headers-frozen", + "raw-drain", + "reassigned-informational", + "strict-end", + "strict-sequence", + "strict-first-write", + "strict-no-body", + "strict-mutable-header", + "reassigned-no-body", + "native-raw-drain", + "native-queued", + "standalone", + "standalone-strict", + "informational-pressure", +]; describe.each(corkRuntimes)("response cork Node oracle (%s)", (_runtime, executable) => { - for (const mode of [ - "pipeline", - "counts", - "cross-content-length", - "cross-chunked", - "socket-cork", - "end-corked", - "high-water", - "next-tick", - "timer", - "destroy", - "destroy-no-error", - "headers", - "informational", - "callbacks", - "reassigned", - "interleaved", - "after-uncork", - "accounting", - "headers-frozen", - "raw-drain", - "reassigned-informational", - "strict-end", - "strict-sequence", - "strict-first-write", - "strict-no-body", - "strict-mutable-header", - "reassigned-no-body", - "native-raw-drain", - "native-queued", - "standalone", - "standalone-strict", - "informational-pressure", - ]) { + for (const mode of corkModes) { test.concurrent(mode, async () => { await using child = Bun.spawn({ cmd: [ @@ -871,3 +379,35 @@ describe.concurrent("cork buffer: no cross-socket data bleed", () => { `); }); }); + +test.skipIf(!!aotUnsupported)("compiled responses retain corked socket operations", async () => { + using dir = tempDir("compiled-response-cork", { + "main.ts": await Bun.file(join(import.meta.dir, "node-http-cork-oracle.fixture.ts")).text(), + }); + const executable = join(String(dir), process.platform === "win32" ? "app.exe" : "app"); + await using build = Bun.spawn({ + cmd: [bunExe(), "build", "--compile", "--aot", "main.ts", "--outfile", executable], + cwd: String(dir), + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [buildOut, buildError, buildExit] = await Promise.all([build.stdout.text(), build.stderr.text(), build.exited]); + expect({ buildOut: buildExit ? buildOut : "", buildError: buildExit ? buildError : "", buildExit }).toEqual({ + buildOut: "", + buildError: "", + buildExit: 0, + }); + for (const mode of corkModes) { + await using proc = Bun.spawn({ + cmd: [executable, mode], + cwd: String(dir), + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + timeout: 10000, + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ mode, stdout, stderr, exitCode }).toEqual({ mode, stdout: "ok\n", stderr: "", exitCode: 0 }); + } +}); From e6f336ba00d9676b08837c45bb495fd9ade2a3be Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 27 Oct 2026 20:26:22 -0700 Subject: [PATCH 05/10] fix(http): mirror cork state export in AOT module --- src/js/ahead-of-time/node/_http_outgoing.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/js/ahead-of-time/node/_http_outgoing.ts b/src/js/ahead-of-time/node/_http_outgoing.ts index 75c54845f9cc..c7dacc5649fc 100644 --- a/src/js/ahead-of-time/node/_http_outgoing.ts +++ b/src/js/ahead-of-time/node/_http_outgoing.ts @@ -1,5 +1,5 @@ // What the compiler reads in place of src/js/node/_http_outgoing.ts, which `bun run` evaluates. The two do the same. -// Written against that file with the SHA-256 7eb68379dcc85e37da343c90d1da52f21b0bd91937cea041c06b56a6db7551fc. +// Written against that file with the SHA-256 844b8c71bb274a66021c5d1bf7478198b9f0155beb008519d5a1618352888462. // When that file changes, bring the change here and write the new hash: test/internal/second-texts-of-modules.test.ts. // Hardcoded module "node:_http_outgoing" @@ -2427,6 +2427,7 @@ ObjectDefineProperty(OutgoingMessage.prototype, "_headerNames", { }); export default { + kCorked, kHighWaterMark, kUniqueHeaders, kOutHeaders, From 680b964166fd162785b29e85071f4a77349f868b Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 03:31:46 +0000 Subject: [PATCH 06/10] [autofix.ci] apply automated fixes --- test/js/node/http/node-http-nested-cork.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/js/node/http/node-http-nested-cork.test.ts b/test/js/node/http/node-http-nested-cork.test.ts index 2cb0217b2aef..4c28e0c9cbe6 100644 --- a/test/js/node/http/node-http-nested-cork.test.ts +++ b/test/js/node/http/node-http-nested-cork.test.ts @@ -1,8 +1,8 @@ import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe, nodeExeAtLeast, tempDir } from "harness"; import { join } from "node:path"; -import { responseCorkOracle } from "./node-http-cork-oracle.fixture"; import { unsupported as aotUnsupported } from "../../../aot/supported"; +import { responseCorkOracle } from "./node-http-cork-oracle.fixture"; const corkRuntimes: Array<[string, string]> = [["bun", bunExe()]]; const corkNode = nodeExeAtLeast("24.0.0"); From b0c77fadf7069ab2ffaa415276df184c8b86db87 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 4 Nov 2026 20:27:32 -0800 Subject: [PATCH 07/10] fix(http): synchronize implicit end framing in AOT responses --- src/js/ahead-of-time/node/_http_server.ts | 10 ++++++++-- test/js/node/http/node-http-cork-oracle.fixture.ts | 10 ++++++++++ test/js/node/http/node-http-nested-cork.test.ts | 1 + 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/src/js/ahead-of-time/node/_http_server.ts b/src/js/ahead-of-time/node/_http_server.ts index 9a9cac200d88..6494a2ab7d0b 100644 --- a/src/js/ahead-of-time/node/_http_server.ts +++ b/src/js/ahead-of-time/node/_http_server.ts @@ -59,11 +59,17 @@ function onResponseSocketDrain(this: any) { function freezeBufferedResponseHeaders(res: any, fromEnd: boolean) { if (res[headerStateSymbol] !== NodeHTTPHeaderState.none) return; - if (fromEnd) res[kImplicitHeaderFromEnd] = true; + if (fromEnd) { + storeRarely(res, kImplicitHeaderFromEnd, true); + followResponse(res.$exchange); + } try { res._implicitHeader(); } finally { - if (fromEnd) res[kImplicitHeaderFromEnd] = false; + if (fromEnd) { + storeRarely(res, kImplicitHeaderFromEnd, false); + followResponse(res.$exchange); + } } } diff --git a/test/js/node/http/node-http-cork-oracle.fixture.ts b/test/js/node/http/node-http-cork-oracle.fixture.ts index 1d83f0f52c75..117ed727ea83 100644 --- a/test/js/node/http/node-http-cork-oracle.fixture.ts +++ b/test/js/node/http/node-http-cork-oracle.fixture.ts @@ -107,6 +107,11 @@ export async function responseCorkOracle(mode: string) { await immediate(); }; async function handle(res) { + if (mode === "implicit-end") { + res.cork(); + res.end("AB"); + return; + } if (mode === "informational-pressure") { res.setHeader("content-length", 2); const written = Promise.withResolvers(); @@ -471,6 +476,11 @@ export async function responseCorkOracle(mode: string) { if (mode === "reassigned-informational") await reassigned.promise; await closed; if (mode === "headers-frozen") assert.match(Buffer.concat(chunks).toString(), /x-proof: before\r\n/i); + if (mode === "implicit-end") { + const headers = Buffer.concat(chunks).toString().split("\r\n\r\n", 1)[0]; + assert.match(headers, /content-length: 2/i); + assert.doesNotMatch(headers, /transfer-encoding/i); + } if (mode === "strict-no-body" || mode === "bodyless-write") { const bytes = Buffer.concat(chunks).toString(); assert.match(bytes, /^HTTP\/1.1 204 /); diff --git a/test/js/node/http/node-http-nested-cork.test.ts b/test/js/node/http/node-http-nested-cork.test.ts index 4c28e0c9cbe6..12ac2d48a6b6 100644 --- a/test/js/node/http/node-http-nested-cork.test.ts +++ b/test/js/node/http/node-http-nested-cork.test.ts @@ -15,6 +15,7 @@ const corkModes = [ "cross-chunked", "socket-cork", "end-corked", + "implicit-end", "high-water", "next-tick", "timer", From 7aa4b3db361372f51a654d20d197c2f62d6401a8 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 7 Nov 2026 14:37:21 -0800 Subject: [PATCH 08/10] fix(http): adapt queued responses to rewritten AOT exchanges --- src/js/ahead-of-time/node/_http_server.ts | 52 ++++++++++++++++++----- 1 file changed, 42 insertions(+), 10 deletions(-) diff --git a/src/js/ahead-of-time/node/_http_server.ts b/src/js/ahead-of-time/node/_http_server.ts index 6494a2ab7d0b..496d6ae03e5e 100644 --- a/src/js/ahead-of-time/node/_http_server.ts +++ b/src/js/ahead-of-time/node/_http_server.ts @@ -4851,6 +4851,7 @@ ServerResponse.prototype._writeRaw = function ( return queued.bytes < this.writableHighWaterMark; } sendWhatIsHeld(this); + if (shouldQueueResponseWrite(this)) return queueResponseSocketWrite(this, "raw", chunk, encoding, callback); // Write through the response handle's AsyncSocket buffer (same path as // writeHead/end) so 1xx lines share ordering with the final response bytes; // socket.write() would land in the socket handle's separate stream buffer. @@ -5708,6 +5709,10 @@ ServerResponse.prototype.end = function ( const fromSocket = token === kResponseSocketWrite; const exchange = this.$exchange; + if (fromSocket || shouldQueueResponseWrite(this)) { + sendWhatIsHeld(this); + return endInFull(this, exchange, chunk, encoding, callback, fromSocket, operation); + } // The short road is for a response as its class made it: no field of it has become an accessor, so what is read off it below runs no code of the program. let flags = exchange !== undefined && @@ -5874,13 +5879,7 @@ ServerResponse.prototype.end = function ( !(!chunk && handleFlags & NodeHTTPResponseFlags.ended) && !(handleFlags & NodeHTTPResponseFlags.socket_closed) ) { - draining = - handle.end( - chunk, - encoding, - undefined, - fromSocket ? operation.strict : strictContentLength(this, headerState, true), - ) < 0; + draining = handle.end(chunk, encoding, undefined, strictContentLength(this, headerState, true)) < 0; } } this._header = " "; @@ -6040,6 +6039,21 @@ ServerResponse.prototype.write = function ( operation: any = undefined, ) { const exchange = this.$exchange; + const fromSocket = token === kResponseSocketWrite; + if (fromSocket || shouldQueueResponseWrite(this)) { + if ($isCallable(chunk)) { + callback = chunk; + chunk = undefined; + encoding = undefined; + } else if ($isCallable(encoding)) { + callback = encoding; + encoding = undefined; + } else if (!$isCallable(callback)) { + callback = undefined; + } + sendWhatIsHeld(this); + return writeInFull(this, exchange, chunk, encoding, callback, fromSocket, operation); + } const isItself = exchange !== undefined && exchange.response === this && @@ -7050,7 +7064,18 @@ function endInFull( const headerState = res[headerStateSymbol]; if (!droppedChunk) callWriteHeadIfObservable(res, headerState, true); - return endInFullBehindHead(res, exchange, handle, chunk, encoding, callback, droppedChunk, headerState); + return endInFullBehindHead( + res, + exchange, + handle, + chunk, + encoding, + callback, + droppedChunk, + headerState, + fromSocket, + operation, + ); } function endInFullBehindHead( @@ -7062,6 +7087,8 @@ function endInFullBehindHead( callback: $Node.HttpWriteCallback | undefined, droppedChunk: boolean, headerState: NodeHTTPHeaderState, + fromSocket = false, + operation: any = undefined, ) { const flags = handle.flags; if (!!(flags & NodeHTTPResponseFlags.closed_or_completed)) { @@ -7133,13 +7160,16 @@ function writeInFull( return bufferPipelinedWrite(res, queuedState, chunk, encoding, callback); } + if (!fromSocket && handle && shouldQueueResponseWrite(res)) { + return queueResponseSocketWrite(res, "write", chunk, encoding, callback); + } const headerState = res[headerStateSymbol]; callWriteHeadIfObservable(res, headerState); if (!handle) { return OutgoingMessagePrototype.write.$call(res, chunk!, encoding, callback); } - return writeInFullBehindHead(res, exchange, handle, chunk, encoding, callback, headerState); + return writeInFullBehindHead(res, exchange, handle, chunk, encoding, callback, headerState, fromSocket, operation); } function writeInFullBehindHead( @@ -7150,6 +7180,8 @@ function writeInFullBehindHead( encoding: string | null | undefined, callback: $Node.HttpWriteCallback | undefined, headerState: NodeHTTPHeaderState, + fromSocket = false, + operation: any = undefined, ) { let result = 0; @@ -7195,7 +7227,7 @@ function writeInFullBehindHead( written += written.toString(16).length + 4; } } - return countWritten(res, exchange, written); + return countWritten(res, exchange, fromSocket ? 0 : written); } function countWritten(res: $Node.ServerResponse, exchange: Exchange | undefined, written: number): boolean { From fe56ba46911c0339d621a2760a960f0db5723bd5 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 7 Nov 2026 15:01:13 -0800 Subject: [PATCH 09/10] fix(http): retain drain obligations on compiled native writes --- src/js/ahead-of-time/node/_http_server.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/js/ahead-of-time/node/_http_server.ts b/src/js/ahead-of-time/node/_http_server.ts index 496d6ae03e5e..a9f2cab13508 100644 --- a/src/js/ahead-of-time/node/_http_server.ts +++ b/src/js/ahead-of-time/node/_http_server.ts @@ -6095,7 +6095,11 @@ ServerResponse.prototype.write = function ( } if (result <= 0) { ("unlikely"); - if (result < 0 || (handle.flags & NodeHTTPResponseFlags.closed_or_completed) !== 0) return false; + if (result < 0) { + this[kResponseQueueNeedDrain] = true; + return false; + } + if ((handle.flags & NodeHTTPResponseFlags.closed_or_completed) !== 0) return false; } connection = socket; hasChunk = true; @@ -6274,6 +6278,7 @@ ServerResponse.prototype.write = function ( if (result < 0 && hasBody) { ("unlikely"); + this[kResponseQueueNeedDrain] = true; if (callback) addPendingCallback(this, callback); return false; } From b48f2edc679b89912fb8d7582db370148cc2f249 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 7 Nov 2026 15:06:59 -0800 Subject: [PATCH 10/10] fix(http): use current AOT header flags for queued accounting --- src/js/ahead-of-time/node/_http_server.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/js/ahead-of-time/node/_http_server.ts b/src/js/ahead-of-time/node/_http_server.ts index a9f2cab13508..880c4a1e38a3 100644 --- a/src/js/ahead-of-time/node/_http_server.ts +++ b/src/js/ahead-of-time/node/_http_server.ts @@ -184,11 +184,11 @@ function snapshotResponseSocketHeaders(res: any, fromEnd: boolean) { for (let i = 0; i < flat.length; i += 2) { if (flat[i] !== "\u0000") bytes += flat[i].length + flat[i + 1].length + 4; } - if (headers.autoHeaders & AUTO_HEADER_DATE) bytes += 37; - if (headers.autoHeaders & AUTO_HEADER_CONN_KEEP_ALIVE) bytes += 24; - if (headers.autoHeaders & AUTO_HEADER_CONN_CLOSE) bytes += 19; - if (headers.autoHeaders & AUTO_HEADER_KEEP_ALIVE_TIMEOUT) bytes += 22 + String(headers.keepAliveSecs).length; - if (headers.autoHeaders & AUTO_HEADER_TRANSFER_ENCODING_CHUNKED) bytes += 28; + if (headers.autoHeaders & AutoHeaders.date) bytes += 37; + if (headers.autoHeaders & AutoHeaders.conn_keep_alive) bytes += 24; + if (headers.autoHeaders & AutoHeaders.conn_close) bytes += 19; + if (headers.autoHeaders & AutoHeaders.keep_alive_timeout) bytes += 22 + String(headers.keepAliveSecs).length; + if (headers.autoHeaders & AutoHeaders.transfer_encoding_chunked) bytes += 28; else if (!res.hasHeader("content-length") && !res.hasHeader("transfer-encoding") && res._hasBody) { if (fromEnd && !res._removedContLen && !res._removedTE) bytes += 18 + String(res._contentLength).length; else if (willBeChunked(res)) bytes += 28;