Close a body's stream when a consumer takes its payload without reading it

src/jsc/bindings/webcore/streams/WebStreamsExports.cpp

@@ -199,6 +199,20 @@ extern "C" void ReadableStream__markConsumedAsBody(JSC::EncodedJSValue possibleR
     stream->m_consumedAsBody = true;
 }
 
+// markConsumedAsBody for Rust `to_any_blob`, which took the payload of a stream nothing started: no reader exists to close it.
+extern "C" void ReadableStream__closeConsumedAsBody(JSC::EncodedJSValue possibleReadableStream, Zig::GlobalObject* globalObject)
+{
+    auto* stream = dynamicDowncast<JSReadableStream>(JSValue::decode(possibleReadableStream));
+    if (!stream) [[unlikely]]
+        return;
+    stream->m_disturbed = true;
+    stream->m_consumedAsBody = true;
+    ASSERT(!stream->m_reader);
+    auto scope = DECLARE_TOP_EXCEPTION_SCOPE(JSC::getVM(globalObject));
+    readableStreamCloseIfPossible(globalObject, stream);
+    scope.assertNoExceptionExceptTermination();
+}
+
 // A native sink (fetch body / S3 / FileSink) has attached directly without a reader.
 // Mark the stream disturbed+locked so .locked, .getReader(), and the body-mixin
 // disturbed checks behave as they do after readStreamIntoSink acquires a reader.
@@ -222,12 +236,18 @@ extern "C" JSC::EncodedJSValue ReadableStream__empty(Zig::GlobalObject* globalOb
     return JSValue::encode(stream);
 }
 
-extern "C" JSC::EncodedJSValue ReadableStream__used(Zig::GlobalObject* globalObject)
+// A locked stand-in for a body's stream. `consumed`: the body was read to its end (closed, disturbed), not still being read.
+extern "C" JSC::EncodedJSValue ReadableStream__used(Zig::GlobalObject* globalObject, bool consumed)
 {
     auto& vm = JSC::getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
     auto* stream = createReadableStream(globalObject, SourceKind::Nothing, nullptr, jsUndefined());
     RETURN_IF_EXCEPTION(scope, {});
+    if (consumed) {
+        stream->m_disturbed = true;
+        readableStreamClose(globalObject, stream);
+        RETURN_IF_EXCEPTION(scope, {});
+    }
     acquireReadableStreamDefaultReader(globalObject, stream);
     RETURN_IF_EXCEPTION(scope, {});
     return JSValue::encode(stream);

src/jsc/bindings/webcore/streams/WebStreamsInternals.h

@@ -721,8 +721,9 @@ void ReadableStream__cancel(JSC::EncodedJSValue possibleReadableStream, Zig::Glo
 void ReadableStream__cancelWithReason(JSC::EncodedJSValue possibleReadableStream, Zig::GlobalObject*, JSC::EncodedJSValue reason); // userJS: yes
 bool ReadableStream__isClosedUnread(JSC::EncodedJSValue possibleReadableStream, Zig::GlobalObject*); // userJS: no
 void ReadableStream__markConsumedAsBody(JSC::EncodedJSValue possibleReadableStream, Zig::GlobalObject*); // userJS: no
+void ReadableStream__closeConsumedAsBody(JSC::EncodedJSValue possibleReadableStream, Zig::GlobalObject*); // userJS: no
 JSC::EncodedJSValue ReadableStream__empty(Zig::GlobalObject*); // userJS: no
-JSC::EncodedJSValue ReadableStream__used(Zig::GlobalObject*); // userJS: no
+JSC::EncodedJSValue ReadableStream__used(Zig::GlobalObject*, bool consumed); // userJS: no
 JSC::EncodedJSValue ReadableStream__errored(Zig::GlobalObject*, JSC::EncodedJSValue reason); // userJS: no
 JSC::EncodedJSValue ReadableStream__fromDecodedText(Zig::GlobalObject*, JSC::EncodedJSValue string); // userJS: no
 JSC::EncodedJSValue ReadableStream__textDecodeFrom(Zig::GlobalObject*, JSC::EncodedJSValue source); // userJS: yes

src/runtime/webcore/Body.rs

@@ -890,7 +890,7 @@ impl Value {
         // producer's remaining callbacks with that foreign context.
         if locked.promise.is_some() || !locked.action.is_none() || locked.on_receive_value.is_some()
         {
-            return ReadableStream::used(global_this);
+            return ReadableStream::in_use(global_this);
         }
         let mut drain_result = DrainResult::EstimatedSize(0);
 

src/runtime/webcore/ReadableStream.rs

@@ -142,11 +142,12 @@ unsafe extern "C" {
         global_object: &JSGlobalObject,
     ) -> bool;
     safe fn ReadableStream__empty(global: &JSGlobalObject) -> JSValue;
-    safe fn ReadableStream__used(global: &JSGlobalObject) -> JSValue;
+    safe fn ReadableStream__used(global: &JSGlobalObject, consumed: bool) -> JSValue;
     safe fn ReadableStream__errored(global: &JSGlobalObject, reason: JSValue) -> JSValue;
     safe fn ReadableStream__fromDecodedText(global: &JSGlobalObject, string: JSValue) -> JSValue;
     safe fn ReadableStream__textDecodeFrom(global: &JSGlobalObject, source: JSValue) -> JSValue;
     safe fn ReadableStream__markConsumedAsBody(stream: JSValue, global: &JSGlobalObject);
+    safe fn ReadableStream__closeConsumedAsBody(stream: JSValue, global: &JSGlobalObject);
     safe fn ReadableStream__lockNative(stream: JSValue, global: &JSGlobalObject);
     /// BunStreamSource.cpp: queue the adapter's close on the microtask queue.
     safe fn Bun__NativeStreamSourceAdapter__onClose(global: &JSGlobalObject, adapter: JSValue);
@@ -191,7 +192,7 @@ impl ReadableStream {
         });
     }
 
-    /// Lift the whole payload out of an unread stream. On success the stream is spent (disturbed, locked).
+    /// Lift the whole payload out of an unread stream. On success the stream is spent (closed, disturbed, locked).
     pub fn to_any_blob(&mut self, global_this: &JSGlobalObject) -> Option<webcore::blob::Any> {
         if self.is_disturbed(global_this) || self.is_locked(global_this) {
             return None;
@@ -239,7 +240,7 @@ impl ReadableStream {
         };
 
         self.done();
-        self.mark_consumed_as_body(global_this);
+        ReadableStream__closeConsumedAsBody(self.value, global_this);
         Some(blob)
     }
 
@@ -642,13 +643,19 @@ impl ReadableStream {
         })
     }
 
