diff --git a/src/http/lib.rs b/src/http/lib.rs index 98abce8af610..55da89d80f82 100644 --- a/src/http/lib.rs +++ b/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(&mut self, socket: HttpSocket, 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 diff --git a/test/js/web/fetch/fetch-keepalive.test.ts b/test/js/web/fetch/fetch-keepalive.test.ts index 84aa951dd936..6a92e93b0529 100644 --- a/test/js/web/fetch/fetch-keepalive.test.ts +++ b/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, + }); +});