From 147965f7d6d5f623c64a856c45d50755e0240281 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:04:44 +0000 Subject: [PATCH] node:http: a 'connect' or 'upgrade' socket keeps its unread bytes when the connection closes The socket that a 'connect' listener gets, and the socket of an 'upgrade' listener whose request has no body, is the connection itself as a net.Socket. usockets closes a connection when both sides have sent their FIN, and NodeHTTPServerSocket answered that close with destroy(). The bytes that its reader had not taken were dropped, and there was no 'end'. Like Node's net.Socket, the socket closes after its own 'end' and 'finish'. After the peer's FIN it waits for its reader, or for its owner's destroy(). The end of the stream is push(null) and read(0), which leaves a paused socket paused. usockets records the peer's FIN on the socket (us_socket_read_fin), and uWS reports the end of a tunnel's stream only for it. A connection that closes with no FIN of the peer (a reset, a failed TLS session) closes the socket as before. The server tracks connections. It stops tracking the socket when its connection closes, so a socket that waits is not counted, server.close() does not wait for it, and a socket that its owner dropped is collected. --- packages/bun-usockets/src/context.c | 2 + packages/bun-usockets/src/internal/internal.h | 2 + packages/bun-usockets/src/libusockets.h | 4 + packages/bun-usockets/src/loop.c | 2 + packages/bun-usockets/src/socket.c | 5 + packages/bun-uws/src/HttpContext.h | 5 +- src/js/node/_http_server.ts | 110 ++- test/js/node/http/node-http-connect.node.mts | 545 ++++++++++++++ test/js/node/http/node-http-connect.test.ts | 662 +++++++++++++++++- 9 files changed, 1312 insertions(+), 25 deletions(-) diff --git a/packages/bun-usockets/src/context.c b/packages/bun-usockets/src/context.c index 86ccc3dbb3ca..b42fe2fe348f 100644 --- a/packages/bun-usockets/src/context.c +++ b/packages/bun-usockets/src/context.c @@ -346,6 +346,7 @@ static void us_internal_init_listen_socket(struct us_listen_socket_t *ls, s->flags.allow_half_open = (options & LIBUS_SOCKET_ALLOW_HALF_OPEN); s->unclassified_send_failures = 0; s->read_eof = 0; + s->read_fin = 0; s->hangup_closes_unsent = 0; s->next = 0; s->prev = 0; @@ -526,6 +527,7 @@ static inline void us_internal_init_connect_socket(struct us_socket_t *s, s->flags.last_write_failed = 0; s->unclassified_send_failures = 0; s->read_eof = 0; + s->read_fin = 0; s->hangup_closes_unsent = 0; s->connect_state = NULL; s->connect_next = NULL; diff --git a/packages/bun-usockets/src/internal/internal.h b/packages/bun-usockets/src/internal/internal.h index da99f5a427af..9b6e029390db 100644 --- a/packages/bun-usockets/src/internal/internal.h +++ b/packages/bun-usockets/src/internal/internal.h @@ -369,6 +369,8 @@ struct us_socket_t { unsigned char ssl_has_pending_events : 1; /* Peer FIN was dispatched as on_end on a half-open socket; readable interest is never re-added and on_end never re-fires. */ unsigned char read_eof : 1; + /* A read returned 0: the peer ended its stream with a FIN. on_end is not always dispatched for it (see us_socket_read_fin). */ + unsigned char read_fin : 1; /* A hangup that leaves bytes unsent closes the socket even while it is paused (loop.c defers it otherwise). * For an owner whose pause can wait for those bytes to drain: node:http's pipelining. */ unsigned char hangup_closes_unsent : 1; diff --git a/packages/bun-usockets/src/libusockets.h b/packages/bun-usockets/src/libusockets.h index 62a9f6a496ac..438228aec270 100644 --- a/packages/bun-usockets/src/libusockets.h +++ b/packages/bun-usockets/src/libusockets.h @@ -714,6 +714,10 @@ int us_socket_ssl_handshake_callback_has_fired(us_socket_r s) nonnull_fn_decl; unsigned int us_socket_ssl_spill_pending(us_socket_r s) nonnull_fn_decl; /* Whether some of that ciphertext was thrown away before the kernel took it, which happens when the connection ends. */ int us_socket_ssl_spill_dropped(us_socket_r s) nonnull_fn_decl; +/* Whether a read of this socket returned 0, which is the peer's FIN. The close code does not tell: a close that this + * side starts and a failed TLS session close with code 0 too. When the FIN follows this side's own, the socket closes + * with no on_end, so this is the one record of it. */ +int us_socket_read_fin(us_socket_r s) nonnull_fn_decl; struct us_socket_t *us_socket_close(us_socket_r s, int code, void *reason) __attribute__((nonnull(1))); /* Closes before it returns. A TLS socket first sends what the kernel takes now: its unsent ciphertext, then close_notify. */ diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index f803999f6590..bcd7e23e4935 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -555,6 +555,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in s->flags.last_write_failed = 0; s->unclassified_send_failures = 0; s->read_eof = 0; + s->read_fin = 0; s->hangup_closes_unsent = 0; /* We always use nodelay */ @@ -845,6 +846,7 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in } else if (!length) { eof = LIBUS_POLL_EOF; // lets handle EOF in the same place read_fin = 1; + s->read_fin = 1; break; } else if (length == LIBUS_SOCKET_ERROR && !bsd_would_block()) { /* A read error closes with its errno even when the drain above diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 950c117f5c02..877e367eee81 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -168,6 +168,10 @@ int us_socket_ssl_spill_dropped(struct us_socket_t *s) { return s->ssl && s->ssl_spill_dropped; } +int us_socket_read_fin(struct us_socket_t *s) { + return s->read_fin; +} + int us_connecting_socket_is_closed(struct us_connecting_socket_t *c) { return c->closed; } @@ -452,6 +456,7 @@ struct us_socket_t *us_socket_from_fd(struct us_socket_group_t *group, unsigned s->flags.last_write_failed = 0; s->unclassified_send_failures = 0; s->read_eof = 0; + s->read_fin = 0; s->hangup_closes_unsent = 0; s->connect_state = NULL; diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index 696c9c6cd4cf..166a37f487b4 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -263,7 +263,10 @@ struct HttpContext { bool nodeHttpTunnelAfterBody = false; if constexpr (IsNodeHttp) nodeHttpTunnelAfterBody = (httpResponseData->state & HttpResponseData::HTTP_NODE_TUNNEL_AFTER_BODY) != 0; if(httpResponseData->isConnectRequest || nodeHttpTunnelAfterBody) { - if (httpResponseData->socketData && httpContextData->onSocketData) { + /* node:http: only the peer's FIN ends its stream. A reset, a failed TLS session and a close that this side starts do not. */ + bool streamEnded = true; + if constexpr (IsNodeHttp) streamEnded = us_socket_read_fin(s); + if (streamEnded && httpResponseData->socketData && httpContextData->onSocketData) { httpContextData->onSocketData(httpResponseData->socketData, SSL, s, "", 0, true); } /* node:http: the body of a tunnel that started has had its last chunk from the parser. This close can come from inside that call. */ diff --git a/src/js/node/_http_server.ts b/src/js/node/_http_server.ts index ed77f9dbb9fb..888fbf4ecf83 100644 --- a/src/js/node/_http_server.ts +++ b/src/js/node/_http_server.ts @@ -89,9 +89,24 @@ const kPendingCloseGenerations = Symbol("http.server.pendingCloseGenerations"); const kListenerGeneration = Symbol("http.server.listenerGeneration"); // Set on the 'connect'/'upgrade' handoff; closeAll/closeIdleConnections() skip these, as in Node. const kHandedOff = Symbol("http.server.socketHandedOff"); +// Its values. Node hands its listener the net.Socket itself when the request is complete at the emit, and otherwise an +// UpgradeStream, which the close of the connection destroys: +// https://github.com/nodejs/node/blob/v26.3.0/lib/_http_server.js#L1027-L1050 +// https://github.com/nodejs/node/blob/v26.3.0/lib/_http_server.js#L1110-L1143 +// Bun emits 'upgrade' before it parses the body, so an Upgrade request with a body is never complete there. +const kHandedOffSocket = 1; +const kHandedOffUpgradeStream = 2; +// #closeHandled of a kHandedOffSocket that has the end of its stream. Like Node's net.Socket, it closes after its own +// 'end' and 'finish', so the close of the connection leaves it the bytes that its reader has not taken. +const kClosesAfterEnd = 2; +// The connection has closed, and the socket waits for its reader. +const kOutlivesConnection = 3; function closeHandle(handle) { if (!handle.closed) handle.close(); } +function destroySocket(socket) { + socket.destroy(); +} const kHttpAllowHalfOpen = Symbol("http.server.httpAllowHalfOpen"); const kMaxHeadersCount = Symbol("http.server.maxHeadersCount"); @@ -819,7 +834,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort socket[kHandoffResponse] = handle; // The parser is detached: the socket is handed over with only // net.Socket's 'end' listener left, like Node.js. - detachSocketListenersForHandoff(socket); + detachSocketListenersForHandoff(socket, kHandedOffSocket); // Pass the pipelined data (head buffer) if any was received with the CONNECT request const head = connectHead ? connectHead : kEmptyBuffer; socket.bytesRead += head.length; @@ -1052,7 +1067,7 @@ Server.prototype[kRealListen] = function (tls, port, host, socketPath, reusePort socketHandle.upgradeToTunnel(hasBody, handle); socket[kHandoffResponse] = handle; socket[kEnableStreaming](true); - detachSocketListenersForHandoff(socket); + detachSocketListenersForHandoff(socket, hasBody ? kHandedOffUpgradeStream : kHandedOffSocket); // Node frees the parser before emitting 'upgrade' (socket.parser === null there). releaseServerParserShim(socket, http_req); if (hasBody) { @@ -1313,7 +1328,8 @@ function onServerRequestEvent(this: NodeHTTPServerSocket, event: NodeHTTPRespons const socket: NodeHTTPServerSocket = this; switch (event) { case NodeHTTPResponseAbortEvent.abort: { - if (!socket.destroyed) { + // A handed-off socket closes in #onClose, which the same native close calls next. + if (!socket.destroyed && !socket[kHandedOff]) { socket.destroy(); } break; @@ -1482,8 +1498,8 @@ function resumePausedUpgradeIncoming(socket) { // connection-listener set removed (onParserExecuteCommon removes its data/end/ // close/drain/error/timeout listeners) and only net.Socket's own 'end' listener // left in place. -function detachSocketListenersForHandoff(socket) { - socket[kHandedOff] = true; +function detachSocketListenersForHandoff(socket, as: typeof kHandedOffSocket | typeof kHandedOffUpgradeStream) { + socket[kHandedOff] = as; socket.removeListener("error", socketOnError); socket.removeListener("timeout", onNodeHTTPServerSocketTimeout); socket.on("end", onReadableStreamEnd); @@ -1670,7 +1686,7 @@ function getNodeHTTPServerSocket() { [kOnReadParsed] = undefined; [kHandoffResponse] = undefined; [kDestroySoon] = false; - [kHandedOff] = false; + [kHandedOff]: false | typeof kHandedOffSocket | typeof kHandedOffUpgradeStream = false; server: Server; _httpMessage; _secureEstablished = false; @@ -1679,15 +1695,17 @@ function getNodeHTTPServerSocket() { _paused = false; #pendingCallback = null; #pendingAbortMessage; - #closeHandled = false; + #closeHandled: boolean | typeof kClosesAfterEnd | typeof kOutlivesConnection = false; #peer; #closingHandle; #resetSupported; #closeError: Error | undefined = undefined; declare encrypted: boolean; + // A slot of net.Socket. + declare _server: Server; declare resetAndClosing: boolean; declare _writableState: { emitClose: boolean; decodeStrings: boolean }; - declare _readableState: { emitClose: boolean }; + declare _readableState: { emitClose: boolean; ended: boolean }; declare writableBuffer: { chunk: Buffer }[]; constructor(server: Server, handle, encrypted, listenerGeneration) { // allowHalfOpen: node's connectionListener sockets never auto-end the @@ -1707,6 +1725,8 @@ function getNodeHTTPServerSocket() { this._readableState.emitClose = true; this._writableState.decodeStrings = true; this.server = server; + // The server that accepted the connection tracks it. A 'connection' listener of any server reassigns `server`. + this._server = server; this.#resetSupported = !encrypted && !listenerGeneration?.isUnix; this[kHandle] = handle; this._secureEstablished = !!handle?.secureEstablished; @@ -1805,11 +1825,19 @@ function getNodeHTTPServerSocket() { this[kHandoffResponse]?.unref(); } - this.push(null); - // Like Node's net.Socket (and its UpgradeStream): EOF on a socket with no - // pending data emits 'end' even when nothing is reading the socket. - if (this.readableLength === 0) { - this.resume(); + if (this[kHandedOff] === kHandedOffSocket) { + // Before push(null), which runs 'readable' listeners. + this.#closeHandled = kClosesAfterEnd; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L231-L232 + this.push(null); + // 'end' with nothing to read and nobody reading. A paused socket stays paused. A flowing one has its 'end' from push(null). + if (this.readableFlowing !== true) this.read(0); + } else { + this.push(null); + // https://github.com/nodejs/node/blob/v26.3.0/lib/_http_server.js#L1042-L1050 + if (this.readableLength === 0) { + this.resume(); + } } } } @@ -1831,9 +1859,39 @@ function getNodeHTTPServerSocket() { handle.close(); } } + // Whether a handed-off socket whose connection closed with no FIN of the peer gets 'end' when its reader has + // taken every byte, as it always has. That is the close after a reset of the peer, and after a failed TLS + // session. Node emits 'error' there. pipe() needs the 'end' to end its destination. + #endsWithItsConnection() { + // This side closed the connection. Node's net.Socket emits no 'end' after destroySoon(). + if (this.destroyed || (this[kDestroySoon] && this[kHandedOff] === kHandedOffSocket)) return false; + // A WebSocket took the connection over, and reads it in native code. + return !this[kHandoffResponse]?.upgraded; + } #onClose(closedHandle = this[kHandle]) { - // Once: a queued response's abort can destroy the socket, and run this, before the native close calls it. - if (this.#closeHandled) return; + if (this.#closeHandled) { + // Once: a queued response's abort can destroy the socket, and run this, before the native close calls it. + if (this.#closeHandled === true) return; + if (!this.destroyed && !this.readableEnded && !this[kDestroySoon] && !this.#pendingCallback) { + // The peer's FIN ended the stream of a net.Socket, and its reader has not taken all of it. Like Node's, the + // socket closes after its own 'end' and 'finish': _destroy() runs the rest of this function. A write that + // waits for a drain has failed, and destroySoon() asked for the close, so neither waits for the reader. + this.#closeHandled = kOutlivesConnection; + // The handle went with the connection. + this[kBytesWritten] = handleBytesWritten(closedHandle); + this[kClosedWithUnsentData] = closedHandle.closedWithUnsentData; + this[kHandle] = null; + // So did the server's part: it tracks connections, and keeps nothing for a socket that its owner can drop. + // Node, whose socket keeps its handle, counts the socket until it closes, and server.close() waits for it. + const server = this._server; + const tracked = server[kTrackedConnections]; + tracked.delete(this); + if (tracked.size === 0 && server[kPendingDrainClose]) process.nextTick(emitCloseServer, server); + // The keep-alive timer that the dispatcher leaves armed with no timeout would refresh itself for ever. + if (this.timeout === 0) this.setTimeout(0); + return; + } + } this.#closeHandled = true; const errored = this.errored; // freeParser equivalent: runs before 'close' listeners so they observe the @@ -1844,7 +1902,10 @@ function getNodeHTTPServerSocket() { this[kBytesWritten] = handleBytesWritten(closedHandle); this[kClosedWithUnsentData] = closedHandle.closedWithUnsentData; // Peer FIN: 'end' before 'close', like Node's net.Socket. read(0) emits it when nothing reads the socket. - if (closedHandle.peerEnded) { + if ( + closedHandle.peerEnded || + (this[kHandedOff] && !this._readableState.ended && this.#endsWithItsConnection()) + ) { this.push(null); this.read(0); } @@ -1856,7 +1917,7 @@ function getNodeHTTPServerSocket() { this.#closeError = er.code === closeError.code ? er : closeError; } } - const server = this.server; + const server = this._server; const tracked = server?.[kTrackedConnections]; if (tracked) { tracked.delete(this); @@ -2000,6 +2061,8 @@ function getNodeHTTPServerSocket() { } const handle = this[kHandle]; if (!handle) { + // The connection closed before, and the socket waited for its reader: the rest of its close runs now. + if (this.#closeHandled === kOutlivesConnection) this.#onClose(null); if ($isCallable(callback)) callback(err); return; } @@ -2152,6 +2215,12 @@ function getNodeHTTPServerSocket() { } this.resetAndClosing = true; this.destroy(); + } else if (this.#closeHandled === kOutlivesConnection) { + // Nothing is left to reset. Node, which still has the connection, resets it and emits no error. + if (!this.#resetSupported) { + throw $ERR_INVALID_HANDLE_TYPE(); + } + this.destroy(); } else { this.destroy($ERR_SOCKET_CLOSED()); } @@ -2230,6 +2299,13 @@ function getNodeHTTPServerSocket() { this.#pendingCallback = _callback; return false; } + } else if (this.#closeHandled === kOutlivesConnection) { + // The connection has closed, and the socket waits for its reader. The write fails like one that waited for a + // drain at the close: ERR_STREAM_DESTROYED to its callback, and no 'error'. Node fails it with EPIPE, and + // emits 'error'. + this.#pendingCallback = _callback; + process.nextTick(destroySocket, this); + return false; } } catch (e) { err = e; diff --git a/test/js/node/http/node-http-connect.node.mts b/test/js/node/http/node-http-connect.node.mts index af217c8ff396..cee029d0456e 100644 --- a/test/js/node/http/node-http-connect.node.mts +++ b/test/js/node/http/node-http-connect.node.mts @@ -57,6 +57,7 @@ import { createProxy } from "proxy"; import { once } from "node:events"; import type { AddressInfo } from "node:net"; import net from "node:net"; +import { pipeline, Writable } from "node:stream"; function connectClient(proxyAddress: AddressInfo, targetAddress: AddressInfo, add_http_prefix: boolean) { const client = net.connect({ port: proxyAddress.port, host: proxyAddress.address }, () => { @@ -423,3 +424,547 @@ describe("HTTP server CONNECT", () => { expect(combined.match(/Hello/g)?.length).toBe(2); }); }); + +// The socket of a 'connect' listener, or of an 'upgrade' listener whose request has no body, is the connection +// itself as a net.Socket. It closes after its own 'end' and 'finish', so it outlives a connection that closes +// before its reader has taken every byte. +describe("a tunnel socket whose connection closes before it is read", () => { + const exchanges = { + connect: { + request: "CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n", + reply: "HTTP/1.1 200 Connection Established\r\n\r\n", + }, + upgrade: { + request: "GET / HTTP/1.1\r\nHost: example.com\r\nConnection: Upgrade\r\nUpgrade: raw\r\n\r\n", + reply: "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: raw\r\n\r\n", + }, + }; + const handoffs = ["connect", "upgrade"] as const; + const finOrders = [ + ["the listener ends the socket, then the peer sends and ends", true], + ["the peer sends and ends, then the listener ends the socket", false], + ] as const; + + // 251 is prime, so the pattern lines up with no chunk and no read. + const pattern = Buffer.from(Array.from({ length: 251 }, (_, i) => i)); + const patterned = (length: number) => Buffer.alloc(length, pattern); + // Half of the high water mark: the socket reads all of it with nobody reading the socket. + const belowHighWaterMark = (socket: net.Socket) => patterned(socket.readableHighWaterMark / 2); + + // One exchange on a connection of its own. When it is over, the server has read what was sent before it. + async function barrier(port: number) { + const socket = net.connect(port, "127.0.0.1"); + socket.on("error", () => {}); + socket.resume(); + socket.end("GET /barrier HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n"); + await once(socket, "close"); + } + + const connectionsOf = (server: net.Server) => + new Promise((resolve, reject) => + server.getConnections((error, count) => (error ? reject(error) : resolve(count))), + ); + + // The listener runs `early`, answers, and ends the socket if `ends`. The client is half-open, so it still sends + // after the server's FIN. `events` has the name of each event in `watch` that the socket emits. + async function openTunnel( + event: "connect" | "upgrade", + options: { ends: boolean; watch?: string[]; early?: (socket: net.Socket) => void }, + ) { + const events: string[] = []; + const { promise: handedOff, resolve: onHandoff } = Promise.withResolvers(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + const server = http.createServer((req, res) => void res.end("ok")); + server.on(event, (req, socket) => { + for (const name of options.watch ?? ["end", "error", "close"]) socket.on(name, () => events.push(name)); + socket.on("close", () => onClosed()); + options.early?.(socket as net.Socket); + socket.write(exchanges[event].reply); + if (options.ends) socket.end(); + onHandoff(socket as net.Socket); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address() as AddressInfo; + const client = net.connect({ port, host: "127.0.0.1", allowHalfOpen: true }); + client.on("error", () => {}); + client.on("data", () => {}); + const ready = once(client, options.ends ? "end" : "data"); + client.write(exchanges[event].request); + const tunnel = await handedOff; + await ready; + return { + server, + port, + client, + tunnel, + events, + closed, + async [Symbol.asyncDispose]() { + client.destroy(); + tunnel.destroy(); + server.close(); + }, + }; + } + + type Reader = { + // Runs in the listener. + early?: (socket: net.Socket) => void; + // Runs when the connection has closed. + late: (socket: net.Socket) => unknown; + }; + const readers: Record Reader> = { + "a 'data' listener that comes late": chunks => ({ + late: socket => void socket.on("data", (chunk: Buffer) => chunks.push(chunk)), + }), + "pause() with a 'data' listener and a late resume()": chunks => ({ + early: socket => void socket.on("data", (chunk: Buffer) => chunks.push(chunk)).pause(), + late: socket => void socket.resume(), + }), + "a 'readable' listener that comes late": chunks => ({ + late: socket => + void socket.on("readable", () => { + for (let chunk: Buffer | null; (chunk = socket.read()) !== null;) chunks.push(chunk); + }), + }), + "a for await loop that starts late": chunks => ({ + late: async socket => { + for await (const chunk of socket) chunks.push(chunk); + }, + }), + "pipe() to a Writable that is busy with its first chunk": chunks => { + const { promise: busy, resolve: onConnectionClosed } = Promise.withResolvers(); + const destination = new Writable({ + highWaterMark: 1, + write(chunk, encoding, callback) { + chunks.push(chunk); + busy.then(() => callback()); + }, + }); + return { + early: socket => void socket.pipe(destination), + late: socket => { + onConnectionClosed(); + // A socket that was destroyed does not end its destination. + if (!socket.destroyed) return once(destination, "finish"); + }, + }; + }, + }; + + for (const [order, listenerEndsFirst] of finOrders) { + describe(order, () => { + for (const event of handoffs) { + // The socket does the same for every reader, so 'upgrade' runs one of them. + for (const name of event === "connect" ? Object.keys(readers) : ["a 'data' listener that comes late"]) { + test(`'${event}': every byte, then 'end', then 'close', for ${name}`, async () => { + const chunks: Buffer[] = []; + const reader = readers[name](chunks); + await using t = await openTunnel(event, { ends: listenerEndsFirst, early: reader.early }); + const sent = belowHighWaterMark(t.tunnel); + // In two reads, for the reader that is busy with the first one. + t.client.write(sent.subarray(0, sent.length / 2)); + await barrier(t.port); + t.client.end(sent.subarray(sent.length / 2)); + await barrier(t.port); + if (!listenerEndsFirst) { + t.tunnel.end(); + await barrier(t.port); + } + + // Both sides have sent their FIN, so the connection has closed. The socket stays for its reader. + const atClose = { destroyed: t.tunnel.destroyed, events: [...t.events] }; + await reader.late(t.tunnel); + await t.closed; + expect({ atClose, intact: Buffer.concat(chunks).equals(sent), events: t.events }).toEqual({ + atClose: { destroyed: false, events: [] }, + intact: true, + events: ["end", "close"], + }); + }); + } + } + }); + } + + for (const event of handoffs) { + describe(`'${event}'`, () => { + // The end of the stream is push(null), then read(0). That emits 'end' when there is nothing to read, and it + // does not resume the socket: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L231-L232 + for (const [when, ends] of [ + ["the listener ended the socket", true], + ["the listener left the socket open", false], + ] as const) { + test(`the peer's FIN leaves a paused socket paused when ${when}`, async () => { + await using t = await openTunnel(event, { + ends, + watch: ["pause", "resume", "end", "error", "close"], + early: socket => void socket.pause(), + }); + const ended = once(t.tunnel, "end"); + t.client.end(); + await ended; + if (ends) await t.closed; + else await barrier(t.port); + expect({ events: t.events, paused: t.tunnel.isPaused(), flowing: t.tunnel.readableFlowing }).toEqual({ + // A socket that has ended both ways closes. + events: ends ? ["pause", "end", "close"] : ["pause", "end"], + paused: true, + flowing: false, + }); + }); + } + + for (const method of ["destroy", "destroySoon", "resetAndDestroy"] as const) { + test(`${method}() closes a socket that waits for its reader`, async () => { + await using t = await openTunnel(event, { ends: true }); + const sent = belowHighWaterMark(t.tunnel); + t.client.end(sent); + await barrier(t.port); + + const before = { destroyed: t.tunnel.destroyed, buffered: t.tunnel.readableLength }; + t.tunnel[method](); + await t.closed; + const connections = await connectionsOf(t.server); + await new Promise(resolve => t.server.close(() => resolve())); + expect({ before, events: t.events, connections }).toEqual({ + before: { destroyed: false, buffered: sent.length }, + events: ["close"], + connections: 0, + }); + }); + } + + test("socket.setTimeout() fires on a socket that nobody reads", async () => { + await using t = await openTunnel(event, { + ends: true, + watch: ["finish", "timeout", "end", "error", "close"], + }); + t.client.end(belowHighWaterMark(t.tunnel)); + await barrier(t.port); + const waiting = { destroyed: t.tunnel.destroyed, events: [...t.events] }; + + t.tunnel.setTimeout(50, () => t.tunnel.destroy()); + await t.closed; + expect({ waiting, events: t.events, connections: await connectionsOf(t.server) }).toEqual({ + waiting: { destroyed: false, events: ["finish"] }, + events: ["finish", "timeout", "close"], + connections: 0, + }); + }); + + // The peer has not ended its stream, so there is no 'end'. + test("destroySoon() while the peer is silent emits 'finish', then 'close'", async () => { + let received = 0; + await using t = await openTunnel(event, { + ends: false, + watch: ["finish", "end", "error", "close"], + early: socket => void socket.on("data", (chunk: Buffer) => (received += chunk.length)), + }); + const sent = belowHighWaterMark(t.tunnel); + t.client.write(sent); + await barrier(t.port); + + t.tunnel.destroySoon(); + await t.closed; + expect({ received, events: t.events }).toEqual({ received: sent.length, events: ["finish", "close"] }); + }); + }); + } + + // More than the high water mark, so the socket reads the end of the stream while its reader is still behind. + describe("'connect': a reader that lags behind 1 MiB", () => { + const sent = patterned(1024 * 1024); + + for (const [order, listenerEndsFirst] of finOrders) { + for (const [reader, loops] of [ + ["stream.pipeline() to a slow Writable", false], + ["for await with a slow loop body", true], + ] as const) { + test(`${order}: ${reader} gets every byte`, async () => { + const chunks: Buffer[] = []; + const { promise: socketIsAhead, resolve: onSocketIsAhead } = Promise.withResolvers(); + // The reader holds its first chunk until the socket has read ahead of it. Then it takes one chunk in each + // turn of the event loop, so it stays behind the socket up to the end of the stream. + const lag = () => (chunks.length === 1 ? socketIsAhead : new Promise(resolve => setImmediate(resolve))); + const { promise: read, resolve: onRead } = Promise.withResolvers(); + await using t = await openTunnel("connect", { + ends: listenerEndsFirst, + early: socket => { + if (loops) { + (async () => { + for await (const chunk of socket) { + chunks.push(chunk); + await lag(); + } + })().then(onRead, onRead); + return; + } + const slow = new Writable({ + highWaterMark: 1, + write(chunk, encoding, callback) { + chunks.push(chunk); + lag().then(() => callback()); + }, + }); + pipeline(socket, slow, error => onRead(error)); + }, + }); + t.client.end(sent, () => { + if (!listenerEndsFirst) t.tunnel.end(); + }); + await barrier(t.port); + onSocketIsAhead(); + + await t.closed; + // The events of the socket come first, so that a socket that closes with no 'end' fails here and not by + // a timeout. + expect(t.events).toEqual(["end", "close"]); + expect({ error: await read, intact: Buffer.concat(chunks).equals(sent) }).toEqual({ + error: undefined, + intact: true, + }); + }); + } + } + }); + + // Node.js gives the listener an UpgradeStream when the body of the request is still on its way. The close of the + // connection destroys that stream, with the bytes that nobody has read. + describe("an Upgrade request with a body that arrives after the 'upgrade' listener ran", () => { + const head = + "POST / HTTP/1.1\r\nHost: example.com\r\nConnection: Upgrade\r\nUpgrade: raw\r\nContent-Length: 10\r\n\r\n"; + + for (const reads of [true, false]) { + const name = reads + ? "a reader that comes late gets nothing, and the socket emits 'close' with no 'end'" + : "a socket that nobody reads closes with the connection, and server.close() calls back"; + test(name, async () => { + const events: string[] = []; + let body = 0; + const { promise: handedOff, resolve: onHandoff } = Promise.withResolvers(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + const server = http.createServer((req, res) => void res.end("ok")); + server.on("upgrade", (req, socket) => { + for (const name of ["end", "error", "close"]) socket.on(name, () => events.push(name)); + socket.on("close", () => onClosed()); + req.on("data", (chunk: Buffer) => (body += chunk.length)); + // The listener ends the socket when it has the body of the request. + req.on("end", () => socket.end()); + socket.write(exchanges.upgrade.reply); + onHandoff(socket as net.Socket); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address() as AddressInfo; + const client = net.connect({ port, host: "127.0.0.1", allowHalfOpen: true }); + let tunnel: net.Socket | undefined; + try { + client.on("error", () => {}); + client.on("data", () => {}); + const replied = once(client, "data"); + const serverEnded = once(client, "end"); + client.write(head); + tunnel = await handedOff; + await replied; + // The listener has run. The body ends the request, and the listener ends the socket. + client.write("0123456789"); + await serverEnded; + client.end(belowHighWaterMark(tunnel)); + // The socket does not wait for a reader. + await closed; + + const atClose = { destroyed: tunnel.destroyed, events: [...events] }; + let received = 0; + if (reads) { + tunnel.on("data", (chunk: Buffer) => (received += chunk.length)); + await barrier(port); + } else { + await new Promise(resolve => server.close(() => resolve())); + } + expect({ body, atClose, received, events }).toEqual({ + body: 10, + atClose: { destroyed: true, events: ["close"] }, + received: 0, + events: ["close"], + }); + } finally { + client.destroy(); + tunnel?.destroy(); + server.close(); + } + }); + } + }); + + test("a CONNECT pipelined behind a response that never ends keeps its unread bytes", async () => { + const events: string[] = []; + let responseClosed = false; + let responseClosedAtEnd: boolean | undefined; + const { promise: handedOff, resolve: onHandoff, reject: onFailure } = Promise.withResolvers(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + const { promise: responseDone, resolve: onResponseClosed } = Promise.withResolvers(); + const server = http.createServer((req, res) => { + if (req.method === "CONNECT") onFailure(new Error("the CONNECT was dispatched as a request")); + if (req.url !== "/first") return void res.end("ok"); + res.on("close", () => { + responseClosed = true; + onResponseClosed(); + }); + }); + server.on("connect", (req, socket) => { + socket.on("end", () => { + events.push("end"); + responseClosedAtEnd = responseClosed; + }); + socket.on("error", () => events.push("error")); + socket.on("close", () => { + events.push("close"); + onClosed(); + }); + socket.write(exchanges.connect.reply); + socket.end(); + onHandoff(socket as net.Socket); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address() as AddressInfo; + const client = net.connect({ port, host: "127.0.0.1", allowHalfOpen: true }); + let tunnel: net.Socket | undefined; + try { + client.on("error", () => {}); + client.resume(); + const serverEnded = once(client, "end"); + client.write("GET /first HTTP/1.1\r\nHost: example.com\r\n\r\n" + exchanges.connect.request); + tunnel = await handedOff; + await serverEnded; + client.end("tunneled"); + await barrier(port); + + // The connection has closed. The response ahead closes with the socket, which stays for its reader. + const atClose = { destroyed: tunnel.destroyed, events: [...events], responseClosed }; + let tunneled = ""; + tunnel.on("data", chunk => (tunneled += chunk)); + await closed; + await responseDone; + expect({ atClose, tunneled, events, responseClosedAtEnd }).toEqual({ + atClose: { destroyed: false, events: [], responseClosed: false }, + tunneled: "tunneled", + events: ["end", "close"], + responseClosedAtEnd: false, + }); + } finally { + client.destroy(); + tunnel?.destroy(); + server.close(); + } + }); + + // serverB parses the socket in JavaScript, and its 'upgrade' listener gets the same socket. + test("a 'connect' socket that was emitted as a connection of a second server keeps its unread bytes", async () => { + const events: string[] = []; + const { promise: handedOff, resolve: onHandoff } = Promise.withResolvers(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + const serverA = http.createServer((req, res) => void res.end("ok")); + const serverB = http.createServer(); + serverA.on("connect", (req, socket) => { + socket.write(exchanges.connect.reply); + serverB.emit("connection", socket); + }); + serverB.on("upgrade", (req, socket) => { + for (const name of ["end", "error", "close"]) socket.on(name, () => events.push(name)); + socket.on("close", () => onClosed()); + socket.write(exchanges.upgrade.reply); + socket.end(); + onHandoff(socket as net.Socket); + }); + await once(serverA.listen(0, "127.0.0.1"), "listening"); + const { port } = serverA.address() as AddressInfo; + const client = net.connect({ port, host: "127.0.0.1", allowHalfOpen: true }); + let tunnel: net.Socket | undefined; + try { + client.on("error", () => {}); + // The 200 of serverA. The Upgrade request is for serverB. + client.once("data", () => client.write(exchanges.upgrade.request)); + client.on("data", () => {}); + const serverEnded = once(client, "end"); + client.write(exchanges.connect.request); + tunnel = await handedOff; + await serverEnded; + const sent = belowHighWaterMark(tunnel); + client.end(sent); + await barrier(port); + const waiting = { destroyed: tunnel.destroyed, events: [...events] }; + + const chunks: Buffer[] = []; + tunnel.on("data", (chunk: Buffer) => chunks.push(chunk)); + await closed; + // serverA accepted the connection, so it is the server that stops counting it. + const connectionsOfA = await connectionsOf(serverA); + await new Promise(resolve => serverA.close(() => resolve())); + expect({ waiting, intact: Buffer.concat(chunks).equals(sent), events, connectionsOfA }).toEqual({ + waiting: { destroyed: false, events: [] }, + intact: true, + events: ["end", "close"], + connectionsOfA: 0, + }); + } finally { + client.destroy(); + tunnel?.destroy(); + serverA.close(); + } + }); + + // https://nodejs.org/api/http.html#event-connect + test("the tunneling proxy of the Node.js documentation keeps a socket whose last bytes nobody reads", async () => { + const origin = net.createServer(socket => { + socket.on("error", () => {}); + socket.end("from the origin"); + }); + const proxy = http.createServer((req, res) => { + res.writeHead(200, { "Content-Type": "text/plain" }); + res.end("okay"); + }); + const { promise: handedOff, resolve: onHandoff } = Promise.withResolvers<[net.Socket, net.Socket]>(); + proxy.on("connect", (req, clientSocket, head) => { + const { port, hostname } = new URL(`http://${req.url}`); + const serverSocket = net.connect(Number(port) || 80, hostname, () => { + clientSocket.write("HTTP/1.1 200 Connection Established\r\n" + "Proxy-agent: Node.js-Proxy\r\n" + "\r\n"); + serverSocket.write(head); + serverSocket.pipe(clientSocket); + clientSocket.pipe(serverSocket); + }); + onHandoff([clientSocket as net.Socket, serverSocket]); + }); + await once(origin.listen(0, "127.0.0.1"), "listening"); + await once(proxy.listen(0, "127.0.0.1"), "listening"); + const { port } = proxy.address() as AddressInfo; + const client = net.connect({ port, host: "127.0.0.1", allowHalfOpen: true }); + let sockets: net.Socket[] = []; + try { + client.on("error", () => {}); + client.resume(); + const proxyEnded = once(client, "end"); + const target = `127.0.0.1:${(origin.address() as AddressInfo).port}`; + client.write(`CONNECT ${target} HTTP/1.1\r\nHost: ${target}\r\n\r\n`); + const [tunnel, originSide] = (sockets = await handedOff); + // The origin ends its stream, so the proxy ends the tunnel. + await proxyEnded; + client.end(Buffer.alloc(24, "x")); + await barrier(port); + // The proxy's socket to the origin closes as well. + if (!originSide.closed) await once(originSide, "close"); + + // The end of the origin's stream removes the pipe to the origin, and unpipe() leaves the socket paused. So + // nobody takes the 24 bytes, and the socket stays until its owner reads or destroys it. + expect({ closed: tunnel.closed, paused: tunnel.isPaused(), buffered: tunnel.readableLength }).toEqual({ + closed: false, + paused: true, + buffered: 24, + }); + } finally { + client.destroy(); + for (const socket of sockets) socket.destroy(); + proxy.close(); + origin.close(); + } + }); +}); diff --git a/test/js/node/http/node-http-connect.test.ts b/test/js/node/http/node-http-connect.test.ts index aba4e11854e7..f8d006be0419 100644 --- a/test/js/node/http/node-http-connect.test.ts +++ b/test/js/node/http/node-http-connect.test.ts @@ -7,8 +7,7 @@ import https from "node:https"; import type { AddressInfo } from "node:net"; import net from "node:net"; import { join } from "node:path"; -import type { Duplex } from "node:stream"; -import { Writable } from "node:stream"; +import { Duplex, Writable } from "node:stream"; import tls from "node:tls"; import { WebSocketServer } from "ws"; function connectClient(proxyAddress: AddressInfo, targetAddress: AddressInfo, add_http_prefix: boolean) { @@ -291,11 +290,17 @@ describe("HTTP server CONNECT", () => { return server; } - function connect(server: http.Server) { + function connect(server: http.Server, options: { allowHalfOpen?: boolean; servername?: string } = {}) { const { port } = server.address() as AddressInfo; const socket = secure - ? tls.connect({ port, host: "127.0.0.1", rejectUnauthorized: false }) - : net.connect(port, "127.0.0.1"); + ? tls.connect({ port, host: "127.0.0.1", rejectUnauthorized: false, ...options }) + : net.connect({ port, host: "127.0.0.1", ...options }); + sockets.push(socket); + return socket; + } + + // A socket that the test made itself. It goes with the other sockets of the test. + function track(socket: T) { sockets.push(socket); return socket; } @@ -339,6 +344,7 @@ describe("HTTP server CONNECT", () => { return { listen, connect, + track, barrier, untilReadsStop, openTunnel, @@ -410,6 +416,120 @@ describe("HTTP server CONNECT", () => { expect({ intact: Buffer.concat(chunks).equals(payload), drains }).toEqual({ intact: true, drains: 0 }); }); + // The listener of a CONNECT request, or of an Upgrade request without a body, gets the connection as a + // net.Socket. Like any net.Socket it closes after its own 'end' and 'finish', so the bytes that its reader + // has not taken outlive the connection. node-http-connect.node.mts has the http rows, which Node.js runs too. + // Over TLS the session goes with the connection, and the socket still answers for its peer. Node.js v26.3.0 + // gives the same results, except that its socket has no remoteAddress when the connection has closed. + if (secure) { + describe("the connection closes before the reader has taken every byte", () => { + // A TLS client that is half-open, so that it still sends after the server's close_notify. Resolves when the + // listener has answered, and when its close_notify has arrived if it ends the socket. The function that it + // resolves with sends the payload, then the close_notify of the client. + async function halfOpenPeer(t: ReturnType, server: http.Server, serverEndsFirst: boolean) { + const client = t.connect(server, { allowHalfOpen: true, servername: "localhost" }); + client.on("data", () => {}); + const ready = once(client, serverEndsFirst ? "end" : "data"); + client.write(handshake); + await unlessClosed(client, ready); + return async (sent: Buffer) => void client.end(sent); + } + + // A TLS client that writes its records to `wire`, and the TCP socket `raw` sends them. After the handshake + // the client gets nothing more from the server: the close_notify of the listener would end its session, + // and it has a payload to send. Resolves when the FIN of the listener has arrived. The function that it + // resolves with sends the payload, then a FIN that no close_notify precedes. + async function peerOverWire(t: ReturnType, server: http.Server) { + const { port } = server.address() as AddressInfo; + const raw = t.track(net.connect({ port, host: "127.0.0.1", allowHalfOpen: true })); + const wire = new Duplex({ + read() {}, + write(chunk, encoding, callback) { + raw.write(chunk, callback); + }, + }); + const toClient = (chunk: Buffer) => void wire.push(chunk); + raw.on("data", toClient); + raw.on("error", () => {}); + const client = t.track(tls.connect({ socket: wire, rejectUnauthorized: false, servername: "localhost" })); + client.on("error", () => {}); + await once(client, "secureConnect"); + raw.off("data", toClient).resume(); + const serverEnded = once(raw, "end"); + client.write(handshake); + await serverEnded; + return async (sent: Buffer) => { + await new Promise(resolve => client.write(sent, resolve)); + raw.end(); + }; + } + + test.each([ + ["the listener ends the socket, then the peer sends and ends", true, true], + ["the peer sends and ends, then the server ends the socket", false, true], + // A FIN with no close_notify ends the stream of a TLSSocket of Node.js v26.3.0 too. + ["the listener ends the socket, then the peer sends and leaves with no close_notify", true, false], + ] as const)("%s", async (order, serverEndsFirst, closeNotify) => { + const t = setup(); + const events: string[] = []; + const { promise: tunnelPromise, resolve: onTunnel } = Promise.withResolvers(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + await using server = await t.listen(socket => { + socket.on("end", () => events.push("end")); + socket.on("error", error => events.push(String(error))); + socket.on("close", () => { + events.push("close"); + onClosed(); + }); + socket.write(accept); + if (serverEndsFirst) socket.end(); + onTunnel(socket); + }); + using _ = t.destroySockets(); + const leave = await (closeNotify ? halfOpenPeer(t, server, serverEndsFirst) : peerOverWire(t, server)); + const tunnel: net.Socket & { servername?: string | false | null } = await tunnelPromise; + + // Half of the high water mark: the socket reads all of it with nobody reading the socket. + const sent = payload.subarray(0, tunnel.readableHighWaterMark / 2); + const writtenBeforeFin = tunnel.bytesWritten; + await leave(sent); + await t.barrier(server); + if (!serverEndsFirst) { + tunnel.end(); + await t.barrier(server); + } + + // Both sides have sent their FIN, so the connection is gone. The socket stays for its reader, and it + // still answers for its peer. + const atClose = { + destroyed: tunnel.destroyed, + events: [...events], + belowMark: tunnel.readableLength < tunnel.readableHighWaterMark, + servername: tunnel.servername, + remoteAddress: tunnel.remoteAddress, + bytesWritten: tunnel.bytesWritten, + }; + // The reader is a 'data' listener that comes now. + const chunks: Buffer[] = []; + tunnel.on("data", (chunk: Buffer) => chunks.push(chunk)); + await closed; + expect({ writtenBeforeFin, atClose, intact: Buffer.concat(chunks).equals(sent), events }).toEqual({ + writtenBeforeFin: accept.length, + atClose: { + destroyed: false, + events: [], + belowMark: true, + servername: "localhost", + remoteAddress: "127.0.0.1", + bytesWritten: accept.length, + }, + intact: true, + events: ["end", "close"], + }); + }); + }); + } + // An open Upgrade tunnel runs the same code as a CONNECT tunnel. The tests below move 8 to 24 MiB // each, on one thread, so they run for CONNECT only. if (kind === "Upgrade") return; @@ -535,8 +655,6 @@ describe("HTTP server CONNECT", () => { expect(tunnel.readableLength).toBe(buffered); }); - // The reader takes each chunk at once here. A reader that falls behind still loses what it has - // not taken when the connection closes. test("end() behind a full buffer leaves the reads stopped, and the rest, 'end' and 'close' still arrive", async () => { const t = setup(); const { promise: tunnelPromise, resolve: onTunnel } = Promise.withResolvers(); @@ -1924,6 +2042,536 @@ test("a half-open tunnel with bytes left to send keeps the process alive until t } }); +// The socket of a 'connect' listener, or of an 'upgrade' listener whose request has no body, is the connection as a +// net.Socket. When the connection closes after the peer's FIN and the reader has not taken every byte, the socket +// waits for its reader. node-http-connect.node.mts has the tests that Node.js v26.3.0 passes too. The tests here +// are for what Bun does in its own way, and for what only Bun can observe. Each says what Node.js does. +describe("a tunnel socket whose connection has closed", () => { + const exchanges = { + connect: { + request: "CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n", + reply: "HTTP/1.1 200 Connection Established\r\n\r\n", + }, + upgrade: { + request: "GET / HTTP/1.1\r\nHost: example.com\r\nConnection: Upgrade\r\nUpgrade: raw\r\n\r\n", + reply: "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: raw\r\n\r\n", + }, + }; + // An application data record of 32 bytes that are not ciphertext. + const recordThatDoesNotDecrypt = Buffer.concat([Buffer.from([0x17, 0x03, 0x03, 0x00, 0x20]), Buffer.alloc(32, 7)]); + + type TunnelOptions = { + // The listener ends the socket behind its reply. + ends: boolean; + // Listens on an AF_UNIX socket. + path?: string; + secure?: boolean; + // false: the socket keeps the listeners that it was handed off with, and none of them is for 'error'. + errorListener?: boolean; + // Replaces the request of the exchange. + request?: string; + // Runs in the listener. + onTunnel?: (socket: net.Socket) => void; + }; + + // The client is half-open: it can send after the server's FIN. Over TLS it writes its records to a Duplex, and + // `raw` sends them. So a test can reset the connection under the session, or put a record of its own behind them. + async function openTunnel(event: "connect" | "upgrade", options: TunnelOptions) { + const events: string[] = []; + const { promise: handedOff, resolve: onTunnel } = Promise.withResolvers(); + const { promise: closed, resolve: onClosed } = Promise.withResolvers(); + const listener = (req: http.IncomingMessage, res: http.ServerResponse) => void res.end("ok"); + const server = options.secure + ? https.createServer({ key: tlsCert.key, cert: tlsCert.cert }, listener) + : http.createServer(listener); + server.on(event, (req, socket) => { + socket.on("end", () => events.push("end")); + if (options.errorListener !== false) { + socket.on("error", (error: NodeJS.ErrnoException) => events.push(String(error.code))); + } + socket.on("close", () => { + events.push("close"); + onClosed(); + }); + options.onTunnel?.(socket as net.Socket); + socket.write(exchanges[event].reply); + if (options.ends) socket.end(); + onTunnel(socket as net.Socket); + }); + await once(options.path ? server.listen(options.path) : server.listen(0, "127.0.0.1"), "listening"); + const connect = (more: net.SocketConstructorOpts = {}) => + options.path + ? net.connect({ path: options.path, ...more }) + : net.connect({ port: (server.address() as AddressInfo).port, host: "127.0.0.1", ...more }); + + const raw = connect({ allowHalfOpen: true }); + raw.on("error", () => {}); + let client = raw; + if (options.secure) { + const wire = new Duplex({ + read() {}, + write(chunk, encoding, callback) { + raw.write(chunk, callback); + }, + }); + raw.on("data", chunk => wire.push(chunk)); + raw.on("end", () => wire.push(null)); + client = tls.connect({ socket: wire, rejectUnauthorized: false }); + client.on("error", () => {}); + await once(client, "secureConnect"); + } + client.on("data", () => {}); + const ready = once(client, options.ends ? "end" : "data"); + client.write(options.request ?? exchanges[event].request); + await ready; + const tunnel = await handedOff; + return { + server, + tunnel, + client, + raw, + events, + closed, + // Half of the high water mark: the socket reads all of it with nobody reading the socket. + payload: Buffer.alloc(tunnel.readableHighWaterMark / 2, "x"), + connections: () => new Promise(resolve => server.getConnections((error, count) => resolve(count))), + // One exchange on a connection of its own: the server has read what the client sent before it. + async barrier() { + const socket = options.secure + ? tls.connect({ port: (server.address() as AddressInfo).port, host: "127.0.0.1", rejectUnauthorized: false }) + : connect(); + socket.resume(); + socket.end("GET /barrier HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n"); + await once(socket, "close"); + }, + async [Symbol.asyncDispose]() { + raw.destroy(); + client.destroy(); + tunnel.destroy(); + server.close(); + }, + }; + } + + // A connection that closes with no FIN of the peer, and not from this side: the peer resets it, or its TLS session + // fails. That is not the end of the peer's stream, so the socket never waits for a reader. It closes with the + // connection, after an 'end' if its reader has taken every byte, and with no 'error'. Node.js emits no 'end'. It + // emits 'error': ECONNRESET and then 'close' for a reset, and ERR_SSL_DECRYPTION_FAILED_OR_BAD_RECORD_MAC for the + // record. With no 'error' listener, the reset of a plain socket is an uncaught exception there. + describe.each([ + ["http", "a reset of the peer"], + ["https", "a reset of the peer"], + ["https", "a record that does not decrypt"], + ] as const)("%s: %s closes the socket with the connection", (protocol, cause) => { + for (const event of ["connect", "upgrade"] as const) { + for (const errorListener of [true, false]) { + for (const reads of [true, false]) { + const listener = errorListener ? "an 'error' listener" : "no 'error' listener"; + const outcome = reads + ? "a reader that keeps up gets 'end', then 'close'" + : "bytes that nobody has read go with the socket"; + test(`'${event}' with ${listener}: ${outcome}`, async () => { + let received = 0; + const read = (socket: net.Socket) => void socket.on("data", chunk => (received += chunk.length)); + await using t = await openTunnel(event, { + ends: false, + secure: protocol === "https", + errorListener, + onTunnel: reads ? read : undefined, + }); + await new Promise(resolve => t.client.write(t.payload, resolve)); + await t.barrier(); + + if (cause === "a reset of the peer") t.raw.resetAndDestroy(); + else await new Promise(resolve => t.raw.write(recordThatDoesNotDecrypt, resolve)); + await t.barrier(); + const atClose = { + destroyed: t.tunnel.destroyed, + events: [...t.events], + connections: await t.connections(), + }; + + // A reader that comes now gets nothing. + if (!reads) { + read(t.tunnel); + await t.barrier(); + } + expect({ atClose, received }).toEqual({ + atClose: { destroyed: true, events: reads ? ["end", "close"] : ["close"], connections: 0 }, + received: reads ? t.payload.length : 0, + }); + }); + } + } + } + }); + + // A relay moves the bytes with two pipe() calls, and the 'end' of the tunnel socket is what ends its upstream leg. + // In Node.js the reset is an 'error' on the tunnel socket. The WebSocket pass of http-proxy listens for it and ends + // the upstream leg there. A relay with no listener stops the process with an uncaught exception. + test.each([ + ["connect", "two pipe() calls", false], + ["upgrade", "two pipe() calls", false], + ["upgrade", "two pipe() calls and the 'error' listener of http-proxy", true], + ] as const)( + "'%s': a relay of %s ends its upstream leg when the client resets", + async (event, shape, likeHttpProxy) => { + const upstreamEvents: string[] = []; + const { promise: upstreamClosed, resolve: onUpstreamClosed } = Promise.withResolvers(); + // Echoes, so that the client knows when its bytes have been through the relay. + await using upstreamServer = net.createServer(connection => { + connection.on("end", () => upstreamEvents.push("end")); + connection.on("close", () => { + upstreamEvents.push("close"); + onUpstreamClosed(); + }); + connection.pipe(connection); + }); + await once(upstreamServer.listen(0, "127.0.0.1"), "listening"); + await using server = http.createServer(); + server.on(event, (req, socket) => { + const upstream = net.connect((upstreamServer.address() as AddressInfo).port, "127.0.0.1", () => { + socket.write(exchanges[event].reply); + upstream.pipe(socket); + socket.pipe(upstream); + }); + upstream.on("error", () => {}); + if (likeHttpProxy) socket.on("error", () => upstream.end()); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + + const client = net.connect((server.address() as AddressInfo).port, "127.0.0.1"); + try { + client.on("error", () => {}); + const { promise: echoed, resolve: onEchoed } = Promise.withResolvers(); + let received = ""; + client.on("data", chunk => { + received += chunk; + if (received === exchanges[event].reply) client.write("ping"); + if (received.endsWith("ping")) onEchoed(); + }); + client.write(exchanges[event].request); + await echoed; + client.resetAndDestroy(); + await upstreamClosed; + expect(upstreamEvents).toEqual(["end", "close"]); + } finally { + client.destroy(); + } + }, + ); + + // The socket of an Upgrade request with a body behaves like Node's UpgradeStream, which never waits: the close of + // the connection destroys it. A reset ends it like the socket of a request with no body. Node.js emits 'error' + // with ECONNRESET on its UpgradeStream, and no 'end'. + test("'upgrade' with a request body: a reset of the peer closes the socket, after 'end' for a reader that keeps up", async () => { + let received = ""; + await using t = await openTunnel("upgrade", { + ends: false, + request: exchanges.upgrade.request.replace("\r\n\r\n", "\r\nContent-Length: 4\r\n\r\n"), + onTunnel: socket => void socket.on("data", chunk => (received += chunk)), + }); + // The 101 has arrived. The body follows it, and bytes for the socket follow the body. + t.client.write("bodytunneled"); + await t.barrier(); + t.client.resetAndDestroy(); + await t.barrier(); + expect({ destroyed: t.tunnel.destroyed, events: t.events, received }).toEqual({ + destroyed: true, + events: ["end", "close"], + received: "tunneled", + }); + }); + + // A hangup: the connection closes after the peer's FIN, and the listener has not ended the socket. The peer of an + // AF_UNIX socket closes both ways at once, which epoll and kqueue report as a hangup. Those two rows are for that + // report, so Windows does not run them. A TCP peer does the same on every platform with its FIN and then a reset. + describe.each([ + ["AF_UNIX, the peer closes and the server has not ended the socket", true], + ["TCP, the peer ends and then resets, and the server has not ended the socket", false], + ] as const)("%s", (transport, unix) => { + // The peer leaves behind its payload. Resolves when the connection has closed on the server. + async function hangUp(t: Awaited>) { + if (unix) { + t.client.end(t.payload, () => t.client.destroy()); + await once(t.client, "close"); + } else { + t.client.end(t.payload); + await t.barrier(); + t.client.resetAndDestroy(); + } + await t.barrier(); + } + + // Node.js v26.3.0 gives the same result. + test.skipIf(unix && isWindows)("a late reader gets every byte, then 'end'", async () => { + using dir = unix ? tempDir("connect-unix-unread", {}) : undefined; + await using t = await openTunnel("connect", { ends: false, path: dir && join(String(dir), "tunnel.sock") }); + await hangUp(t); + + const atClose = { destroyed: t.tunnel.destroyed, events: [...t.events] }; + let received = 0; + t.tunnel.on("data", chunk => (received += chunk.length)); + t.tunnel.on("end", () => t.tunnel.end()); + await t.closed; + expect({ atClose, received, events: t.events }).toEqual({ + atClose: { destroyed: false, events: [] }, + received: t.payload.length, + events: ["end", "close"], + }); + }); + + // Node.js fails the write with EPIPE and emits 'error'. Bun fails it like a write that waits for a drain when + // the connection closes, with no 'error'. Windows does not run this: it does not report the reset that follows a + // FIN, so the connection is still open there when the write comes. + test.skipIf(isWindows)("a write fails, and the unread bytes go with the socket", async () => { + using dir = unix ? tempDir("connect-unix-write", {}) : undefined; + await using t = await openTunnel("connect", { ends: false, path: dir && join(String(dir), "tunnel.sock") }); + await hangUp(t); + + const atClose = { destroyed: t.tunnel.destroyed, events: [...t.events] }; + const { promise: written, resolve: onWritten } = Promise.withResolvers(); + t.tunnel.write("x", onWritten); + const error = (await written) as NodeJS.ErrnoException | undefined; + await t.closed; + expect({ atClose, write: error?.code, events: t.events }).toEqual({ + atClose: { destroyed: false, events: [] }, + write: "ERR_STREAM_DESTROYED", + events: ["close"], + }); + }); + }); + + // The peer's FIN has arrived, nobody has read the payload, and then the connection closes. Two sockets do not wait + // for their reader then. + describe("a socket does not wait for its reader", () => { + // Its owner asked for the close. Node.js v26.3.0 gives the same result. + test("when destroySoon() closes the connection", async () => { + await using t = await openTunnel("connect", { ends: false }); + t.client.end(t.payload); + await t.barrier(); + const halfOpen = { destroyed: t.tunnel.destroyed, buffered: t.tunnel.readableLength }; + + // It sends the FIN of this side, so both sides have sent theirs. + t.tunnel.destroySoon(); + await t.barrier(); + const atClose = { destroyed: t.tunnel.destroyed, events: [...t.events], connections: await t.connections() }; + + // A reader that comes now gets nothing. + let received = 0; + t.tunnel.on("data", chunk => (received += chunk.length)); + await t.barrier(); + expect({ halfOpen, atClose, received }).toEqual({ + halfOpen: { destroyed: false, buffered: t.payload.length }, + atClose: { destroyed: true, events: ["close"], connections: 0 }, + received: 0, + }); + }); + + // The write has failed, and its callback has to hear that. Node.js fails the writes with EPIPE and emits 'error'. + // Windows does not run this: Winsock takes megabytes whole, so no write is sure to wait there. + test.skipIf(isWindows)("when a write waits for a drain", async () => { + await using t = await openTunnel("connect", { ends: false }); + const failed: (string | undefined)[] = []; + const onWritten = (error?: NodeJS.ErrnoException | null) => void (error && failed.push(error.code)); + // The client does not read. This is more than a loopback socket takes, so most of it waits for a drain. + t.client.pause(); + t.tunnel.write(Buffer.alloc(16 * 1024 * 1024, "w"), onWritten); + t.tunnel.write("behind", onWritten); + + t.client.end(t.payload); + await t.barrier(); + const halfOpen = { + destroyed: t.tunnel.destroyed, + buffered: t.tunnel.readableLength, + unsent: t.tunnel.writableLength > 0, + failed: [...failed], + }; + + // A hangup: the reset closes the connection behind the FIN. + t.client.resetAndDestroy(); + await t.barrier(); + expect({ + halfOpen, + atClose: { destroyed: t.tunnel.destroyed, events: t.events, connections: await t.connections() }, + failed, + }).toEqual({ + halfOpen: { destroyed: false, buffered: t.payload.length, unsent: true, failed: [] }, + atClose: { destroyed: true, events: ["close"], connections: 0 }, + // The write that waited, and the write behind it. + failed: ["ERR_STREAM_DESTROYED", "ERR_STREAM_DESTROYED"], + }); + }); + }); + + // The server counts connections, and this one has closed: the socket that waits has no file descriptor. So the + // server does not count it, and server.close() does not wait for it. Node.js keeps the handle of such a socket, + // counts it until it closes, and holds server.close() for it. + test.each(["connect", "upgrade"] as const)( + "'%s': a socket that waits is not a connection of its server", + async event => { + await using t = await openTunnel(event, { ends: true }); + t.client.end(t.payload); + await t.barrier(); + + let drops = 0; + t.server.on("drop", () => drops++); + t.server.maxConnections = 1; + // One exchange on a new connection. Resolves with what the client received. + const { promise: response, resolve: onResponse } = Promise.withResolvers(); + const next = net.connect((t.server.address() as AddressInfo).port, "127.0.0.1"); + let received = ""; + next.on("error", () => {}); + next.on("data", chunk => (received += chunk)); + next.on("close", () => onResponse(received)); + next.end("GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n"); + const waiting = { + served: (await response).slice(-2), + drops, + connections: await t.connections(), + destroyed: t.tunnel.destroyed, + events: [...t.events], + }; + + await new Promise(resolve => t.server.close(() => resolve())); + const afterServerClose = { destroyed: t.tunnel.destroyed, events: [...t.events] }; + let read = 0; + t.tunnel.on("data", chunk => (read += chunk.length)); + await t.closed; + expect({ waiting, afterServerClose, read, events: t.events }).toEqual({ + waiting: { served: "ok", drops: 0, connections: 0, destroyed: false, events: [] }, + afterServerClose: { destroyed: false, events: [] }, + read: t.payload.length, + events: ["end", "close"], + }); + }, + ); + + // Nothing of the server keeps a socket that waits: one that its listener dropped is collected. Node.js keeps the + // socket, and its file descriptor. + test("a socket that waits and that its owner dropped is collected", async () => { + const count = 20; + const refs: WeakRef[] = []; + let highWaterMark = 0; + const server = http.createServer((req, res) => void res.end("ok")); + server.on("connect", (req, socket) => { + refs.push(new WeakRef(socket)); + highWaterMark = socket.readableHighWaterMark; + socket.write(exchanges.connect.reply); + socket.end(); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + const { port } = server.address() as AddressInfo; + const connections = () => new Promise(resolve => server.getConnections((error, n) => resolve(n))); + const clients: net.Socket[] = []; + const closed: Promise[] = []; + try { + for (let i = 0; i < count; i++) { + const client = net.connect({ port, host: "127.0.0.1", allowHalfOpen: true }); + clients.push(client); + closed.push(once(client, "close")); + client.on("error", () => {}); + // Every second connection has had a request before its CONNECT, and the keep-alive timer of that request + // is still armed. A timer keeps its socket. + if (i % 2) { + client.write("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"); + await once(client, "data"); + } + client.resume(); + // The server's FIN. The client answers with bytes that nobody reads, and its own FIN. + client.on("end", () => client.end(Buffer.alloc(highWaterMark / 2, "x"))); + client.write(exchanges.connect.request); + } + await Promise.all(closed); + await untilTheServerHasRead(() => net.connect(port, "127.0.0.1")); + const waiting = await connections(); + + // deref() keeps its target for the rest of the turn that calls it, so each collection has a turn of its own. + // A server that kept its sockets would keep all of them, so most of them have to go. + const deadline = Date.now() + 2000; + let alive = refs.length; + do { + await new Promise(resolve => setImmediate(resolve)); + Bun.gc(true); + alive = refs.filter(ref => ref.deref() !== undefined).length; + } while (alive >= 5 && Date.now() < deadline); + expect({ sockets: refs.length, waiting, collected: alive < 5 }).toEqual({ + sockets: count, + waiting: 0, + collected: true, + }); + } finally { + for (const client of clients) client.destroy(); + server.close(); + } + }); + + // Nobody reads the sockets, so they do not close. Their connections have closed, so the servers close, and the + // process does not stay for them: the child exits. In Node.js v26.3.0 the servers do not close. + test("a socket that its listener ended and never reads holds neither server.close() nor the process", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const http = require("node:http"); + const net = require("node:net"); + const exchanges = ${JSON.stringify(exchanges)}; + const results = {}; + for (const event of ["connect", "upgrade"]) { + for (const when of ["before", "after"]) { + const events = (results[event + ", server.close() " + when + " the peer's FIN"] = []); + let highWaterMark; + const server = http.createServer((req, res) => res.end("ok")); + const close = () => server.close(() => events.push("server close")); + server.on(event, (req, socket) => { + highWaterMark = socket.readableHighWaterMark; + socket.on("end", () => events.push("socket end")); + socket.on("close", () => events.push("socket close")); + socket.write(exchanges[event].reply); + socket.end(); + if (when === "before") close(); + }); + server.listen(0, "127.0.0.1", () => { + const { port } = server.address(); + const client = net.connect({ port, host: "127.0.0.1", allowHalfOpen: true }); + client.resume(); + client.write(exchanges[event].request); + // The server's FIN. The client answers with bytes that nobody reads, and its own FIN. + client.on("end", () => client.end(Buffer.alloc(highWaterMark / 2, "x"))); + if (when === "before") return; + client.on("close", () => { + // One exchange on a connection of its own: the server has read the client's FIN when it is over. + const barrier = net.connect(port, "127.0.0.1"); + barrier.resume(); + barrier.end("GET /barrier HTTP/1.1\\r\\nHost: example.com\\r\\nConnection: close\\r\\n\\r\\n"); + barrier.on("close", () => { + server.getConnections((error, count) => { + events.push("connections " + count); + close(); + }); + }); + }); + }); + } + } + process.on("exit", () => console.log(JSON.stringify(results)));`, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + // No 'end' and no 'close' of a socket. + const results = { + "connect, server.close() before the peer's FIN": ["server close"], + "connect, server.close() after the peer's FIN": ["connections 0", "server close"], + "upgrade, server.close() before the peer's FIN": ["server close"], + "upgrade, server.close() after the peer's FIN": ["connections 0", "server close"], + }; + expect({ stdout, stderr, exitCode }).toEqual({ stdout: JSON.stringify(results) + "\n", stderr: "", exitCode: 0 }); + }); +}); + // A listener with nothing to send ends its side in the listener. It still reads: an http.Server // socket has allowHalfOpen. Every expectation below is Node v26.3.0's. describe("a tunnel behind a request head that took more than one read", () => {