diff --git a/packages/bun-usockets/src/crypto/openssl.c b/packages/bun-usockets/src/crypto/openssl.c index 412381c3bbc3..21d12c0398e2 100644 --- a/packages/bun-usockets/src/crypto/openssl.c +++ b/packages/bun-usockets/src/crypto/openssl.c @@ -1821,7 +1821,23 @@ struct us_socket_t *us_internal_ssl_on_writable(struct us_socket_t *s) { struct loop_ssl_data *loop_ssl_data = (struct loop_ssl_data *)s->group->loop->data.ssl_data; /* Ciphertext from a partial batch flush goes out before anything else; * while it is pending nothing new may be written for this socket. */ + unsigned int spill_off_before = loop_ssl_data ? loop_ssl_data->ssl_spill_off : 0; if (loop_ssl_data && !ssl_drain_spill(loop_ssl_data, s)) { + /* A writable event that moves zero spill bytes after the peer's + * readable side has already ended means the peer is gone (send() + * hit EPIPE/ECONNRESET, folded to 0 by us_socket_raw_write) and + * this spill will never drain. Returning here would spin the + * re-armed writable poll on kqueue, since EPOLLERR is not delivered + * there. Mark the SSL fatal so us_internal_ssl_write returns 0 (the + * uWS layer's flushed==0-after-FIN guard, or hasFullyDrained() when + * nothing is buffered, then closes the connection on this dispatch) + * and dispatch directly, bypassing the is_shut_down gate below that + * ssl_fatal_error would otherwise trip. */ + if (s->ssl_end_delivered && loop_ssl_data->ssl_spill_off == spill_off_before) { + ssl_release_spill(s->group->loop, s); + s->ssl_fatal_error = 1; + return us_dispatch_writable(s); + } return s; } if (s->ssl_shutdown_after_spill) { @@ -2098,6 +2114,17 @@ void *us_internal_ssl_get_native_handle(struct us_socket_t *s) { return s->ssl ? s_ssl(s) : NULL; } +/* Ciphertext bytes already sealed for `s` and counted as written by + * us_internal_ssl_write, still waiting on a writable event to reach the + * kernel. uWS's AsyncSocket::hasFullyDrained() checks this so the HTTP + * close-after-drain gates do not fire while the last batch is still in + * userspace. */ +unsigned int us_internal_ssl_spill_pending(struct us_socket_t *s) { + struct loop_ssl_data *loop_ssl_data = (struct loop_ssl_data *)s->group->loop->data.ssl_data; + if (!loop_ssl_data || loop_ssl_data->ssl_spill_owner != s) return 0; + return loop_ssl_data->ssl_spill_len - loop_ssl_data->ssl_spill_off; +} + int us_internal_ssl_write(struct us_socket_t *s, const char *data, int length) { if (us_socket_is_closed(s) || us_internal_ssl_is_shut_down(s) || length == 0) return 0; diff --git a/packages/bun-usockets/src/internal/internal.h b/packages/bun-usockets/src/internal/internal.h index 27148b318f55..b785d3938487 100644 --- a/packages/bun-usockets/src/internal/internal.h +++ b/packages/bun-usockets/src/internal/internal.h @@ -224,6 +224,7 @@ int us_internal_ssl_handshake_callback_has_fired(us_socket_r s); int us_internal_ssl_is_shut_down(us_socket_r s); void us_internal_ssl_shutdown(us_socket_r s); int us_internal_ssl_write(us_socket_r s, const char *data, int length); +unsigned int us_internal_ssl_spill_pending(us_socket_r s); void *us_internal_ssl_get_native_handle(us_socket_r s); struct us_bun_verify_error_t us_internal_ssl_verify_error(us_socket_r s); void *us_internal_ssl_sni_userdata(us_socket_r s); diff --git a/packages/bun-usockets/src/libusockets.h b/packages/bun-usockets/src/libusockets.h index 7a7d72016b01..e763eb8c73f6 100644 --- a/packages/bun-usockets/src/libusockets.h +++ b/packages/bun-usockets/src/libusockets.h @@ -632,6 +632,11 @@ int us_socket_is_closed(us_socket_r s) nonnull_fn_decl; int us_socket_is_tls(us_socket_r s) nonnull_fn_decl; int us_socket_is_ssl_handshake_finished(us_socket_r s) nonnull_fn_decl; int us_socket_ssl_handshake_callback_has_fired(us_socket_r s) nonnull_fn_decl; +/* TLS ciphertext bytes already sealed for this socket and reported as + * written by us_socket_write(), still waiting on a writable event to reach + * the kernel (the loop-wide spill slot owned by this socket). 0 for + * plain-TCP sockets and for TLS sockets with nothing spilled. */ +unsigned int us_socket_ssl_spill_pending(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))); diff --git a/packages/bun-usockets/src/loop.c b/packages/bun-usockets/src/loop.c index 987120f6bda6..441bdad08f69 100644 --- a/packages/bun-usockets/src/loop.c +++ b/packages/bun-usockets/src/loop.c @@ -757,6 +757,19 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in } #undef LOOP_ISNT_VERY_BUSY_THRESHOLD #else + /* Windows eof-drain, same as the POSIX branch above: + * poll_cb maps AFD DISCONNECT to the eof hint for a + * socket whose write side we already shut down, and + * AFD reports DISCONNECT with the tail of the peer's + * stream still queued in the kernel. Stopping here + * lets the is_shut_down raw-close below discard it + * (a half-closed TLS/net client dropping the end of + * a large response on Windows only). recv() returning + * 0 or WSAEWOULDBLOCK ends the loop, so this is + * bounded by the kernel receive buffer. */ + if (s && !us_socket_is_closed(s) && !s->flags.is_paused && (eof || error)) { + continue; + } /* Windows AFD_POLL_ABORT is not level-triggered the way * epoll's EPOLLHUP|EPOLLERR are: a peer RST that lands * while this poll_cb is on the stack — typically when an diff --git a/packages/bun-usockets/src/socket.c b/packages/bun-usockets/src/socket.c index 63af324915ea..25f798b1a017 100644 --- a/packages/bun-usockets/src/socket.c +++ b/packages/bun-usockets/src/socket.c @@ -157,6 +157,13 @@ int us_socket_ssl_handshake_callback_has_fired(struct us_socket_t *s) { return 1; } +unsigned int us_socket_ssl_spill_pending(struct us_socket_t *s) { + if (s->ssl) { + return us_internal_ssl_spill_pending(s); + } + return 0; +} + int us_connecting_socket_is_closed(struct us_connecting_socket_t *c) { return c->closed; } diff --git a/packages/bun-uws/src/AsyncSocket.h b/packages/bun-uws/src/AsyncSocket.h index b8a3db918195..5e2fafae7eb6 100644 --- a/packages/bun-uws/src/AsyncSocket.h +++ b/packages/bun-uws/src/AsyncSocket.h @@ -188,6 +188,21 @@ struct AsyncSocket { return getAsyncSocketData()->buffer.length(); } + /* Whether every byte handed to us_socket_write() has reached the kernel. + * For TLS, us_socket_write() can report a batch as written while its + * ciphertext still sits in the loop's spill slot (openssl.c + * ssl_flush_write_batch); the close-after-drain gates in HttpResponse / + * HttpContext must wait for that too. Kept separate from + * getBufferedAmount() so WebSocket's maxBackpressure policy and the + * JS-exposed bufferedAmount stay a plaintext count. */ + bool hasFullyDrained() { + if (getAsyncSocketData()->buffer.length()) return false; + if constexpr (SSL) { + return us_socket_ssl_spill_pending((us_socket_t *) this) == 0; + } + return true; + } + /* Returns the text representation of an IPv4 or IPv6 address */ std::string_view addressAsText(std::string_view binary) { static thread_local char buf[64]; diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index 0e6f21dc88dc..a6e243d947e9 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -206,14 +206,13 @@ struct HttpContext { /* A peer FIN must not tear the connection down at the loop level: * onEnd() below decides whether to close right away (idle) or to * keep writing the responses that are still in flight / pipelined - * (Node's socketOnEnd semantics). Without this flag the loop - * force-closes the socket right after dispatching onEnd. TLS - * (openssl.c us_internal_ssl_on_end) does not consult this flag - * and force-closes on FIN regardless, so this half of the compat - * block is http-only for now. */ - if constexpr (!SSL) { - s->flags.allow_half_open = 1; - } + * (Node's socketOnEnd semantics). Without this flag the loop (and + * openssl.c us_internal_ssl_on_end for TLS) force-closes the + * socket right after dispatching onEnd, discarding the buffered + * response bytes. onEnd's defer and onWritable's close gate use + * hasFullyDrained(), which accounts for the TLS ciphertext spill, + * so they are accurate for both transports. */ + s->flags.allow_half_open = 1; } if(!SSL) { @@ -612,7 +611,7 @@ struct HttpContext { /* We need to check if we should close this socket here now */ if (httpResponseData->shouldCloseConnection()) { if ((httpResponseData->state & HttpResponseData::HTTP_RESPONSE_PENDING) == 0) { - if (((AsyncSocket *) s)->getBufferedAmount() == 0) { + if (((AsyncSocket *) s)->hasFullyDrained()) { ((AsyncSocket *) s)->shutdown(); /* We need to force close after sending FIN since we want to hinder * clients from keeping to send their huge data */ @@ -742,7 +741,7 @@ struct HttpContext { responseDone = true; } } - if (responseDone && asyncSocket->getBufferedAmount() == 0) { + if (responseDone && asyncSocket->hasFullyDrained()) { asyncSocket->shutdown(); /* We need to force close after sending FIN since we want to hinder * clients from keeping to send their huge data */ @@ -800,7 +799,7 @@ struct HttpContext { * callback is still draining) must not be discarded by the close() * below; the connection shuts down from the shouldCloseConnection() * gates once they have flushed. */ - bool hasQueuedOutgoing = asyncSocket->getBufferedAmount() > 0 + bool hasQueuedOutgoing = !asyncSocket->hasFullyDrained() || httpResponseData->onWritable != nullptr; bool responseInFlight = httpResponseData->nodeHttpQueuedPipelinedCount > 0 || (httpResponseData->state & HttpResponseData::HTTP_RESPONSE_PENDING); diff --git a/packages/bun-uws/src/HttpResponse.h b/packages/bun-uws/src/HttpResponse.h index 080a200bbf4b..7155dfd60015 100644 --- a/packages/bun-uws/src/HttpResponse.h +++ b/packages/bun-uws/src/HttpResponse.h @@ -191,7 +191,7 @@ struct HttpResponse : public AsyncSocket { if (!Super::isCorked()) { if (httpResponseData->shouldCloseConnection()) { if ((httpResponseData->state & HttpResponseData::HTTP_RESPONSE_PENDING) == 0) { - if (((AsyncSocket *) this)->getBufferedAmount() == 0) { + if (((AsyncSocket *) this)->hasFullyDrained()) { ((AsyncSocket *) this)->shutdown(); /* We need to force close after sending FIN since we want to hinder * clients from keeping to send their huge data */ @@ -257,7 +257,7 @@ struct HttpResponse : public AsyncSocket { if (!Super::isCorked()) { if (httpResponseData->shouldCloseConnection()) { if ((httpResponseData->state & HttpResponseData::HTTP_RESPONSE_PENDING) == 0) { - if (((AsyncSocket *) this)->getBufferedAmount() == 0) { + if (((AsyncSocket *) this)->hasFullyDrained()) { ((AsyncSocket *) this)->shutdown(); /* We need to force close after sending FIN since we want to hinder * clients from keeping to send their huge data */ @@ -861,7 +861,7 @@ struct HttpResponse : public AsyncSocket { HttpResponseData *httpResponseData = getHttpResponseData(); if (httpResponseData->shouldCloseConnection()) { if ((httpResponseData->state & HttpResponseData::HTTP_RESPONSE_PENDING) == 0) { - if (((AsyncSocket *) this)->getBufferedAmount() == 0) { + if (((AsyncSocket *) this)->hasFullyDrained()) { ((AsyncSocket *) this)->shutdown(); /* We need to force close after sending FIN since we want to hinder * clients from keeping to send their huge data */ diff --git a/test/js/node/http/node-http-backpressure.test.ts b/test/js/node/http/node-http-backpressure.test.ts index c34c788e6909..af2ff4c9a463 100644 --- a/test/js/node/http/node-http-backpressure.test.ts +++ b/test/js/node/http/node-http-backpressure.test.ts @@ -6,9 +6,13 @@ * A handful of older tests do not run in Node in this file. These tests should be updated to run in Node, or deleted. */ import { once } from "node:events"; +import { readFileSync } from "node:fs"; import http from "node:http"; +import https from "node:https"; import type { AddressInfo } from "node:net"; import net from "node:net"; +import path from "node:path"; +import nodeTls from "node:tls"; describe("backpressure", () => { // Writes `total` bytes to `res` in `chunk`-sized pieces, waiting for "drain" @@ -212,6 +216,97 @@ describe("backpressure", () => { expect(ended).toBe(true); expect([BODY, BODY * 2]).toContain(body); }); + + // TLS variants of the it.each above: the server's TLS write-batch spill + // (up to one 128 KiB ciphertext batch the kernel did not fully accept) is + // reported as written by us_socket_write() while it sits in userspace, so + // the post-FIN close gate (hasFullyDrained()) must wait for it. Looped a + // few times so the on_writable drain cycle is exercised past the first + // kernel-accepted write. This is also the client-side regression test for + // the Windows eof-drain (a half-closed client must read out the kernel + // receive buffer when AFD DISCONNECT is mapped to eof). + describe("https", () => { + const keysDir = path.join(import.meta.dirname, "..", "test", "fixtures", "keys"); + const tlsOptions = { + cert: readFileSync(path.join(keysDir, "agent1-cert.pem")), + key: readFileSync(path.join(keysDir, "agent1-key.pem")), + }; + + async function halfCloseTlsRequestBodyBytes(port: number): Promise<{ body: number; ended: boolean }> { + const socket = nodeTls.connect({ port, host: "127.0.0.1", rejectUnauthorized: false }); + let body = 0; + let head = ""; + let gotHead = false; + let ended = false; + socket.on("data", chunk => { + if (!gotHead) { + head += chunk.toString("latin1"); + const i = head.indexOf("\r\n\r\n"); + if (i >= 0) { + gotHead = true; + body = Buffer.byteLength(head.slice(i + 4), "latin1"); + } + } else { + body += chunk.length; + } + }); + socket.on("end", () => (ended = true)); + socket.on("error", () => {}); + await once(socket, "secureConnect"); + socket.end("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); + await once(socket, "close"); + return { body, ended }; + } + + it.each([ + ["client half-close, res.write() then res.end()", "write-end"], + ["client half-close, res.end(payload)", "end"], + ["client half-close, httpAllowHalfOpen, res.end() after drain", "drain"], + ] as const)("%s", async (_name, endMode) => { + await using server = https.createServer(tlsOptions, (req, res) => { + res.writeHead(200, { "Content-Length": String(BODY) }); + if (endMode === "end") { + res.end(payload); + } else { + res.write(payload); + if (endMode === "write-end") res.end(); + else res.once("drain", () => res.end()); + } + }); + if (endMode === "drain") server.httpAllowHalfOpen = true; + await once(server.listen(0, "127.0.0.1"), "listening"); + const port = (server.address() as AddressInfo).port; + for (let i = 0; i < 5; i++) { + expect(await halfCloseTlsRequestBodyBytes(port)).toEqual({ body: BODY, ended: true }); + } + }); + + // allow_half_open defers the close to the writable drain; a peer that + // FINs then resets must not wedge that drain on a spill send() that + // keeps failing (us_internal_ssl_on_writable releases a zero-progress + // spill after EOF so the dispatch reaches the close gate). A wedge + // would leave the server-side socket open past the test timeout. + it("closes promptly when the client half-closes then resets mid-drain", async () => { + const closed = Promise.withResolvers(); + await using server = https.createServer(tlsOptions, (req, res) => { + req.socket.on("close", () => closed.resolve()); + res.writeHead(200, { "Content-Length": String(BODY) }); + res.end(payload); + res.on("error", () => {}); + }); + server.requestTimeout = 0; + server.headersTimeout = 0; + await once(server.listen(0, "127.0.0.1"), "listening"); + const port = (server.address() as AddressInfo).port; + const sock = nodeTls.connect({ port, host: "127.0.0.1", rejectUnauthorized: false }); + sock.on("error", () => {}); + await once(sock, "secureConnect"); + sock.end("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); + await once(sock, "data"); + sock.destroy(); + await closed.promise; + }); + }); }); it("should handle backpressure with INT_MAX bytes", async () => {