From ee14570db6499ebacf9c0ac474bdd154688e9366 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 16 Oct 2026 05:42:05 -0700 Subject: [PATCH 1/4] fix(http): retain corked responses in the socket write queue --- CHANGELOG.md | 2 + docs/runtime/nodejs-compat.mdx | 2 + packages/bun-uws/src/HttpContext.h | 2 +- src/js/internal/streams/utils.ts | 2 + src/js/internal/streams/writable.ts | 10 + src/js/node/_http_outgoing.ts | 1 + src/js/node/_http_server.ts | 325 ++++++++++-- .../bindings/node/JSNodeHTTPServerSocket.cpp | 5 + .../bindings/node/JSNodeHTTPServerSocket.h | 2 +- .../node/JSNodeHTTPServerSocketPrototype.cpp | 4 +- .../node/http/node-http-nested-cork.test.ts | 482 +++++++++++++++++- test/js/node/http/node-http-with-ws.test.ts | 22 +- 12 files changed, 787 insertions(+), 72 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fabad5ab23bf..72b322645525 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Retain corked HTTP response writes in the socket Writable queue, preserving raw-write ordering, backpressure, and callbacks through uncork and end. Adapts dispatcher-cork cleanup from [oven-sh/bun#35664](https://github.com/oven-sh/bun/pull/35664); thanks robobun. + - Match Node.js `SourceMap.findOrigin()` coordinates and field names, and return the closest preceding mapping from `findEntry()`. Adapts [oven-sh/bun#35123](https://github.com/oven-sh/bun/pull/35123); thanks robobun. - Keep `os.availableParallelism()` independent of changes to the global `navigator` object and its `hardwareConcurrency` property. diff --git a/docs/runtime/nodejs-compat.mdx b/docs/runtime/nodejs-compat.mdx index 4cc203680506..380cd831d2a2 100644 --- a/docs/runtime/nodejs-compat.mdx +++ b/docs/runtime/nodejs-compat.mdx @@ -75,6 +75,8 @@ On macOS, `fs.copyFile` and `fs.cp` accept regular-file `/dev/fd` sources withou ### [`node:http`](https://nodejs.org/api/http.html) +`ServerResponse.cork()` retains writes until uncork or end. Response and raw socket writes share the socket Writable queue while corked or draining, preserving their order and backpressure. Response and socket cork counts match Node.js. + `ServerResponse.writableFinished` stays false while a standalone response has output queued on its assigned socket. `net`, `http`, and `https` servers validate listen ports synchronously with `ERR_SOCKET_BAD_PORT`. Bind failures such as `EADDRINUSE` emit an asynchronous `error` event. diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index c0ce0a98f073..25ab7082ddf9 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -918,7 +918,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 044bd19cc5d9..9e28a9f335a7 100644 --- a/src/js/internal/streams/utils.ts +++ b/src/js/internal/streams/utils.ts @@ -15,6 +15,7 @@ const kIsWritable = SymbolFor("nodejs.stream.writable"); const kIsDisturbed = SymbolFor("nodejs.stream.disturbed"); const kOnConstructed = Symbol("kOnConstructed"); +const kWriteOperation = Symbol("kWriteOperation"); const kIsClosedPromise = SymbolFor("nodejs.webstream.isClosedPromise"); const kControllerErrorFunction = SymbolFor("nodejs.webstream.controllerErrorFunction"); @@ -269,6 +270,7 @@ function isErrored(stream) { export default { kOnConstructed, + kWriteOperation, isDestroyed, kIsDestroyed, isDisturbed, diff --git a/src/js/internal/streams/writable.ts b/src/js/internal/streams/writable.ts index d192409e5c13..9cfae1886978 100644 --- a/src/js/internal/streams/writable.ts +++ b/src/js/internal/streams/writable.ts @@ -12,6 +12,7 @@ const { addAbortSignal } = require("internal/streams/add-abort-signal"); const { getHighWaterMark, getDefaultHighWaterMark } = require("internal/streams/state"); const { kOnConstructed, + kWriteOperation, kState, kObjectMode, kErrorEmitted, @@ -437,6 +438,15 @@ function _write(stream, chunk, encoding, cb?) { } } + return writeOrError(stream, state, chunk, encoding, cb); +} + +// Internal byte-counted operations share the byte stream's queue and lifecycle. +Writable.prototype[kWriteOperation] = function (operation, callback) { + return writeOrError(this, this._writableState, operation, kWriteOperation, callback ?? nop) === true; +}; + +function writeOrError(stream, state, chunk, encoding, cb) { let err; if ((state[kState] & kEnding) !== 0) { err = $ERR_STREAM_WRITE_AFTER_END(); diff --git a/src/js/node/_http_outgoing.ts b/src/js/node/_http_outgoing.ts index 5627d3e00d0b..c8ce779c2ba3 100644 --- a/src/js/node/_http_outgoing.ts +++ b/src/js/node/_http_outgoing.ts @@ -1292,6 +1292,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 860f9e216914..82977230cb58 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -1,6 +1,7 @@ // Hardcoded module "node:_http_server" const EventEmitter: typeof import("node:events").EventEmitter = require("node:events"); const { Stream } = require("node:stream"); +const { kWriteOperation } = require("internal/streams/utils"); const setImmediate = $newCppFunction("node/NodeTimers.cpp", "functionSetImmediate", 1); const { _checkInvalidHeaderChar: checkInvalidHeaderChar, @@ -80,6 +81,7 @@ const { emitListeningEvent, lookupListenAddress, registerListenCallback } = requ const { IncomingMessage, kReqShouldKeepAlive } = require("node:_http_incoming"); const { OutgoingMessage, + kCorked, kErrored, kHighWaterMark, kSocket, @@ -1266,7 +1268,6 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort } } - socket[kCorkForDispatcher](); // Node's parserOnMessageComplete: a message without a body is complete once the listener returns. EOF stays lazy (_read). if (!hasBody) http_req.complete = true; @@ -1600,9 +1601,117 @@ function onReadableStreamEnd() {} function clearUpgradeIncoming(socket) { socket[kUpgradeIncoming] = undefined; } -const kCorkForDispatcher = Symbol("corkForDispatcher"); -const kReleaseDispatcherCork = Symbol("releaseDispatcherCork"); -const kDispatcherCorkDepth = Symbol("dispatcherCorkDepth"); +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))) && + isNodeHTTPServerSocket(socket) + ); +} + +function onResponseSocketDrain() { + const response = this._httpMessage; + if (response) flushWriteAccountingNT(response); +} + +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) handle.onwritable = callback; + else callback(); + } +} // Node's UpgradeStream._read: resume the request only while its body is incomplete. function resumePausedUpgradeIncoming(socket) { @@ -1624,7 +1733,6 @@ function detachSocketListenersForHandoff(socket) { socket[kHandedOff] = true; socket.removeListener("error", socketOnError); socket.removeListener("timeout", onNodeHTTPServerSocketTimeout); - socket[kReleaseDispatcherCork](); socket.on("end", onReadableStreamEnd); // Like Node's onParserExecuteCommon: a flowing socket would drop tunnel bytes that arrive before the listener attaches a reader. socket.readableFlowing = null; @@ -1825,7 +1933,6 @@ function getNodeHTTPServerSocket() { [kBytesWritten] = 0; [kHandle]; [kUpgradeIncoming]: import("node:http").IncomingMessage | undefined = undefined; - [kDispatcherCorkDepth] = 0; [kOnReadParsed] = undefined; [kHandoffResponse] = undefined; [kPendingHandoff]: (() => void)[] | undefined = undefined; @@ -1838,6 +1945,7 @@ function getNodeHTTPServerSocket() { // connection whose pipelined responses have buffered past the high water mark. _paused = false; #pendingCallback = null; + #pendingResponseCallback = null; #pendingAbortMessage; #closeHandled = false; #resetSupported; @@ -1916,30 +2024,14 @@ function getNodeHTTPServerSocket() { handle.ondrain = this.#onDrain.bind(this); } else { handle.ondata = undefined; - handle.ondrain = undefined; + if (!this.#pendingCallback) handle.ondrain = undefined; } } } - [kCorkForDispatcher]() { - if (this[kDispatcherCorkDepth] !== 0) return; - const corkedBefore = this.writableCorked; - this.cork(); - const corkedAfter = this.writableCorked; - if (corkedAfter > corkedBefore) this[kDispatcherCorkDepth] = corkedAfter; - } - [kReleaseDispatcherCork]() { - if (this[kDispatcherCorkDepth] === 0) return; - this[kDispatcherCorkDepth] = 0; - super.uncork(); - } uncork() { - const dispatcherCorkDepth = this[kDispatcherCorkDepth]; const result = super.uncork(); - // Caller corks can sit below the dispatcher's cork. Follow partial - // releases down to zero so handoff still removes exactly Bun's cork. - if (dispatcherCorkDepth !== 0) { - this[kDispatcherCorkDepth] = Math.min(dispatcherCorkDepth, this.writableCorked); - } + const response = this._httpMessage; + if (!hasResponseSocketWork(this) && response) scheduleWriteAccountingFlush(response); return result; } #onDrain() { @@ -1950,7 +2042,6 @@ function getNodeHTTPServerSocket() { this.#pendingCallback = null; (callback as Function)(); } - this.emit("drain"); } #onData(chunk, last) { this._unrefTimer(); @@ -2104,6 +2195,7 @@ function getNodeHTTPServerSocket() { this.#pendingCallback = null; (pendingCallback as Function)(writeFailure); } + this.#pendingResponseCallback?.(writeFailure); } #onCloseForDestroy(closeCallback, err: Error | undefined, handle) { this.#onClose(handle); @@ -2384,24 +2476,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() { @@ -2426,6 +2545,10 @@ function getNodeHTTPServerSocket() { return this[kHandoffResponse] ?? this[kHandle]?.response; } } 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; } @@ -2648,6 +2771,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; @@ -3247,7 +3402,6 @@ function advanceResponsePipeline(server, socket) { } function replayResponseWrites(res, queued) { - const handle = res[kHandle]; // Replay the writes buffered while the response was unassigned. // The buffered bytes are handed to the native handle below, so they no // longer count as pending output (Node's _flush does the same). @@ -3268,12 +3422,10 @@ function replayResponseWrites(res, queued) { 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 if (kind === "end") { ServerResponsePrototypeEnd.$call(res, op[1], op[2], op[3]); } else { @@ -3319,10 +3471,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]; @@ -3347,6 +3496,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; @@ -3410,6 +3565,7 @@ function bufferPipelinedEnd(res, queued, chunk, encoding, callback) { addPipelineOutgoingData(queued, bytes); } queued.ended = true; + res[kCorked] = 0; res.finished = true; return res; } @@ -3524,6 +3680,10 @@ Object.defineProperty(ServerResponse.prototype, "socket", { return this[fakeSocketSymbol] ?? null; }, 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 @@ -3556,18 +3716,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) { @@ -3632,6 +3794,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. @@ -3639,6 +3804,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"); @@ -3735,6 +3901,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?.(); @@ -3744,8 +3915,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 = undefined) { const handle = this[kHandle]; + const fromSocket = token === kResponseSocketWrite; if ($isCallable(chunk)) { callback = chunk; @@ -3761,7 +3933,7 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) { // Node's write_() drops the chunk of a destroyed response, and end() goes on: https://github.com/nodejs/node/blob/v26.3.0/lib/_http_outgoing.js#L951-L964 if (chunk && this.destroyed && !this.finished) chunk = undefined; - if (hasServerResponseFinished(this, chunk, callback, true)) { + if (!fromSocket && hasServerResponseFinished(this, chunk, callback, true)) { return this; } @@ -3777,6 +3949,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 @@ -3829,7 +4012,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 @@ -3840,7 +4023,7 @@ ServerResponse.prototype.end = function (chunk, encoding, callback) { renderedHeaders, chunk, encoding, - strictContentLength(this, headerState, true), + fromSocket ? operation.strict : strictContentLength(this, headerState, true), renderedAutoHeaders, renderedKeepAliveSecs, ); @@ -3876,10 +4059,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' @@ -3944,8 +4134,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 = undefined) { const handle = this[kHandle]; + const fromSocket = token === kResponseSocketWrite; if ($isCallable(chunk)) { callback = chunk; @@ -3958,7 +4149,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; } @@ -3971,6 +4162,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]; @@ -4011,14 +4205,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, @@ -4043,6 +4237,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. @@ -4056,6 +4251,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; @@ -4081,7 +4277,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); } @@ -4099,7 +4295,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"); @@ -4157,6 +4358,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 ( @@ -4164,6 +4369,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)) ); @@ -4184,6 +4390,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 @@ -4214,6 +4424,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) { @@ -4361,12 +4572,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, @@ -4431,7 +4648,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 33fb6491308d..54b965d89a92 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp @@ -181,6 +181,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 0942d0b4772d..fee7d0daaa85 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.h +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.h @@ -153,7 +153,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 69a6a1477dd2..435a7eb92d82 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocketPrototype.cpp @@ -284,8 +284,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..8e72526bc6ab 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,485 @@ 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"); + 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.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"); + 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", + ]) { + 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 94219ff27e09..e4aa4ab2cd98 100644 --- a/test/js/node/http/node-http-with-ws.test.ts +++ b/test/js/node/http/node-http-with-ws.test.ts @@ -262,27 +262,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"); @@ -299,12 +291,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 85f1ff18af4064de121a3b79d661f4b109a134b3 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 16 Oct 2026 05:48:25 -0700 Subject: [PATCH 2/4] fix(http): type queued response callbacks and operations --- src/js/node/_http_server.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index 82977230cb58..1ef68a709a57 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -1944,8 +1944,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; - #pendingResponseCallback = null; + #pendingCallback: ((error?: Error) => void) | null = null; + #pendingResponseCallback: ((error?: Error) => void) | null = null; #pendingAbortMessage; #closeHandled = false; #resetSupported; @@ -3915,7 +3915,7 @@ 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, token = undefined, operation = undefined) { +ServerResponse.prototype.end = function (chunk, encoding, callback, token = undefined, operation: any = undefined) { const handle = this[kHandle]; const fromSocket = token === kResponseSocketWrite; @@ -4134,7 +4134,7 @@ Object.defineProperty(ServerResponse.prototype, "writable", { set(_value) {}, }); -ServerResponse.prototype.write = function (chunk, encoding, callback, token = undefined, operation = undefined) { +ServerResponse.prototype.write = function (chunk, encoding, callback, token = undefined, operation: any = undefined) { const handle = this[kHandle]; const fromSocket = token === kResponseSocketWrite; From 1da9a8710fc378ee4fe5af57418d8dfc8d15df1b Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 16 Oct 2026 06:26:30 -0700 Subject: [PATCH 3/4] test(http): await standalone response write callbacks --- test/js/node/http/node-http-nested-cork.test.ts | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 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 2af84583fc71..7eee10ca27aa 100644 --- a/test/js/node/http/node-http-nested-cork.test.ts +++ b/test/js/node/http/node-http-nested-cork.test.ts @@ -11,6 +11,8 @@ async function responseCorkOracle(mode: string) { 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); @@ -20,14 +22,17 @@ async function responseCorkOracle(mode: string) { res.assignSocket(socket); res.strictContentLength = mode === "standalone-strict"; res.setHeader("content-length", 2); - let callbacks = 0; + const onWrite = error => { + assert.ifError(error); + if (++callbacks === 2) written.resolve(); + }; res.on("finish", () => { - finished.resolve(callbacks); + finished.resolve(); socket.end(); }); res.cork(); - res.write("A", () => callbacks++); - res.write("B", () => callbacks++); + res.write("A", onWrite); + res.write("B", onWrite); res.end(); }); await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); @@ -37,7 +42,8 @@ async function responseCorkOracle(mode: string) { 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"); - assert.equal(await finished.promise, 2); + 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 { From 193d527de5fff45ac98a9dbf1c85412f25fb5a6d Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 16 Oct 2026 06:36:02 -0700 Subject: [PATCH 4/4] fix(http): preserve queued header drain callbacks --- src/js/node/_http_server.ts | 6 +++-- .../node/http/node-http-nested-cork.test.ts | 23 +++++++++++++++++++ 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index ba2a65b9038d..695821174869 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -1709,8 +1709,10 @@ function executeResponseSocketWrite(operation, callback) { if (kind === "raw") handle.writeInformational(chunk, encoding); else if (kind === "continue") handle.writeContinue(); else ServerResponsePrototypeFlushHeaders.$call(response, kResponseSocketWrite); - if (handle.bufferedAmount > 0) handle.onwritable = callback; - else callback(); + if (handle.bufferedAmount > 0) { + response[kPendingCallbacks].push(callback); + handle.onwritable = allowWritesToContinue.bind(response); + } else callback(); } } 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 7eee10ca27aa..6c12ebba449d 100644 --- a/test/js/node/http/node-http-nested-cork.test.ts +++ b/test/js/node/http/node-http-nested-cork.test.ts @@ -110,6 +110,27 @@ async function responseCorkOracle(mode: string) { 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"]; @@ -420,6 +441,7 @@ async function responseCorkOracle(mode: string) { 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 => { @@ -511,6 +533,7 @@ describe.each(corkRuntimes)("response cork Node oracle (%s)", (_runtime, executa "native-queued", "standalone", "standalone-strict", + "informational-pressure", ]) { test.concurrent(mode, async () => { await using child = Bun.spawn({