Don't write Connection: close into an in-flight chunked response body

packages/bun-uws/src/HttpResponse.h

@@ -170,8 +170,11 @@ struct HttpResponse : public AsyncSocket<SSL> {
                 /* Write mark, this propagates to WebSockets too */
                 writeMark();
 
-                /* WebSocket upgrades does not allow content-length */
-                if (allowContentLength) {
+                /* WebSocket upgrades does not allow content-length.
+                 * Once write() has been called the header section is already terminated and body
+                 * bytes are on the wire (e.g. a close-delimited HTTP/1.0 streaming response), so
+                 * writing a header here would corrupt the response body. */
+                if (allowContentLength && !(httpResponseData->state & (HttpResponseData<SSL>::HTTP_WRITE_CALLED))) {
                     /* Even zero is a valid content-length */
                     Super::write("Content-Length: ", 16);
                     writeUnsigned64(totalSize);

src/uws_sys/libuwsockets.cpp

@@ -1340,13 +1340,15 @@ extern "C"
       auto *data = uwsRes->getHttpResponseData();
       if (close_connection)
       {
-        if (!(data->state & uWS::HttpResponseData<true>::HTTP_CONNECTION_CLOSE))
+        /* Once write() has been called, the header section is already terminated and body
+         * bytes are on the wire; injecting a header here would corrupt the response body. */
+        if (!(data->state & uWS::HttpResponseData<true>::HTTP_CONNECTION_CLOSE) && !(data->state & uWS::HttpResponseData<true>::HTTP_WRITE_CALLED))
         {
           uwsRes->writeHeader("Connection", "close");
         }
         data->state |= uWS::HttpResponseData<true>::HTTP_CONNECTION_CLOSE;
       }
-      if (!(data->state & uWS::HttpResponseData<true>::HTTP_END_CALLED))
+      if (!(data->state & (uWS::HttpResponseData<true>::HTTP_END_CALLED | uWS::HttpResponseData<true>::HTTP_WRITE_CALLED)))
       {
         uwsRes->AsyncSocket<true>::write("\r\n", 2);
       }
@@ -1360,13 +1362,15 @@ extern "C"
       auto *data = uwsRes->getHttpResponseData();
       if (close_connection)
       {
-        if (!(data->state & uWS::HttpResponseData<false>::HTTP_CONNECTION_CLOSE))
+        /* Once write() has been called, the header section is already terminated and body
+         * bytes are on the wire; injecting a header here would corrupt the response body. */
+        if (!(data->state & uWS::HttpResponseData<false>::HTTP_CONNECTION_CLOSE) && !(data->state & uWS::HttpResponseData<false>::HTTP_WRITE_CALLED))
         {
           uwsRes->writeHeader("Connection", "close");
         }
         data->state |= uWS::HttpResponseData<false>::HTTP_CONNECTION_CLOSE;
       }
-      if (!(data->state & uWS::HttpResponseData<false>::HTTP_END_CALLED))
+      if (!(data->state & (uWS::HttpResponseData<false>::HTTP_END_CALLED | uWS::HttpResponseData<false>::HTTP_WRITE_CALLED)))
       {
         // Some HTTP clients require the complete "<header>\r\n\r\n" to be sent.
         // If not, they may throw a ConnectionError.

test/js/bun/http/serve-direct-readable-stream.test.ts

@@ -334,3 +334,80 @@ test("sync pull() under AsyncLocalStorage releases the request on end()", async
   const counts = heapStats().objectTypeCounts;
   expect((counts.ReadableStream ?? 0) - baseline).toBeLessThan(10);
 });
+
+// https://github.com/oven-sh/bun/issues/28019
+// A close-delimited HTTP/1.0 streaming response has no framing, so nothing but
+// body bytes may be written once the body starts. Ending the sink while it
+// still held buffered data used to route through uWS::internalEnd's
+// content-length branch and inject "Content-Length: <n>\r\n\r\n" into the body.
+test("ending an HTTP/1.0 streaming response does not inject a Content-Length header", async () => {
+  const first = Buffer.alloc(65536, "x");
+  const expectedBody = first.toString() + "Hello Bun!\n";
+  const firstBytesReceived = Promise.withResolvers<void>();
+  await using server = Bun.serve({
+    port: 0,
+    async fetch() {
+      return new Response(
+        new ReadableStream({
+          type: "direct",
+          async pull(ctrl) {
+            // At or above the sink's highWaterMark: flushed to the socket
+            // immediately, so the response body is started on the wire.
+            ctrl.write(first);
+            // Wait until the client holds body bytes before finishing.
+            await firstBytesReceived.promise;
+            // Below the highWaterMark: stays in the sink's buffer, so ending
+            // the sink ends the response with buffered data left over.
+            ctrl.write("Hello Bun!\n");
+            ctrl.end();
+          },
+        } as any),
+      );
+    },
+  });
+
+  const { promise, resolve, reject } = Promise.withResolvers<string>();
+  let received = "";
+  let headerEnd = -1;
+  await Bun.connect({
+    hostname: server.hostname,
+    port: server.port!,
+    socket: {
+      open(socket) {
+        // HTTP/1.0 without keep-alive: the response is delimited by the
+        // connection close, so the server streams it without chunked framing.
+        socket.write(`GET / HTTP/1.0\r\nHost: ${server.hostname}\r\n\r\n`);
+      },
+      data(socket, data) {
+        received += data.toString("latin1");
+        if (headerEnd === -1) {
+          headerEnd = received.indexOf("\r\n\r\n");
+        }
+        if (headerEnd !== -1) {
+          const body = received.slice(headerEnd + 4);
+          if (body.length > 0) {
+            firstBytesReceived.resolve();
+          }
+          // A corrupted body has extra injected bytes, so it reaches the
+          // expected length too; compare as soon as the length is there.
+          if (body.length >= expectedBody.length) {
+            resolve(body);
+            socket.end();
+          }
+        }
+      },
+      close() {
+        reject(new Error(`connection closed after ${received.length} bytes, before the full body arrived`));
+      },
+      error(_socket, error) {
+        reject(error);
+      },
+    },
+  });
+
+  const body = await promise;
+  // The first 64 KiB were already on the wire when the stream ended; anything
+  // injected by the end path lands right after them.
+  expect(body.slice(65536)).toBe(expectedBody.slice(65536));
+  expect(body).toBe(expectedBody);
+});

test/js/node/http/node-http-transfer-encoding.test.ts

@@ -55,6 +55,50 @@ test(`should not duplicate transfer-encoding header in request`, async () => {
   return promise;
 });
 
+// res.destroy() while a chunked body is in flight must not write anything else to the
+// socket. uws_res_end_without_body used to inject "Connection: close\r\n\r\n" into the
+// stream, which clients then parsed as chunk framing and delivered as body data
+// (seen as a flaky extra 'data' event in test-http-server-capture-rejections.js).
+test("destroying a response mid-chunked-body does not write header bytes into the stream", async () => {
+  await using server = createServer((req, res) => {
+    res.setHeader("Content-Type", "application/json");
+    res.write("{");
+    server.once("destroy-response", () => res.destroy());
+  });
+
+  await once(server.listen(0, "127.0.0.1"), "listening");
+
+  const { port } = server.address() as AddressInfo;
+
+  const { promise, resolve, reject } = Promise.withResolvers<string>();
+  const socket = connect(port, "127.0.0.1", () => {
+    socket.write("GET / HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: keep-alive\r\n\r\n");
+  });
+
+  let received = "";
+  let baseline = -1;
+  socket.on("data", (chunk: Buffer) => {
+    received += chunk.toString("latin1");
+    if (baseline === -1 && received.includes("\r\n\r\n1\r\n{\r\n")) {
+      // Headers and the complete first chunk arrived; everything received
+      // from here on is written by the destroy path.
+      baseline = received.length;
+      server.emit("destroy-response");
+    }
+  });
+  socket.on("error", () => {}); // an abrupt close is fine; 'close' always follows
+  socket.on("close", () => {
+    if (baseline === -1) {
+      reject(new Error(`connection closed before the first chunk arrived: ${JSON.stringify(received)}`));
+    } else {
+      resolve(received.slice(baseline));
+    }
+  });
+
+  // The connection just dies; no bytes may follow the first chunk.
+  expect(await promise).toBe("");
+});
+
 test("should not duplicate transfer-encoding header in response when explicitly set", async () => {
   await using server = createServer((req, res) => {
     res.writeHead(200, { "Transfer-Encoding": "chunked" });