http: don't retry a fetch with a ReadableStream body on keep-alive disconnect

src/http/lib.rs

@@ -1899,6 +1899,10 @@ impl<'a> HTTPClient<'a> {
 
         if self.allow_retry
             && self.method.is_idempotent()
+            // Only a Bytes body can be rebuilt from `original_request_body`.
+            // Stream/Sendfile bodies are consumed as they are written, so a
+            // retry would silently replay a truncated request.
+            && matches!(self.state.original_request_body, HTTPRequestBody::Bytes(_))
             && self.state.response_stage != ResponseStage::Body
             && self.state.response_stage != ResponseStage::BodyChunk
         {
@@ -2882,6 +2886,15 @@ impl<'a> HTTPClient<'a> {
 
     pub fn write_to_stream<const IS_SSL: bool>(&mut self, socket: HttpSocket<IS_SSL>, data: &[u8]) {
         bun_core::scoped_log!(fetch, "flushStream");
+        // Never write body bytes before the request headers: drain_queued_writes can
+        // reach this via the not-yet-opened socket start_() puts in the abort tracker,
+        // and request_sent_len still indexes headers. on_writable's Body arm re-flushes.
+        if !matches!(
+            self.state.request_stage,
+            RequestStage::Body | RequestStage::ProxyBody
+        ) {
+            return;
+        }
         // reshaped for borrowck — copy out the Copy bits we need
         // (`upgrade_state`, the stream-buffer NonNull, `ended`) so the
         // `&mut self.state.original_request_body` borrow is dropped before any

test/js/web/fetch/fetch-keepalive.test.ts

@@ -1,5 +1,5 @@
 import { expect, test } from "bun:test";
-import { tls } from "harness";
+import { bunEnv, bunExe, tls } from "harness";
 
 test("keepalive", async () => {
   using server = Bun.serve({
@@ -72,3 +72,100 @@ test("fetch does not reuse a pooled TLS connection for a request with a differen
   const plain = await get();
   expect(plain).not.toBe(overrideA);
 });
+
+// A reused keep-alive connection reset during a streaming PUT must reject with
+// ECONNRESET, not retry: the stream body is already consumed, and the retry
+// panicked in send_initial_request_payload. Subprocess: the panic aborts the process.
+test("PUT with a ReadableStream body is not retried on keep-alive disconnect", async () => {
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+      const CRLF = String.fromCharCode(13, 10);
+      let warmRequests = 0;
+      let streamRequests = 0;
+
+      const server = Bun.listen({
+        hostname: "127.0.0.1",
+        port: 0,
+        socket: {
+          open(socket) { socket.data = { buffer: "" }; },
+          data(socket, data) {
+            socket.data.buffer += data.toString("latin1");
+            if (!socket.data.buffer.includes(CRLF)) return;
+            if (socket.data.buffer.startsWith("PUT /warm")) {
+              // Wait for the full 4-byte body before replying keep-alive.
+              const i = socket.data.buffer.indexOf(CRLF + CRLF);
+              if (i < 0 || socket.data.buffer.length < i + 4 + 4) return;
+              warmRequests++;
+              socket.data.buffer = "";
+              socket.write("HTTP/1.1 200 OK" + CRLF + "Content-Length: 2" + CRLF + "Connection: keep-alive" + CRLF + CRLF + "ok");
+              return;
+            }
+            if (socket.data.buffer.startsWith("PUT /stream")) {
+              // Wait for the full headers plus at least one body byte so the
+              // stream body has actually started being consumed before the reset.
+              const i = socket.data.buffer.indexOf(CRLF + CRLF);
+              if (i < 0 || socket.data.buffer.length <= i + 4) return;
+              streamRequests++;
+              socket.data.buffer = "";
+              // Reset the connection mid-upload.
+              socket.terminate();
+            }
+          },
+          close() {},
+          error() {},
+          drain() {},
+        },
+      });
+
+      const base = "http://127.0.0.1:" + server.port;
+      const chunk = new Uint8Array(1024);
+      const streamBody = () => {
+        let pending = 32;
+        return new ReadableStream({
+          pull(c) {
+            if (pending-- <= 0) return c.close();
+            c.enqueue(chunk);
+          },
+        });
+      };
+
+      const errors = [];
+      for (let i = 0; i < 4; i++) {
+        // Park a keep-alive connection so the stream PUT reuses it.
+        await (await fetch(base + "/warm", { method: "PUT", body: "warm" })).text();
+        try {
+          await fetch(base + "/stream", { method: "PUT", body: streamBody(), duplex: "half" });
+          errors.push(null);
+        } catch (e) {
+          errors.push(e && (e.code || e.name));
+        }
+      }
+
+      server.stop();
+      console.log(JSON.stringify({ warmRequests, streamRequests, errors }));
+      process.exit(0);
+      `,
+    ],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+  // If the subprocess crashed there is no JSON; surface the raw output instead.
+  const result = stdout.startsWith("{") ? JSON.parse(stdout.trim()) : { stdout, stderr };
+  expect({ result, exitCode }).toEqual({
+    // Without the fix every attempt is retried on a fresh connection, so the
+    // server sees each PUT /stream twice (streamRequests === 8).
+    result: {
+      warmRequests: 4,
+      streamRequests: 4,
+      errors: ["ECONNRESET", "ECONNRESET", "ECONNRESET", "ECONNRESET"],
+    },
+    exitCode: 0,
+  });
+});