+    /// A locked stand-in for the stream of a body that was read to its end: closed and disturbed.
     pub fn used(global_this: &JSGlobalObject) -> JsResult<JSValue> {
         bun_jsc::from_js_host_call(global_this, || {
             // SAFETY: FFI call into JSC bindings; global_this is a valid &JSGlobalObject.
-            ReadableStream__used(global_this)
+            ReadableStream__used(global_this, true)
         })
     }
 
+    /// A locked stand-in for the stream of a body a consumer is still reading: it stays readable.
+    pub fn in_use(global_this: &JSGlobalObject) -> JsResult<JSValue> {
+        bun_jsc::from_js_host_call(global_this, || ReadableStream__used(global_this, false))
+    }
+
     /// A stream already in the `errored` state, so every read rejects with
     /// `reason` instead of closing cleanly.
     pub fn errored(global_this: &JSGlobalObject, reason: JSValue) -> JsResult<JSValue> {

test/js/web/fetch/body.test.ts

@@ -2,7 +2,9 @@ import { file, spawn, version, type Socket } from "bun";
 import { describe, expect, test } from "bun:test";
 import { bunEnv, bunExe, exampleSite, tempDir } from "harness";
 import net from "net";
-import { Readable } from "node:stream";
+import { join } from "node:path";
+import { isDisturbed, isErrored, isReadable, Readable } from "node:stream";
+import { finished } from "node:stream/promises";
 
 const exampleServer = exampleSite("http");
 
@@ -1575,6 +1577,13 @@ describe("body stream bookkeeping does not depend on the body's source", () => {
     }
   };
   const settled = (p: Promise<unknown>) => p.then(String, e => (e as Error).constructor.name);
+  const streamState = (stream: ReadableStream) => ({
+    readable: isReadable(stream),
+    errored: isErrored(stream),
+    disturbed: isDisturbed(stream),
+    locked: stream.locked,
+  });
+  const consumedState = { readable: false, errored: false, disturbed: true, locked: true };
 
   for (const [ownerName, make] of owners) {
     describe(ownerName, () => {
@@ -1672,6 +1681,41 @@ describe("body stream bookkeeping does not depend on the body's source", () => {
         });
       });
 
+      // That reader ran the stream to its end, so the stream is closed. node:stream's
+      // helpers read the stream's own state, and finished() waits for it to close.
+      describe("a consumed body's stream is closed", () => {
+        for (const [name, init] of sources) {
+          const methods = ["text", "json", "arrayBuffer", "bytes", "blob"] as const;
+          for (const method of name === "URLSearchParams" ? ([...methods, "formData"] as const) : methods) {
+            test(`${name} .body then ${method}()`, async () => {
+              const owner = make(init());
+              const stream = owner.body!;
+              await owner[method]().catch(() => {});
+              expect(streamState(stream)).toEqual(consumedState);
+              await finished(stream);
+            });
+          }
+        }
+        for (const method of ["text", "json", "arrayBuffer", "bytes", "blob"] as const) {
+          test(`${method}() then .body`, async () => {
+            const owner = make("[1]");
+            await owner[method]();
+            const stream = owner.body!;
+            expect(streamState(stream)).toEqual(consumedState);
+            await finished(stream);
+          });
+        }
+        test("clone() takes an unread native stream's content", async () => {
+          const owner = make("payload");
+          const stream = owner.body!;
+          const clone = owner.clone();
+          expect(owner.body).not.toBe(stream);
+          expect(streamState(stream)).toEqual(consumedState);
+          await finished(stream);
+          expect([await owner.text(), await clone.text()]).toEqual(["payload", "payload"]);
+        });
+      });
+
       // A zero-length body is a body: reading it, through a mixin method or
       // through its stream, uses it up.
       describe("a zero-length body is used up like any other", () => {
@@ -1838,4 +1882,94 @@ describe("body stream bookkeeping does not depend on the body's source", () => {
       expect(response.endsWith(`\r\n\r\n${expected}`)).toBe(true);
     }
   });
+
+  // Lifting the payload back out of an unread native stream is not a read through
+  // a reader, so the stream has nothing left to close it: whoever lifts it does.
+  describe("lifting the payload out of an unread native stream closes it", () => {
+    const expectClosed = async (stream: ReadableStream) => {
+      expect(streamState(stream)).toEqual(consumedState);
+      await finished(stream);
+    };
+
+    test("Blob.stream() behind a new Response", async () => {
+      const stream = new Blob(["payload"]).stream();
+      // Registered first, so the lift has a pending promise to settle.
+      const closed = finished(stream);
+      expect(await new Response(stream).bytes()).toEqual(new TextEncoder().encode("payload"));
+      expect(streamState(stream)).toEqual(consumedState);
+      await closed;
+    });
+
+    test("a Bun.file() stream", async () => {
+      using dir = tempDir("body-stream-closed", { "payload.txt": "payload" });
+      const path = join(String(dir), "payload.txt");
+
+      const stream = file(path).stream();
+      expect((await new Response(stream).arrayBuffer()).byteLength).toBe(7);
+      await expectClosed(stream);
+
+      const response = new Response(file(path));
+      const body = response.body!;
+      expect((await response.blob()).size).toBe(7);
+      await expectClosed(body);
+    });
+
+    test("Bun.serve sending a Response whose .body was accessed", async () => {
+      let body!: ReadableStream;
+      await using server = Bun.serve({
+        port: 0,
+        fetch() {
+          const response = new Response("payload");
+          body = response.body!;
+          return response;
+        },
+      });
+      expect(await (await fetch(server.url)).text()).toBe("payload");
+      await expectClosed(body);
+    });
+
+    test("fetch() uploading it and Bun.write() writing it", async () => {
+      using dir = tempDir("body-stream-closed", {});
+      await using server = Bun.serve({ port: 0, fetch: async req => new Response(await req.text()) });
+
+      const uploaded = new Blob(["payload"]).stream();
+      expect(await (await fetch(server.url, { method: "POST", body: uploaded })).text()).toBe("payload");
+      await expectClosed(uploaded);
+
+      const response = new Response("payload");
+      const written = response.body!;
+      expect(await Bun.write(join(String(dir), "out.txt"), response)).toBe(7);
+      await expectClosed(written);
+    });
+  });
+
+  // While a consumer still reads a body that never had a stream, .body is a locked
+  // stand-in that nothing updates. It must not report the end of a read that can still fail.
+  test(".body of a body that is still being read is not closed", async () => {
+    const { promise: released, resolve: release } = Promise.withResolvers<void>();
+    await using server = Bun.serve({
+      port: 0,
+      fetch: () =>
+        new Response(
+          new ReadableStream({
+            async pull(controller) {
+              controller.enqueue(new TextEncoder().encode("pay"));
+              await released;
+              controller.enqueue(new TextEncoder().encode("load"));
+              controller.close();
+            },
+          }),
+        ),
+    });
+    const response = await fetch(server.url);
+    const text = response.text();
+    const body = response.body!;
+    expect({ readable: isReadable(body), locked: body.locked, bodyUsed: response.bodyUsed }).toEqual({
+      readable: true,
+      locked: true,
+      bodyUsed: true,
+    });
+    release();
+    expect(await text).toBe("payload");
+  });
 });