Skip to content
Closed
19 changes: 12 additions & 7 deletions src/runtime/webcore/fetch/FetchTasklet.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2528,13 +2528,18 @@ impl FetchTasklet {
}

if let BodyValue::Locked(locked) = body {
if let Some(promise) = locked.promise {
if promise.is_empty_or_undefined_or_null() {
// Scenario 2b.
this.ignore_remaining_response_body(true);
}
} else {
// Scenario 3.
let has_live_promise = locked
.promise
.is_some_and(|p| !p.is_empty_or_undefined_or_null());
if !has_live_promise {
// Scenario 2a / 3: the body was never consumed. Close the
// connection (undici destroys on GC) instead of resuming
// the paused transport and draining the rest of the body
// just to pool the socket. Same abort-then-ignore shape as
// on_stream_cancelled_callback; abort_task() only touches
// atomics and the HTTP-thread shutdown queue, so it is safe
// from this Weak finalizer.
Comment thread
robobun marked this conversation as resolved.
Resolved
this.abort_task();
this.ignore_remaining_response_body(true);
}
}
Expand Down
100 changes: 100 additions & 0 deletions test/js/web/fetch/fetch-backpressure.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -379,3 +379,103 @@
}
});
});

// Dropping a Response without reading or cancelling its body used to resume the
// paused transport in the Response's Weak finalizer and drain the whole body to
// return the socket to the keep-alive pool. That costs the full transfer for
// the "just read the headers" pattern. The finalizer now aborts the request so
// the server sees a close and stops at the kernel send window, matching undici.
describe.each(["content-length", "chunked"] as const)("fetch() abandoned Response body (%s)", encoding => {
test.concurrent("GC-collecting the Response closes the connection instead of draining the body", async () => {
await using proc = Bun.spawn({
cmd: [
bunExe(),
"-e",
/* js */ `
const net = require("node:net");
const CHUNK = Buffer.alloc(65536, 0x61);
const TOTAL = 64 * 1024 * 1024;
const chunked = ${JSON.stringify(encoding === "chunked")};
let written = 0;

Check failure on line 399 in test/js/web/fetch/fetch-backpressure.test.ts

View check run for this annotation

Claude / Claude Code Review

GC-drop test can flake on CI hosts with large kernel TCP buffers

The new GC-drop test uses `TOTAL = 64 * 1024 * 1024`, but the "server stops writing while the reader is stalled" test ~150 lines earlier in this same file documents that some CI hosts have `tcp_rmem[2]+tcp_wmem[2]` approaching 256 MiB and uses 1 GiB for exactly that reason — on those hosts `push()` writes the full 64 MiB into kernel buffers and calls `settle('drained')` before the finalizer's abort closes the socket, so the test fails with `outcome === 'drained'` even though the fix works. Separ
Comment thread
robobun marked this conversation as resolved.
Resolved
let conns = 0;
const { promise: settled, resolve: settle } = Promise.withResolvers();
const server = net.createServer(socket => {
conns++;
socket.on("error", () => {});
socket.on("close", () => settle("closed"));
socket.once("data", () => {
const head = chunked
? "HTTP/1.1 200 OK\\r\\nTransfer-Encoding: chunked\\r\\n\\r\\n"
: "HTTP/1.1 200 OK\\r\\nContent-Length: " + TOTAL + "\\r\\n\\r\\n";
socket.write(head);
let sent = 0;
const frame = chunked
? Buffer.concat([Buffer.from(CHUNK.length.toString(16) + "\\r\\n"), CHUNK, Buffer.from("\\r\\n")])
: CHUNK;
const push = () => {
while (sent < TOTAL && !socket.destroyed) {
sent += CHUNK.length;
written += CHUNK.length;
if (!socket.write(frame)) return void socket.once("drain", push);
}
if (chunked && !socket.destroyed) socket.write("0\\r\\n\\r\\n");
settle("drained");
};
push();
});
});
await new Promise(r => server.listen(0, "127.0.0.1", r));
const base = "http://127.0.0.1:" + server.address().port;

await (async () => {
const res = await fetch(base + "/big");
if (res.status !== 200) throw new Error("status " + res.status);
})();

// Collect the unreachable Response. Weak finalizers run in
// WeakBlock::sweep, which needs an allocation between GCs to sweep the
// block the dead Response sits in.
for (let i = 0; i < 5; i++) {
new Error("alloc");
Bun.gc(true);
await Bun.sleep(0);
}
// Either the server sees the socket close (fix) or it finishes pushing
// the whole body into a still-open socket (the old drain-to-pool path).
const outcome = await settled;

let second = "skipped";
if (outcome === "closed") {
// The first socket was closed, not pooled; a follow-up request must
// open a fresh connection and still complete.
const r2 = await fetch(base + "/big");
await r2.body.cancel();
second = r2.status === 200 ? "ok" : String(r2.status);
}

process.stdout.write(JSON.stringify({ outcome, written, total: TOTAL, conns, second }));
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 (!stdout) throw new Error(`client exited ${exitCode}: ${stderr}`);
const { outcome, written, total, conns, second } = JSON.parse(stdout);
// Before the fix the finalizer resumed the transport and the server wrote
// the full 64 MiB before pooling the socket (outcome "drained", conns 1).
// After, the server sees the close at the kernel send+recv window, well
// under half of 64 MiB on any lane, and the second fetch opens a fresh
// connection.
expect({ outcome, drained: written >= total, conns, second }).toEqual({
outcome: "closed",
drained: false,
conns: 2,
second: "ok",
});
expect(written).toBeLessThan(total / 2);
expect(exitCode).toBe(0);
}, 30_000);
});
Loading