diff --git a/packages/bun-uws/src/HttpContext.h b/packages/bun-uws/src/HttpContext.h index a6e243d947e9..67d209e5e54f 100644 --- a/packages/bun-uws/src/HttpContext.h +++ b/packages/bun-uws/src/HttpContext.h @@ -203,18 +203,19 @@ struct HttpContext { * so a client that connects and never sends anything still expires. */ if constexpr (IsNodeHttp) { ((HttpResponseData *) us_socket_ext(s))->lastMessageStartMs = nodeCompatMonotonicMs(); - /* 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 (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; } + /* A peer FIN must not tear the connection down at the loop level: + * onEnd() below decides whether to close right away (idle, or a + * response the application is still producing) or to keep writing + * response bytes already handed to uWS. Without this flag the loop + * (and openssl.c us_internal_ssl_on_end for TLS) force-closes the + * socket right after dispatching onEnd, discarding those 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) { /* Call filter */ for (auto &f : httpContextData->filterHandlers) { @@ -665,14 +666,12 @@ struct HttpContext { size_t flushed = asyncSocket->flush(); /* Check if there's still data waiting to be sent after flush attempt */ if (asyncSocket->getBufferedAmount() > 0) { - if constexpr (IsNodeHttp) { - /* onEnd deferred close for these bytes; a writable event that - * moves nothing (EPIPE) means the peer is gone and this would - * otherwise spin onWritable/onEnd until idle timeout. */ - if (flushed == 0 - && (httpResponseData->state & HttpResponseData::HTTP_NODE_RECEIVED_FIN)) { - return asyncSocket->close(); - } + /* onEnd deferred close for these bytes; a writable event that + * moves nothing (EPIPE) means the peer is gone and this would + * otherwise spin the writable dispatch until idle timeout. */ + if (flushed == 0 + && (httpResponseData->state & HttpResponseData::HTTP_NODE_RECEIVED_FIN)) { + return asyncSocket->close(); } /* Socket buffer is not completely empty yet * - Reset the timeout to prevent premature connection closure @@ -698,10 +697,24 @@ struct HttpContext { /* We are now writable, so hang timeout again, the user does not have to do anything so we should hang until end or tryEnd rearms timeout */ us_socket_timeout(s, 0); + [[maybe_unused]] uint64_t offsetBefore = httpResponseData->offset; + /* We expect the developer to return whether or not write was successful (true). * If write was never called, the developer should still return true so that we may drain. */ bool success = httpResponseData->callOnWritable(reinterpret_cast *>(asyncSocket), httpResponseData->offset); + if constexpr (!IsNodeHttp) { + /* Bun.serve: onEnd deferred close for a tryEnd tail (offset < total, + * nothing in AsyncSocketData::buffer). A retry that moves zero bytes + * after the peer's FIN is EPIPE; close instead of spinning. */ + if ((httpResponseData->state & HttpResponseData::HTTP_NODE_RECEIVED_FIN) + && (httpResponseData->state & HttpResponseData::HTTP_RESPONSE_PENDING) + && httpResponseData->offset == offsetBefore + && asyncSocket->hasFullyDrained()) { + return asyncSocket->close(); + } + } + /* The developer indicated that their onWritable failed. */ if (!success) { /* Skip testing if we can drain anything since that might perform an extra syscall */ @@ -807,6 +820,26 @@ struct HttpContext { httpResponseData->state |= HttpResponseData::HTTP_NODE_RECEIVED_FIN; return s; } + } else { + /* Bun.serve: response bytes already handed to uWS must drain before + * the connection shuts down (from the existing shouldCloseConnection() + * gates), not be discarded by the close() below. Only a response that + * is fully determined qualifies: a tryEnd tail (content-length path + * sets HTTP_END_CALLED while offset < total keeps HTTP_RESPONSE_PENDING) + * or a completed response that has not fully drained. A streaming body + * the application is still producing (HTTP_END_CALLED clear, + * HTTP_RESPONSE_PENDING set) closes here so onAborted / request.signal + * fires on client disconnect. */ + HttpResponseData *httpResponseData = (HttpResponseData *) us_socket_ext(s); + uint32_t state = httpResponseData->state; + bool tryEndTail = (state & HttpResponseData::HTTP_END_CALLED) + && (state & HttpResponseData::HTTP_RESPONSE_PENDING); + bool doneButBuffered = !(state & HttpResponseData::HTTP_RESPONSE_PENDING) + && !asyncSocket->hasFullyDrained(); + if (tryEndTail || doneButBuffered) { + httpResponseData->state |= HttpResponseData::HTTP_NODE_RECEIVED_FIN; + return s; + } } asyncSocket->uncorkWithoutSending(); diff --git a/test/js/bun/http/serve.test.ts b/test/js/bun/http/serve.test.ts index 6aa45fb3ac49..d6534b96eae8 100644 --- a/test/js/bun/http/serve.test.ts +++ b/test/js/bun/http/serve.test.ts @@ -22,6 +22,7 @@ import { heapStats } from "bun:jsc"; import { spawn } from "child_process"; import net from "node:net"; import { networkInterfaces } from "node:os"; +import nodeTls from "node:tls"; import { tmpdir } from "os"; let renderToReadableStream: any = null; @@ -3513,6 +3514,142 @@ it("survives aborted uploads while responding with a tee()d request-body branch" }); }); +// A client that half-closes its write side right after the request (the raw +// socket.end(request) pattern) must receive every response byte already handed +// to uWS, not just what the kernel accepted on the first send. No +// Connection: close on the request: the post-drain shutdown is driven by the +// HTTP_NODE_RECEIVED_FIN clause of shouldCloseConnection(), not by +// HTTP_CONNECTION_CLOSE. +describe("a client half-close after the request does not truncate a large response body", () => { + const BODY = 8 * 1024 * 1024; + + function countBody(socket: net.Socket | nodeTls.TLSSocket) { + const out = { body: 0, ended: false }; + let head = ""; + let gotHead = false; + socket.on("data", chunk => { + if (!gotHead) { + head += chunk.toString("latin1"); + const i = head.indexOf("\r\n\r\n"); + if (i >= 0) { + gotHead = true; + out.body = Buffer.byteLength(head.slice(i + 4), "latin1"); + } + } else { + out.body += chunk.length; + } + }); + socket.on("end", () => (out.ended = true)); + socket.on("error", () => {}); + return out; + } + + async function halfCloseRequest(port: number): Promise<{ body: number; ended: boolean }> { + const socket = connect(port, "127.0.0.1"); + const out = countBody(socket); + const closed = new Promise(r => socket.once("close", () => r())); + await new Promise(r => socket.once("connect", () => r())); + socket.end("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); + await closed; + return out; + } + + it("fetch handler (tryEnd tail)", async () => { + using server = serve({ + port: 0, + fetch: () => new Response(Buffer.alloc(BODY, "a"), { headers: { "content-length": String(BODY) } }), + }); + expect(await halfCloseRequest(server.port)).toEqual({ body: BODY, ended: true }); + }); + + it("static route (tryEnd tail)", async () => { + using server = serve({ + port: 0, + routes: { + "/": new Response(Buffer.alloc(BODY, "a"), { headers: { "content-length": String(BODY) } }), + }, + fetch: () => new Response("miss", { status: 404 }), + }); + expect(await halfCloseRequest(server.port)).toEqual({ body: BODY, ended: true }); + }); + + it("https fetch handler (tryEnd tail)", async () => { + using server = serve({ + port: 0, + tls, + fetch: () => new Response(Buffer.alloc(BODY, "a"), { headers: { "content-length": String(BODY) } }), + }); + const socket = nodeTls.connect({ port: server.port, host: "127.0.0.1", rejectUnauthorized: false }); + const out = countBody(socket); + const closed = new Promise(r => socket.once("close", () => r())); + await new Promise(r => socket.once("secureConnect", () => r())); + socket.end("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); + await closed; + expect(out).toEqual({ body: BODY, ended: true }); + }); + + // The deferred connection must close promptly, not spin the writable + // dispatch, when the peer goes away mid-drain (tryEnd retry hits EPIPE): + // half-close to enter the defer, then destroy() on first data. idleTimeout + // is high so a spin would miss the poll deadline rather than be masked by + // an idle-timeout close. On platforms whose loopback send buffer swallows + // the whole body (Windows) there is no tryEnd tail and the response + // completes before first data; pendingRequests is the portable signal + // that the connection has closed one way or the other. + it("closes without spinning when the peer goes away mid-drain", async () => { + const dispatched = Promise.withResolvers(); + using server = serve({ + port: 0, + idleTimeout: 60, + fetch() { + dispatched.resolve(); + return new Response(Buffer.alloc(BODY, "a"), { headers: { "content-length": String(BODY) } }); + }, + }); + const socket = connect(server.port, "127.0.0.1"); + socket.on("error", () => {}); + await new Promise(r => socket.once("connect", () => r())); + socket.end("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); + socket.once("data", () => socket.destroy()); + await dispatched.promise; + const deadline = Date.now() + 4000; + while (server.pendingRequests > 0 && Date.now() < deadline) await Bun.sleep(5); + expect(server.pendingRequests).toBe(0); + }); + + // The defer in onEnd is gated on the response being fully determined + // (HTTP_END_CALLED). A streaming body the handler is still producing must + // close on client FIN so onAborted / request.signal fires. + it("request.signal still fires on client FIN for a streaming body", async () => { + const aborted = Promise.withResolvers(); + using server = serve({ + port: 0, + fetch(req) { + req.signal.addEventListener("abort", () => aborted.resolve()); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: hi\n\n")); + }, + cancel() {}, + }), + { headers: { "content-type": "text/event-stream" } }, + ); + }, + }); + const socket = connect(server.port, "127.0.0.1"); + const gotData = Promise.withResolvers(); + socket.once("data", () => gotData.resolve()); + socket.on("error", () => {}); + await new Promise(r => socket.once("connect", () => r())); + socket.write("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); + await gotData.promise; + socket.end(); + await aborted.promise; + socket.destroy(); + }); +}); + // The node:http compat parser tolerates empty lines (and a bare CR/LF) before the // request-line like llhttp's s_start state. That leniency must stay behind the // node-http flag: Bun.serve still rejects a request that does not begin with the