From f90dc48866901435f71f8b268f1a5a0739229899 Mon Sep 17 00:00:00 2001 From: robobun Date: Sat, 27 Jun 2026 00:00:48 +0000 Subject: [PATCH 1/4] http: don't retry a fetch with a ReadableStream body on keep-alive disconnect An idempotent request (PUT) with a ReadableStream body sent over a reused keep-alive connection that the peer resets mid-upload was transparently retried by HTTPClient::on_close. A stream body is consumed as it is uploaded, so the replay is silently truncated, and the chunks still queued by the JS side get flushed onto the retry's not-yet-opened socket ahead of the request headers. That advances request_sent_len past the length of the rebuilt header buffer and send_initial_request_payload panics, aborting the whole process: panic: range start index 1031 out of range for slice of length 172 Two changes: - on_close: only take the idempotent retry when the body is HTTPRequestBody::Bytes. Stream and Sendfile bodies are consumed as they are written and cannot be replayed (Go's net/http and curl apply the same rule). The request now fails with ECONNRESET like any other non-retryable request. - write_to_stream: park until request_stage is Body or ProxyBody. start_() registers the still-connecting socket in the abort tracker (needed for abort-during-connect), which let drain_queued_writes write body bytes to it before the headers went out. The buffered data is re-flushed by on_writable's Body arm once the headers are on the wire. --- src/http/lib.rs | 14 +++ test/js/web/fetch/fetch-keepalive.test.ts | 100 +++++++++++++++++++++- 2 files changed, 113 insertions(+), 1 deletion(-) diff --git a/src/http/lib.rs b/src/http/lib.rs index 98abce8af610..681eb1b52c1e 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,16 @@ 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_() registers in the + // abort tracker, and request_sent_len still indexes the header buffer. + // The data stays buffered; on_writable's Body/ProxyBody 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..539c078e5e4e 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,101 @@ test("fetch does not reuse a pooled TLS connection for a request with a differen const plain = await get(); expect(plain).not.toBe(overrideA); }); + +// An idempotent request (PUT) on a reused keep-alive connection that the peer +// resets is transparently retried, but a ReadableStream body is consumed as it +// is uploaded and cannot be replayed: the already written chunks are gone, so a +// retry silently sends a truncated body, and flushing the still-draining stream +// onto the retry's not-yet-opened socket put body bytes ahead of the headers +// and panicked with "range start index N out of range for slice of length M" +// in send_initial_request_payload. Runs in a subprocess: the panic aborts the +// whole 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")) { + 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, + }); +}); From d36fad997fd22c94b773052df5a96e5e5f125092 Mon Sep 17 00:00:00 2001 From: robobun Date: Sat, 27 Jun 2026 00:12:37 +0000 Subject: [PATCH 2/4] test: reset only after the stream body starts uploading, trim the comment Review feedback: the server now waits for the full request headers plus at least one body byte before resetting the connection, so the test provably exercises a ReadableStream body that has started being consumed. Also shortens the test's header comment to three lines. --- test/js/web/fetch/fetch-keepalive.test.ts | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/test/js/web/fetch/fetch-keepalive.test.ts b/test/js/web/fetch/fetch-keepalive.test.ts index 539c078e5e4e..6a92e93b0529 100644 --- a/test/js/web/fetch/fetch-keepalive.test.ts +++ b/test/js/web/fetch/fetch-keepalive.test.ts @@ -73,14 +73,9 @@ test("fetch does not reuse a pooled TLS connection for a request with a differen expect(plain).not.toBe(overrideA); }); -// An idempotent request (PUT) on a reused keep-alive connection that the peer -// resets is transparently retried, but a ReadableStream body is consumed as it -// is uploaded and cannot be replayed: the already written chunks are gone, so a -// retry silently sends a truncated body, and flushing the still-draining stream -// onto the retry's not-yet-opened socket put body bytes ahead of the headers -// and panicked with "range start index N out of range for slice of length M" -// in send_initial_request_payload. Runs in a subprocess: the panic aborts the -// whole process. +// 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: [ @@ -109,6 +104,10 @@ test("PUT with a ReadableStream body is not retried on keep-alive disconnect", a 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. From c4d28c020cf9ba134a3495969de90889ea7705b0 Mon Sep 17 00:00:00 2001 From: robobun Date: Sat, 27 Jun 2026 00:32:14 +0000 Subject: [PATCH 3/4] ci: retrigger darwin-aarch64 test-bun failed on a buildkite artifact download timeout before running any test; no other job in build 65218 failed. From 18651111efe14de3fecf813ddcef2ec944458f66 Mon Sep 17 00:00:00 2001 From: robobun Date: Sat, 27 Jun 2026 00:44:30 +0000 Subject: [PATCH 4/4] http: trim the write_to_stream stage-guard comment to three lines --- src/http/lib.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/http/lib.rs b/src/http/lib.rs index 681eb1b52c1e..55da89d80f82 100644 --- a/src/http/lib.rs +++ b/src/http/lib.rs @@ -2886,10 +2886,9 @@ 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_() registers in the - // abort tracker, and request_sent_len still indexes the header buffer. - // The data stays buffered; on_writable's Body/ProxyBody arm re-flushes. + // 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