diff --git a/src/jsc/bindings/webcore/streams/JSReadableStream.h b/src/jsc/bindings/webcore/streams/JSReadableStream.h index 1cf50a5166fc..188d9acbedba 100644 --- a/src/jsc/bindings/webcore/streams/JSReadableStream.h +++ b/src/jsc/bindings/webcore/streams/JSReadableStream.h @@ -113,6 +113,9 @@ class JSReadableStream final : public JSC::JSNonFinalObject { // reads the raw slot, so the fetch/server push-side is unaffected. if (m_nativeTextMode) return {}; + // The native source of an errored stream only knows that it ended; the stored error is here. + if (m_state == ReadableStreamState::Errored) + return {}; return m_nativePtr.get(); // may be empty } // Transferred to a node:stream Readable, drained as a Body, or detached: the handle is off limits. diff --git a/src/jsc/bindings/webcore/streams/WebStreamsExports.cpp b/src/jsc/bindings/webcore/streams/WebStreamsExports.cpp index f8e9904ad3b1..425e35a43e31 100644 --- a/src/jsc/bindings/webcore/streams/WebStreamsExports.cpp +++ b/src/jsc/bindings/webcore/streams/WebStreamsExports.cpp @@ -48,6 +48,9 @@ using namespace Bun::WebStreams; // A JSReadableStream's tag and native source. Pure: no scope, no traps, no script. static int32_t tagOfStream(JSReadableStream* stream, void** ptr) { + // An errored stream reads as a JS stream: its readers get the stored error, which the native source does not have. + if (stream->m_state == ReadableStreamState::Errored) + return 0; // The RAW handle slot, not nativePtrForJS(): a transferred stream still tags. JSValue handle = stream->m_nativePtr.get(); if (handle.isEmpty() || !handle.isCell()) diff --git a/src/runtime/webcore/ByteStream.rs b/src/runtime/webcore/ByteStream.rs index 74ec2182fd7e..77a2603581b2 100644 --- a/src/runtime/webcore/ByteStream.rs +++ b/src/runtime/webcore/ByteStream.rs @@ -786,6 +786,20 @@ impl ByteStream { streams::Result::Pending(self.pending.as_ptr()) } + /// The JS stream was errored with `reason`. A native reader that waits now fails with it. + pub(crate) fn error_native_consumer(&self, reason: JSValue) { + let waiting = self.sink.get().is_some() + || self.buffer_action.get().is_some() + || self.pending.get().state == streams::PendingState::Pending; + self.on_data(streams::Result::Err(if waiting { + let global = self.parent_const().global_this(); + streams::StreamError::JSValue(StrongOptional::create(reason, global)) + } else { + // Kept for a reader that already holds this source and pulls later. A stored `reason` would be a GC root. + streams::StreamError::AbortReason(jsc::CommonAbortReason::UserAbort) + })); + } + pub(crate) fn on_cancel(&self) { bun_jsc::mark_binding!(); let view = self.value(); diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index 261e94b6bac7..c22703418fec 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -594,8 +594,32 @@ impl FileReader { } } + /// The JS stream was errored with `reason`. A native reader that waits now fails with it. + pub(crate) fn error_native_consumer(&self, reason: jsc::JSValue) { + // SAFETY: see `parent()`. + let _pin = unsafe { SourcePin::new(self.parent()) }; + let global = self.parent_const().global_this(); + let err = streams::StreamError::JSValue(jsc::strong::Optional::create(reason, global)); + let sink = *self.sink.get(); + if sink.is_some() { + self.detach_sink(Some(&err)); + sink.end(Some(err)); + } else if self.pending.get().state == streams::PendingState::Pending { + self.pending + .with_mut(|p| p.result = streams::Result::Err(err)); + self.pending.with_mut(|p| p.run()); + } + } + pub(crate) fn on_cancel(&self) { + // A sink still wired here must fail, not see an EOF and commit a truncated body. + let sink = *self.sink.get(); self.unpipe_without_deref(); + if sink.is_some() { + sink.end(Some(streams::StreamError::AbortReason( + jsc::CommonAbortReason::UserAbort, + ))); + } if self.done.get() { return; } diff --git a/src/runtime/webcore/ReadableStream.rs b/src/runtime/webcore/ReadableStream.rs index 18b0f908019a..e75931ee0739 100644 --- a/src/runtime/webcore/ReadableStream.rs +++ b/src/runtime/webcore/ReadableStream.rs @@ -290,6 +290,11 @@ impl ReadableStream { /// Like [`Self::cancel`] but pending reads reject with `reason` instead of resolving `{done: true}`. pub(crate) fn error(&self, global_this: &JSGlobalObject, reason: JSValue) -> JsResult<()> { let result = bun_jsc::cpp::ReadableStream__error(self.value, global_this, reason); + if let Some(bytes) = self.ptr.bytes() { + bytes.error_native_consumer(reason); + } else if let Some(file) = self.ptr.file() { + file.error_native_consumer(reason); + } self.done(); result } diff --git a/test/js/web/fetch/fetch-abort-stream-body.test.ts b/test/js/web/fetch/fetch-abort-stream-body.test.ts index 9c057fd018d4..115bf30f37bb 100644 --- a/test/js/web/fetch/fetch-abort-stream-body.test.ts +++ b/test/js/web/fetch/fetch-abort-stream-body.test.ts @@ -1,8 +1,9 @@ -import { expect, test } from "bun:test"; -import { bunEnv, bunExe, isASAN, isMacOS } from "harness"; +import { describe, expect, test } from "bun:test"; +import { bunEnv, bunExe, isASAN, isMacOS, tempDir } from "harness"; import { once } from "node:events"; import net from "node:net"; import { join } from "node:path"; +import { Readable } from "node:stream"; // Aborting a fetch whose request body stream is still uploading must also // settle the response side. The failure callback used to return right after @@ -210,6 +211,303 @@ test.concurrent("abort() errors a fully-buffered fetch response body", async () } }); +// Fetch spec "abort a fetch" step 4 errors the body with the signal's abort reason, so every +// reader of the body rejects with `signal.reason` itself. The readers that take `res.body` +// natively (`new Response(res.body).text()` and `Bun.readableStreamTo*()` buffer it without a +// reader, `Bun.write()`, a fetch() upload, HTMLRewriter and a Bun.serve response wire it to their +// sink, `Readable.fromWeb()` pulls from it) got a fresh "AbortError: The operation was aborted." +// or a clean end when they had already started, and an empty body that ended cleanly when they +// started after the abort. +describe("aborting mid-body fails every reader of res.body with signal.reason", () => { + // A head, 40 of 100 body bytes, then nothing: the body is still arriving when the signal fires. + async function stalledBodyServer() { + const sockets = new Set(); + const server = net.createServer(socket => { + sockets.add(socket); + socket.on("error", () => {}); + socket.once("data", () => socket.write("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n" + Buffer.alloc(40, "x"))); + }); + await once(server.listen(0, "127.0.0.1"), "listening"); + return { + url: `http://127.0.0.1:${(server.address() as net.AddressInfo).port}/`, + [Symbol.dispose]() { + for (const socket of sockets) socket.destroy(); + server.close(); + }, + }; + } + + // Each returns once the headers arrived; `abort()` then fires the signal (the timer does it alone). + const signals: Record Promise<{ res: Response; signal: AbortSignal; abort(): void }>> = { + "AbortSignal.timeout()": async url => { + // A timer that beats the headers rejects fetch() itself: try again with a longer one. + for (let ms = 100; ; ms *= 2) { + const signal = AbortSignal.timeout(ms); + try { + return { res: await fetch(url, { signal }), signal, abort() {} }; + } catch (error) { + if (error !== signal.reason) throw error; + } + } + }, + "abort(new Error())": async url => { + const controller = new AbortController(); + const res = await fetch(url, { signal: controller.signal }); + return { res, signal: controller.signal, abort: () => controller.abort(new Error("custom")) }; + }, + "abort()": async url => { + const controller = new AbortController(); + const res = await fetch(url, { signal: controller.signal }); + return { res, signal: controller.signal, abort: () => controller.abort() }; + }, + }; + + // Each starts to read before it returns its promise. + const consumers: Record Promise> = { + "res.text()": res => res.text(), + "res.body reader": async res => { + const reader = res.body!.getReader(); + while (!(await reader.read()).done); + }, + "res.body.pipeTo()": res => res.body!.pipeTo(new WritableStream({})), + "new Response(res.body).text()": res => new Response(res.body).text(), + "new Response(res.body).json()": res => new Response(res.body).json(), + "new Response(res.body).bytes()": res => new Response(res.body).bytes(), + "new Response(res.body).blob()": res => new Response(res.body).blob(), + "new Response(res.body).arrayBuffer()": res => new Response(res.body).arrayBuffer(), + "new Request(url, { body: res.body }).text()": res => + new Request("http://localhost/", { method: "POST", body: res.body }).text(), + "Bun.readableStreamToText(res.body)": res => Bun.readableStreamToText(res.body!), + "Bun.readableStreamToBytes(res.body)": async res => await Bun.readableStreamToBytes(res.body!), + "res.body.text()": res => res.body!.text(), + "Bun.write(path, new Response(res.body))": async res => { + using dir = tempDir("fetch-abort-reason", {}); + await Bun.write(join(String(dir), "body"), new Response(res.body)); + }, + "new HTMLRewriter().transform(new Response(res.body)).text()": res => + new HTMLRewriter().transform(new Response(res.body)).text(), + }; + + const settle = (promise: Promise) => + promise.then( + () => "resolved", + error => error, + ); + + describe.each(Object.keys(signals))("%s after the reader started", kind => { + test.concurrent.each(Object.keys(consumers))("%s", async name => { + using server = await stalledBodyServer(); + const { res, signal, abort } = await signals[kind](server.url); + const settled = settle(consumers[name](res)); + abort(); + expect(await settled).toBe(signal.reason); + }); + }); + + describe("abort(new Error()) before the reader starts", () => { + test.concurrent.each(Object.keys(consumers))("%s", async name => { + using server = await stalledBodyServer(); + const { res, signal, abort } = await signals["abort(new Error())"](server.url); + // The stream exists when the signal fires. Nothing reads it yet. + expect(res.body).toBeInstanceOf(ReadableStream); + abort(); + expect(await settle(consumers[name](res))).toBe(signal.reason); + }); + }); + + // A reason does not have to be an Error. + test.concurrent.each([[null], [42], ["str"], [Symbol("reason")]])( + "abort(%p) reaches a native reader unchanged", + async reason => { + for (const timing of ["after the reader started", "before the reader starts"]) { + using server = await stalledBodyServer(); + const controller = new AbortController(); + const res = await fetch(server.url, { signal: controller.signal }); + expect(res.body).toBeInstanceOf(ReadableStream); + if (timing === "before the reader starts") controller.abort(reason); + const settled = settle(new Response(res.body).text()); + if (timing === "after the reader started") controller.abort(reason); + expect(await settled).toBe(reason); + } + }, + ); + + test.concurrent.each(["after the upload started", "before the upload starts"])( + "fetch(url, { body: res.body }), abort %s", + async timing => { + using server = await stalledBodyServer(); + const uploading = Promise.withResolvers(); + await using target = Bun.serve({ + port: 0, + async fetch(req) { + const reader = req.body!.getReader(); + // The first body bytes: the upload has wired `res.body` to its sink. + await reader.read(); + uploading.resolve(); + try { + while (!(await reader.read()).done); + } catch {} + return new Response("ok"); + }, + }); + + const controller = new AbortController(); + const reason = new Error("custom"); + const res = await fetch(server.url, { signal: controller.signal }); + const body = res.body; + if (timing === "before the upload starts") controller.abort(reason); + const upload = fetch(target.url, { method: "POST", body }); + if (timing === "after the upload started") { + await uploading.promise; + controller.abort(reason); + } + await expect(upload).rejects.toBe(reason); + }, + ); + + // Readable.fromWeb() takes the native source away from the stream and pulls from it on demand. + test.concurrent.each(["while it waits for more", "before Readable.fromWeb()"])( + "Readable.fromWeb(res.body), abort %s", + async timing => { + using server = await stalledBodyServer(); + const controller = new AbortController(); + const reason = new Error("custom"); + const res = await fetch(server.url, { signal: controller.signal }); + expect(res.body).toBeInstanceOf(ReadableStream); + if (timing === "before Readable.fromWeb()") controller.abort(reason); + const readable = Readable.fromWeb(res.body as any); + const settled = new Promise(resolve => readable.on("error", resolve).on("end", () => resolve("ended"))); + if (timing === "while it waits for more") { + // The 40 bytes arrived, and one turn later the next pull is parked. + await once(readable, "data"); + await new Promise(resolve => setImmediate(resolve)); + controller.abort(reason); + } else { + readable.resume(); + } + expect(await settled).toBe(reason); + }, + ); + + // Nothing waits, so nothing can take the reason, and to keep it natively would root it. The + // source still has to fail: main ended such a stream as if the body were complete. + test.concurrent("Readable.fromWeb(res.body) that has not read yet fails with an AbortError", async () => { + using server = await stalledBodyServer(); + const controller = new AbortController(); + const res = await fetch(server.url, { signal: controller.signal }); + const readable = Readable.fromWeb(res.body as any); + const settled = new Promise(resolve => readable.on("error", resolve).on("end", () => resolve("ended"))); + controller.abort(new Error("custom")); + readable.resume(); + expect((await settled)?.name).toBe("AbortError"); + }); + + // The reason is not kept natively: a Strong there would root reason -> Response -> stream. + test.concurrent("an abort reason that references its Response is collected after the body is read", async () => { + const script = ` + const { heapStats } = require("bun:jsc"); + const net = require("node:net"); + const sockets = new Set(); + const upstream = net.createServer(socket => { + sockets.add(socket); + socket.on("error", () => {}); + socket.on("close", () => sockets.delete(socket)); + socket.once("data", () => + socket.write("HTTP/1.1 200 OK\\r\\nContent-Length: 100\\r\\n\\r\\n" + Buffer.alloc(40, "x")), + ); + }); + await new Promise(resolve => upstream.listen(0, "127.0.0.1", resolve)); + const url = "http://127.0.0.1:" + upstream.address().port; + async function once() { + const controller = new AbortController(); + const res = await fetch(url, { signal: controller.signal }); + void res.body; + controller.abort(new Error("custom", { cause: res })); + await res.text().catch(() => {}); + } + const count = async () => { + for (let i = 0; i < 3; i++) { + Bun.gc(true); + await new Promise(resolve => setImmediate(resolve)); + } + return heapStats().objectTypeCounts.Response || 0; + }; + for (let i = 0; i < 4; i++) await once(); + const baseline = await count(); + for (let i = 0; i < 24; i++) await once(); + const after = await count(); + for (const socket of sockets) socket.destroy(); + console.log(JSON.stringify({ leaked: after > baseline + 8 })); + process.exit(0); + `; + await using proc = Bun.spawn({ cmd: [bunExe(), "-e", script], env: bunEnv, stdout: "pipe", stderr: "pipe" }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).toBe(""); + expect(stdout).toBe('{"leaked":false}\n'); + expect(exitCode).toBe(0); + }); + + // The server reports the failure itself (stderr) and cuts the connection, as for any stream + // that errors. Before this it answered an aborted body with a complete, empty 200. + test.concurrent.each([ + ["before the response starts", '{"rejected":true}'], + ["after the response started", '{"status":200,"first":40}'], + ])("Bun.serve response of new Response(res.body), abort %s", async (timing, seen) => { + const script = ` + import net from "node:net"; + const upstream = net.createServer(socket => { + socket.on("error", () => {}); + socket.once("data", () => + socket.write("HTTP/1.1 200 OK\\r\\nContent-Length: 100\\r\\n\\r\\n" + Buffer.alloc(40, "x")), + ); + }); + await new Promise(resolve => upstream.listen(0, "127.0.0.1", resolve)); + let abort; + const proxy = Bun.serve({ + port: 0, + async fetch() { + const controller = new AbortController(); + abort = () => controller.abort(new Error("custom reason")); + const res = await fetch("http://127.0.0.1:" + upstream.address().port, { signal: controller.signal }); + const body = res.body; + if (process.env.ABORT_TIMING === "before the response starts") abort(); + return new Response(body); + }, + }); + let seen; + try { + const res = await fetch(proxy.url); + const reader = res.body.getReader(); + let first = 0; + while (first < 40) { + const { value, done } = await reader.read(); + if (done) break; + first += value.length; + } + abort(); + await reader.read().catch(() => {}); + seen = { status: res.status, first }; + } catch { + seen = { rejected: true }; + } + console.log(JSON.stringify(seen)); + process.exit(0); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: { ...bunEnv, ABORT_TIMING: timing }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + + expect(stderr).toContain("error: custom reason"); + expect(stdout).toBe(seen + "\n"); + expect(exitCode).toBe(0); + }); +}); + test.concurrent("abort reaches an in-flight fetch whose signal nothing else references, after GC", async () => { const response = Promise.withResolvers(); await using server = Bun.serve({ port: 0, fetch: () => response.promise });