Fixes #7001

src/bun.js/bindings/ZigGlobalObject.cpp

@@ -2220,8 +2220,9 @@ extern "C" void ReadableStream__detach(JSC__JSValue possibleReadableStream, Zig:
         return;
     auto& vm = globalObject->vm();
     auto clientData = WebCore::clientData(vm);
-    readableStream->putDirect(vm, clientData->builtinNames().bunNativePtrPrivateName(), JSC::jsUndefined(), 0);
-    readableStream->putDirect(vm, clientData->builtinNames().bunNativeTypePrivateName(), JSC::jsUndefined(), 0);
+    readableStream->putDirect(vm, clientData->builtinNames().bunNativePtrPrivateName(), jsNumber(-1), 0);
+    readableStream->putDirect(vm, clientData->builtinNames().bunNativeTypePrivateName(), jsNumber(0), 0);
+    readableStream->putDirect(vm, clientData->builtinNames().disturbedPrivateName(), jsBoolean(true), 0);
 }
 extern "C" bool ReadableStream__isDisturbed(JSC__JSValue possibleReadableStream, Zig::GlobalObject* globalObject);
 extern "C" bool ReadableStream__isDisturbed(JSC__JSValue possibleReadableStream, Zig::GlobalObject* globalObject)
@@ -2251,6 +2252,7 @@ extern "C" int32_t ReadableStreamTag__tagged(Zig::GlobalObject* globalObject, JS
     auto* readableStream = jsCast<JSReadableStream*>(object);
     auto& vm = globalObject->vm();
     auto& builtinNames = WebCore::clientData(vm)->builtinNames();
+
     int32_t num = 0;
     if (JSValue numberValue = readableStream->getDirect(vm, builtinNames.bunNativeTypePrivateName())) {
         num = numberValue.toInt32(globalObject);
@@ -3609,8 +3611,10 @@ JSC_DEFINE_HOST_FUNCTION(functionGetDirectStreamDetails, (JSC::JSGlobalObject *
         return JSC::JSValue::encode(JSC::jsNull());
     }
 
-    readableStream->putDirect(vm, clientData->builtinNames().bunNativePtrPrivateName(), jsUndefined(), 0);
-    readableStream->putDirect(vm, clientData->builtinNames().bunNativeTypePrivateName(), jsUndefined(), 0);
+    readableStream->putDirect(vm, clientData->builtinNames().bunNativePtrPrivateName(), jsNumber(0), 0);
+    // -1 === detached
+    readableStream->putDirect(vm, clientData->builtinNames().bunNativeTypePrivateName(), jsNumber(-1), 0);
+    readableStream->putDirect(vm, clientData->builtinNames().disturbedPrivateName(), jsBoolean(true), 0);
 
     auto* resultObject = JSC::constructEmptyObject(globalObject, globalObject->objectPrototype(), 2);
     resultObject->putDirect(vm, clientData->builtinNames().streamPublicName(), ptrValue, 0);

src/bun.js/webcore/body.zig

@@ -539,25 +539,31 @@ pub const Body = struct {
             value.ensureStillAlive();
 
             if (JSC.WebCore.ReadableStream.fromJS(value, globalThis)) |readable| {
+                if (readable.isDisturbed(globalThis)) {
+                    globalThis.throw("ReadableStream has already been used", .{});
+                    return null;
+                }
+
                 switch (readable.ptr) {
                     .Blob => |blob| {
+                        readable.forceDetach(globalThis);
+
                         const result: Value = .{
                             .Blob = Blob.initWithStore(blob.store, globalThis),
                         };
                         blob.store.ref();
 
-                        readable.done();
-
                         if (!blob.done) {
                             blob.done = true;
                             blob.deinit();
                         }
+
                         return result;
                     },
                     else => {},
                 }
 
-                return Body.Value.fromReadableStream(readable, globalThis);
+                return Body.Value.fromReadableStreamWithoutLockCheck(readable, globalThis);
             }
 
             return Body.Value{
@@ -573,11 +579,7 @@ pub const Body = struct {
             };
         }
 
-        pub fn fromReadableStream(readable: JSC.WebCore.ReadableStream, globalThis: *JSGlobalObject) Value {
-            if (readable.isLocked(globalThis)) {
-                return .{ .Error = ZigString.init("Cannot use a locked ReadableStream").toErrorInstance(globalThis) };
-            }
-
+        pub fn fromReadableStreamWithoutLockCheck(readable: JSC.WebCore.ReadableStream, globalThis: *JSGlobalObject) Value {
             readable.value.protect();
             return .{
                 .Locked = .{
@@ -587,6 +589,14 @@ pub const Body = struct {
             };
         }
 
+        pub fn fromReadableStream(readable: JSC.WebCore.ReadableStream, globalThis: *JSGlobalObject) Value {
+            if (readable.isLocked(globalThis)) {
+                return .{ .Error = ZigString.init("Cannot use a locked ReadableStream").toErrorInstance(globalThis) };
+            }
+
+            return fromReadableStreamWithoutLockCheck(readable, globalThis);
+        }
+
         pub fn resolve(to_resolve: *Value, new: *Value, global: *JSGlobalObject) void {
             log("resolve", .{});
             if (to_resolve.* == .Locked) {
@@ -965,9 +975,21 @@ pub fn BodyMixin(comptime Type: type) type {
 
         pub fn getBodyUsed(
             this: *Type,
-            _: *JSC.JSGlobalObject,
+            globalObject: *JSC.JSGlobalObject,
         ) callconv(.C) JSValue {
-            return JSValue.jsBoolean(this.getBodyValue().* == .Used);
+            return JSValue.jsBoolean(
+                switch (this.getBodyValue().*) {
+                    .Used => true,
+                    .Locked => |*pending| brk: {
+                        if (pending.readable) |*stream| {
+                            break :brk stream.isDisturbed(globalObject);
+                        }
+
+                        break :brk false;
+                    },
+                    else => false,
+                },
+            );
         }
 
         pub fn getJSON(

src/bun.js/webcore/response.zig

@@ -1911,7 +1911,23 @@ pub const Fetch = struct {
                     }
                 } else {
                     method = request.method;
+
+                    if (request.body.value == .Locked) {
+                        if (request.body.value.Locked.readable) |stream| {
+                            if (stream.isDisturbed(globalThis)) {
+                                globalThis.throw("ReadableStream has already been consumed", .{});
+                                if (hostname) |host| {
+                                    allocator.free(host);
+                                    hostname = null;
+                                }
+                                return .zero;
+                            }
+                        }
+                    }
+
+                    // TODO: remove second isDisturbed check in useAsAnyBlob
                     body = request.body.value.useAsAnyBlob();
+
                     if (request.headers) |head| {
                         if (head.fastGet(JSC.FetchHeaders.HTTPHeaderName.Host)) |_hostname| {
                             if (hostname) |host| {

src/bun.js/webcore/streams.zig

@@ -52,9 +52,12 @@ pub const ReadableStream = struct {
 
     pub const Strong = struct {
         held: JSC.Strong = .{},
-        globalThis: ?*JSGlobalObject = null,
 
-        pub fn init(this: ReadableStream, globalThis: *JSGlobalObject) !Strong {
+        pub fn globalThis(this: *const Strong) ?*JSGlobalObject {
+            return this.held.globalThis;
+        }
+
+        pub fn init(this: ReadableStream, global: *JSGlobalObject) !Strong {
             switch (this.ptr) {
                 .Blob => |stream| {
                     try stream.parent().incrementCount();
@@ -68,15 +71,14 @@ pub const ReadableStream = struct {
                 else => {},
             }
             return .{
-                .globalThis = globalThis,
-                .held = JSC.Strong.create(this.value, globalThis),
+                .held = JSC.Strong.create(this.value, global),
             };
         }
 
         pub fn get(this: *Strong) ?ReadableStream {
-            if (this.globalThis) |globalThis| {
+            if (this.globalThis()) |global| {
                 if (this.held.get()) |value| {
-                    return ReadableStream.fromJS(value, globalThis);
+                    return ReadableStream.fromJS(value, global);
                 }
             }
             return null;
@@ -85,8 +87,7 @@ pub const ReadableStream = struct {
         pub fn deinit(this: *Strong) void {
             if (this.get()) |readable| {
                 // decrement the ref count and if it's zero we auto detach
-                readable.detachIfPossible(this.globalThis.?);
-                this.globalThis = null;
+                readable.detachIfPossible(this.globalThis().?);
             }
             this.held.deinit();
         }
@@ -96,10 +97,25 @@ pub const ReadableStream = struct {
         return this.value;
     }
 
+    pub fn reloadTag(this: *ReadableStream, globalThis: *JSC.JSGlobalObject) void {
+        if (ReadableStream.fromJS(this.value, globalThis)) |stream| {
+            this.* = stream;
+        } else {
+            this.value.unprotect();
+            this.* = .{ .ptr = .{ .Invalid = {} }, .value = .zero };
+        }
+    }
+
     pub fn toAnyBlob(
         stream: *ReadableStream,
         globalThis: *JSC.JSGlobalObject,
     ) ?JSC.WebCore.AnyBlob {
+        if (stream.isDisturbed(globalThis)) {
+            return null;
+        }
+
+        stream.reloadTag(globalThis);
+
         switch (stream.ptr) {
             .Blob => |blobby| {
                 var blob = JSC.WebCore.Blob.initWithStore(blobby.store, globalThis);
@@ -147,14 +163,19 @@ pub const ReadableStream = struct {
 
     pub fn cancel(this: *const ReadableStream, globalThis: *JSGlobalObject) void {
         JSC.markBinding(@src());
-        this.value.unprotect();
         ReadableStream__cancel(this.value, globalThis);
+        this.value.unprotect();
     }
 
     pub fn abort(this: *const ReadableStream, globalThis: *JSGlobalObject) void {
         JSC.markBinding(@src());
-        this.value.unprotect();
         ReadableStream__cancel(this.value, globalThis);
+        this.value.unprotect();
+    }
+
+    pub fn forceDetach(this: *const ReadableStream, globalObject: *JSGlobalObject) void {
+        ReadableStream__detach(this.value, globalObject);
+        this.value.unprotect();
     }
 
     /// Decrement Source ref count and detach the underlying stream if ref count is zero
@@ -171,8 +192,8 @@ pub const ReadableStream = struct {
         };
 
         if (ref_count == 0) {
-            this.value.unprotect();
             ReadableStream__detach(this.value, globalThis);
+            this.value.unprotect();
         }
     }
 

src/js/builtins/ReadableStreamInternals.ts

@@ -1323,7 +1323,7 @@ export function readableStreamDefaultControllerCallPullIfNeeded(controller) {
 
 export function isReadableStreamLocked(stream) {
   $assert($isReadableStream(stream));
-  return !!$getByIdDirectPrivate(stream, "reader");
+  return !!$getByIdDirectPrivate(stream, "reader") || $getByIdDirectPrivate(stream, "bunNativePtr") === -1;
 }
 
 export function readableStreamDefaultControllerGetDesiredSize(controller) {
@@ -1391,7 +1391,10 @@ export function readableStreamDefaultControllerClose(controller) {
 }
 
 export function readableStreamClose(stream) {
-  $assert($getByIdDirectPrivate(stream, "state") === $streamReadable);
+  $assert(
+    $getByIdDirectPrivate(stream, "state") === $streamReadable ||
+      $getByIdDirectPrivate(stream, "state") === $streamClosing,
+  );
   $putByIdDirectPrivate(stream, "state", $streamClosed);
   if (!$getByIdDirectPrivate(stream, "reader")) return;
 
@@ -1640,6 +1643,8 @@ export function lazyLoadStream(stream, autoAllocateChunkSize) {
     $lazyStreamPrototypeMap.$set(nativeType, Prototype);
   }
 
+  $putByIdDirectPrivate(stream, "disturbed", true);
+
   const chunkSize = Prototype.startSync(nativePtr, autoAllocateChunkSize);
   var drainValue;
   const { drain: drainFn, deinit: deinitFn } = Prototype;

src/js/node/stream.js

@@ -5269,7 +5269,7 @@ function createNativeStreamReadable(nativeType, Readable) {
 
       var ptr = this.#bunNativePtr;
       $debug("ptr @ NativeReadable._read", ptr, this.__id);
-      if (ptr === 0) {
+      if (ptr === 0 || ptr === -1) {
         this.push(null);
         return;
       }

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

@@ -219,10 +219,7 @@ describe("AbortSignal", () => {
         await sleep(1);
         controller.abort();
       }
-      await Promise.all([
-        fetch(`http://127.0.0.1:${server.port}`, { signal: signal }).then(res => res.text()),
-        manualAbort(),
-      ]);
+      await Promise.all([fetch(server.url, { signal: signal }).then(res => res.text()), manualAbort()]);
     }).toThrow(new DOMException("The operation was aborted."));
   });
 
@@ -245,10 +242,7 @@ describe("AbortSignal", () => {
         await sleep(10);
         controller.abort(new Error("My Reason"));
       }
-      await Promise.all([
-        fetch(`http://127.0.0.1:${server.port}`, { signal: signal }).then(res => res.text()),
-        manualAbort(),
-      ]);
+      await Promise.all([fetch(server.url, { signal: signal }).then(res => res.text()), manualAbort()]);
     }).toThrow("My Reason");
   });
 
@@ -268,10 +262,7 @@ describe("AbortSignal", () => {
         await sleep(10);
         controller.abort();
       }
-      await Promise.all([
-        fetch(`http://127.0.0.1:${server.port}`, { signal: signal }).then(res => res.text()),
-        manualAbort(),
-      ]);
+      await Promise.all([fetch(server.url, { signal: signal }).then(res => res.text()), manualAbort()]);
     }).toThrow(new DOMException("The operation was aborted."));
   });
 
@@ -323,7 +314,7 @@ describe("AbortSignal", () => {
     }
 
     try {
-      const request = new Request(`http://127.0.0.1:${server.port}`, { signal });
+      const request = new Request(server.url, { signal });
       await Promise.all([fetch(request).then(res => res.text()), manualAbort()]);
       expect(() => {}).toThrow();
     } catch (ex: any) {
@@ -1258,7 +1249,7 @@ describe("Response", () => {
         },
       });
 
-      var response = await fetch(`http://127.0.0.1:${server.port}`, {
+      var response = await fetch(server.url, {
         method: "POST",
         body: await Bun.file(import.meta.dir + "/fixtures/file.txt").arrayBuffer(),
       });