Fix ReadableStream reference management for Request/Response bodies

src/bun.js/api/bun/spawn/stdio.zig

@@ -277,10 +277,10 @@ pub const Stdio = union(enum) {
         };
     }
 
-    fn extractBodyValue(out_stdio: *Stdio, globalThis: *jsc.JSGlobalObject, i: i32, body: *jsc.WebCore.Body.Value, is_sync: bool) bun.JSError!void {
-        body.toBlobIfPossible();
+    fn extractBodyValue(out_stdio: *Stdio, globalThis: *jsc.JSGlobalObject, i: i32, owner: jsc.WebCore.ReadableStream.Ref.Owner, body: *jsc.WebCore.Body.Value, is_sync: bool) bun.JSError!void {
+        body.toBlobIfPossible(owner);
 
-        if (body.tryUseAsAnyBlob()) |blob| {
+        if (body.tryUseAsAnyBlob(owner)) |blob| {
             return out_stdio.extractBlob(globalThis, blob, i);
         }
 
@@ -313,7 +313,7 @@ pub const Stdio = union(enum) {
                     else => unreachable,
                 }
 
-                const stream_value = try body.toReadableStream(globalThis);
+                const stream_value = try body.toReadableStream(owner, globalThis);
 
                 const stream = (try jsc.WebCore.ReadableStream.fromJS(stream_value, globalThis)) orelse return globalThis.throwInvalidArguments("Failed to create ReadableStream", .{});
 
@@ -391,9 +391,9 @@ pub const Stdio = union(enum) {
         } else if (value.as(jsc.WebCore.Blob)) |blob| {
             return out_stdio.extractBlob(globalThis, .{ .Blob = blob.dupe() }, i);
         } else if (value.as(jsc.WebCore.Request)) |req| {
-            return extractBodyValue(out_stdio, globalThis, i, req.getBodyValue(), is_sync);
+            return extractBodyValue(out_stdio, globalThis, i, .{ .Request = value }, req.getBodyValue(), is_sync);
         } else if (value.as(jsc.WebCore.Response)) |res| {
-            return extractBodyValue(out_stdio, globalThis, i, res.getBodyValue(), is_sync);
+            return extractBodyValue(out_stdio, globalThis, i, .{ .Response = value }, res.getBodyValue(), is_sync);
         }
 
         if (try jsc.WebCore.ReadableStream.fromJS(value, globalThis)) |stream_| {

src/bun.js/api/html_rewriter.zig

@@ -197,7 +197,7 @@ pub const HTMLRewriter = struct {
 
         if (kind != .other) {
             {
-                const body_value = try jsc.WebCore.Body.extract(global, response_value);
+                const body_value = try jsc.WebCore.Body.extract(global, response_value, null);
                 const resp = bun.new(Response, Response{
                     .init = .{
                         .status_code = 200,

src/bun.js/api/server.zig

@@ -1198,6 +1198,8 @@ pub fn NewServer(protocol_enum: enum { http, https }, development_kind: enum { d
                     bun.default_allocator,
                     ctx,
                     false,
+                    first_arg,
+                    null,
                 );
             } else {
                 const fetch_error = jsc.WebCore.Fetch.fetch_type_error_strings.get(bun.jsc.C.JSValueGetType(ctx, first_arg.asRef()));

src/bun.js/api/server/FileRoute.zig

@@ -59,7 +59,8 @@ pub fn memoryCost(this: *const FileRoute) usize {
 
 pub fn fromJS(globalThis: *jsc.JSGlobalObject, argument: jsc.JSValue) bun.JSError!?*FileRoute {
     if (argument.as(jsc.WebCore.Response)) |response| {
-        response.body.value.toBlobIfPossible();
+        const owner = if (response.this_jsvalue.tryGet()) |jsval| jsc.WebCore.ReadableStream.Ref.Owner{ .Response = jsval } else jsc.WebCore.ReadableStream.Ref.Owner{ .empty = {} };
+        response.body.value.toBlobIfPossible(owner);
         if (response.body.value == .Blob and response.body.value.Blob.needsToReadFile()) {
             if (response.body.value.Blob.store.?.data.file.pathlike == .fd) {
                 return globalThis.throwTODO("Support serving files from a file descriptor. Please pass a path instead.");

src/bun.js/api/server/RequestContext.zig

@@ -627,6 +627,8 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
                 abort.cb(abort.data);
             }
 
+            const response_jsvalue = this.response_jsvalue;
+
             this.detachResponse();
             var any_js_calls = false;
             var vm = this.server.?.vm;
@@ -676,13 +678,22 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
                     any_js_calls = true;
                 }
 
+                // Abort request body stream if still locked (e.g., pending read)
+                if (this.request_weakref.get()) |request| {
+                    if (request.body.value == .Locked) {
+                        const owner: jsc.WebCore.ReadableStream.Ref.Owner = if (request.this_jsvalue.tryGet()) |js_value|
+                            .{ .Request = js_value }
+                        else
+                            .empty;
+                        if (request.body.value.Locked.readable.abort(owner, globalThis)) {
+                            any_js_calls = true;
+                        }
+                    }
+                }
+
                 if (this.response_ptr) |response| {
                     if (response.body.value == .Locked) {
-                        var strong_readable = response.body.value.Locked.readable;
-                        response.body.value.Locked.readable = .{};
-                        defer strong_readable.deinit();
-                        if (strong_readable.get(globalThis)) |readable| {
-                            readable.abort(globalThis);
+                        if (response.body.value.Locked.readable.abort(.{ .Response = response_jsvalue }, globalThis)) {
                             any_js_calls = true;
                         }
                     }
@@ -1216,7 +1227,7 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
                             // TODO: should this timeout?
                             this.response_ptr.?.body.value = .{
                                 .Locked = .{
-                                    .readable = jsc.WebCore.ReadableStream.Strong.init(stream, globalThis),
+                                    .readable = .{ .strong = .init(stream, globalThis) },
                                     .global = globalThis,
                                 },
                             };
@@ -1454,7 +1465,8 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
                 }
             }
             // not content-length or transfer-encoding so we need to respect the body
-            response.body.value.toBlobIfPossible();
+            const owner = if (response.this_jsvalue.tryGet()) |jsval| jsc.WebCore.ReadableStream.Ref.Owner{ .Response = jsval } else jsc.WebCore.ReadableStream.Ref.Owner{ .empty = {} };
+            response.body.value.toBlobIfPossible(owner);
             switch (response.body.value) {
                 .InternalBlob, .WTFStringImpl => {
                     var blob = response.body.value.useAsAnyBlobAllowNonUTF8String();
@@ -1559,7 +1571,8 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
                     }
                     return;
                 } else {
-                    response.body.value.toBlobIfPossible();
+                    const owner = if (response.this_jsvalue.tryGet()) |jsval| jsc.WebCore.ReadableStream.Ref.Owner{ .Response = jsval } else jsc.WebCore.ReadableStream.Ref.Owner{ .empty = {} };
+                    response.body.value.toBlobIfPossible(owner);
 
                     switch (response.body.value) {
                         .Blob => |*blob| {
@@ -1618,7 +1631,8 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
                             }
                             return;
                         }
-                        response.body.value.toBlobIfPossible();
+                        const owner = if (response.this_jsvalue.tryGet()) |jsval| jsc.WebCore.ReadableStream.Ref.Owner{ .Response = jsval } else jsc.WebCore.ReadableStream.Ref.Owner{ .empty = {} };
+                        response.body.value.toBlobIfPossible(owner);
                         switch (response.body.value) {
                             .Blob => |*blob| {
                                 if (blob.needsToReadFile()) {
@@ -1662,10 +1676,11 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
 
                 if (resp.body.value == .Locked) {
                     const global = resp.body.value.Locked.global;
-                    if (resp.body.value.Locked.readable.get(global)) |stream| {
-                        stream.done(global);
+                    if (resp.this_jsvalue.tryGet()) |value| {
+                        resp.body.value.Locked.readable.done(.{ .Response = value }, global);
+                    } else {
+                        resp.body.value.Locked.readable.deinit();
                     }
-                    resp.body.value.Locked.readable.deinit();
                     resp.body.value = .{ .Used = {} };
                 }
             }
@@ -1715,10 +1730,11 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
 
             if (req.response_ptr) |resp| {
                 if (resp.body.value == .Locked) {
-                    if (resp.body.value.Locked.readable.get(globalThis)) |stream| {
-                        stream.done(globalThis);
+                    if (resp.this_jsvalue.tryGet()) |value| {
+                        resp.body.value.Locked.readable.done(.{ .Response = value }, globalThis);
+                    } else {
+                        resp.body.value.Locked.readable.deinit();
                     }
-                    resp.body.value.Locked.readable.deinit();
                     resp.body.value = .{ .Used = {} };
                 }
             }
@@ -1801,7 +1817,7 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
 
             // If a ReadableStream can trivially be converted to a Blob, do so.
             // If it's a WTFStringImpl and it cannot be used as a UTF-8 string, convert it to a Blob.
-            value.toBlobIfPossible();
+            value.toBlobIfPossible(.{ .empty = {} });
             const globalThis = this.server.?.globalThis;
             switch (value.*) {
                 .Error => |*err_ref| {
@@ -1827,19 +1843,21 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
                         return;
                     }
 
-                    if (lock.readable.get(globalThis)) |stream_| {
-                        const stream: jsc.WebCore.ReadableStream = stream_;
-                        // we hold the stream alive until we're done with it
-                        this.readable_stream_ref = lock.readable;
-                        value.* = .{ .Used = {} };
+                    if (lock.readable.get(.{ .Response = this.response_jsvalue }, globalThis)) |stream| {
+                        {
+                            var old = this.readable_stream_ref;
+                            this.readable_stream_ref = .init(stream, globalThis);
+                            old.deinit();
+                            value.* = .{ .Used = {} };
+                        }
 
                         if (stream.isLocked(globalThis)) {
                             streamLog("was locked but it shouldn't be", .{});
                             var err = jsc.SystemError{
                                 .code = bun.String.static(@tagName(jsc.Node.ErrorCode.ERR_STREAM_CANNOT_PIPE)),
                                 .message = bun.String.static("Stream already used, please create a new one"),
                             };
-                            stream.value.unprotect();
+                            this.readable_stream_ref.deinit();
                             this.runErrorHandler(err.toErrorInstance(globalThis));
                             return;
                         }
@@ -1883,7 +1901,10 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
                                 }
                                 this.ref();
                                 byte_stream.pipe = jsc.WebCore.Pipe.Wrap(@This(), onPipe).init(this);
-                                this.readable_stream_ref = jsc.WebCore.ReadableStream.Strong.init(stream, globalThis);
+                                if (this.readable_stream_ref.has()) {
+                                    this.readable_stream_ref.deinit();
+                                }
+                                this.readable_stream_ref = .init(stream, globalThis);
 
                                 this.byte_stream = byte_stream;
                                 var response_buf = byte_stream.drain();
@@ -1907,7 +1928,7 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
 
                     if (lock.onReceiveValue != null or lock.task != null) {
                         // someone else is waiting for the stream or waiting for `onStartStreaming`
-                        const readable = value.toReadableStream(globalThis) catch return; // TODO: properly propagate exception upwards
+                        const readable = value.toReadableStream(.{ .Response = this.response_jsvalue }, globalThis) catch return;
                         readable.ensureStillAlive();
                         this.doRenderWithBody(value);
                         return;
@@ -2162,7 +2183,8 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
                     ctx.flags.response_protected = false;
                     ctx.response_ptr = response;
 
-                    response.body.value.toBlobIfPossible();
+                    const owner = if (response.this_jsvalue.tryGet()) |jsval| jsc.WebCore.ReadableStream.Ref.Owner{ .Response = jsval } else jsc.WebCore.ReadableStream.Ref.Owner{ .empty = {} };
+                    response.body.value.toBlobIfPossible(owner);
                     switch (response.body.value) {
                         .Blob => |*blob| {
                             if (blob.needsToReadFile()) {
@@ -2492,7 +2514,7 @@ pub fn NewRequestContext(comptime ssl_enabled: bool, comptime debug_mode: bool,
         pub fn onRequestBodyReadableStreamAvailable(ptr: *anyopaque, globalThis: *jsc.JSGlobalObject, readable: jsc.WebCore.ReadableStream) void {
             var this = bun.cast(*RequestContext, ptr);
             bun.debugAssert(this.request_body_readable_stream_ref.held.impl == null);
-            this.request_body_readable_stream_ref = jsc.WebCore.ReadableStream.Strong.init(readable, globalThis);
+            this.request_body_readable_stream_ref = .init(readable, globalThis);
         }
 
         pub fn onStartBufferingCallback(this: *anyopaque) void {

src/bun.js/api/server/StaticRoute.zig

@@ -91,7 +91,8 @@ pub fn fromJS(globalThis: *jsc.JSGlobalObject, argument: jsc.JSValue) bun.JSErro
 
         // The user may want to pass in the same Response object multiple endpoints
         // Let's let them do that.
-        response.body.value.toBlobIfPossible();
+        const owner = if (response.this_jsvalue.tryGet()) |jsval| jsc.WebCore.ReadableStream.Ref.Owner{ .Response = jsval } else jsc.WebCore.ReadableStream.Ref.Owner{ .empty = {} };
+        response.body.value.toBlobIfPossible(owner);
 
         const blob: AnyBlob = brk: {
             switch (response.body.value) {

src/bun.js/bindings/JSBunRequest.cpp

@@ -96,14 +96,15 @@ JSObject* JSBunRequest::cookies() const
     return m_cookies.get();
 }
 
-extern "C" void* Request__clone(void* internalZigRequestPointer, JSGlobalObject* globalObject);
+extern "C" void* Request__clone(void* internalZigRequestPointer, JSGlobalObject* globalObject, JSC::EncodedJSValue thisValue, JSC::EncodedJSValue* readableStreamTee);
 
 JSBunRequest* JSBunRequest::clone(JSC::VM& vm, JSGlobalObject* globalObject)
 {
     auto throwScope = DECLARE_THROW_SCOPE(vm);
 
+    JSC::EncodedJSValue readableStreamTee[2] { encodedJSValue(), encodedJSValue() };
     auto* structure = defaultGlobalObject(globalObject)->m_JSBunRequestStructure.getInitializedOnMainThread(globalObject);
-    auto* raw = Request__clone(this->wrapped(), globalObject);
+    auto* raw = Request__clone(this->wrapped(), globalObject, JSValue::encode(this), readableStreamTee);
     EXCEPTION_ASSERT(!!raw == !throwScope.exception());
     RETURN_IF_EXCEPTION(throwScope, nullptr);
     auto* clone = this->create(vm, structure, raw, nullptr);
@@ -134,9 +135,18 @@ JSBunRequest* JSBunRequest::clone(JSC::VM& vm, JSGlobalObject* globalObject)
             auto cookieMapClone = cookieMap->clone();
             auto cookies = WebCore::toJSNewlyCreated(globalObject, jsCast<JSDOMGlobalObject*>(globalObject), WTFMove(cookieMapClone));
             clone->setCookies(cookies.getObject());
+            RETURN_IF_EXCEPTION(throwScope, nullptr);
         }
     }
 
+    if (readableStreamTee[0] != encodedJSValue()) {
+        this->m_body.set(vm, this, JSValue::decode(readableStreamTee[0]));
+    }
+
+    if (readableStreamTee[1] != encodedJSValue()) {
+        clone->m_body.set(vm, clone, JSValue::decode(readableStreamTee[1]));
+    }
+
     RELEASE_AND_RETURN(throwScope, clone);
 }
 

src/bun.js/bindings/JSGlobalObject.zig

@@ -899,10 +899,11 @@ pub const JSGlobalObject = opaque {
         }
 
         // We're done validating. From now on, deal with extracting the body.
-        body.toBlobIfPossible();
+        const owner = jsc.WebCore.ReadableStream.Ref.Owner{ .Response = response_value };
+        body.toBlobIfPossible(owner);
 
         var any_blob = switch (body.*) {
-            .Locked => body.tryUseAsAnyBlob() orelse return body.toReadableStream(this),
+            .Locked => body.tryUseAsAnyBlob(owner) orelse return body.toReadableStream(owner, this),
             else => body.useAsAnyBlob(),
         };
 

src/bun.js/node/types.zig

@@ -86,9 +86,9 @@ pub const BlobOrStringOrBuffer = union(enum) {
                 }
                 if (allow_request_response) {
                     if (value.as(jsc.WebCore.Request)) |request| {
-                        request.body.value.toBlobIfPossible();
+                        request.body.value.toBlobIfPossible(.{ .empty = {} });
 
-                        if (request.body.value.tryUseAsAnyBlob()) |any_blob_| {
+                        if (request.body.value.tryUseAsAnyBlob(.{ .empty = {} })) |any_blob_| {
                             var any_blob = any_blob_;
                             defer any_blob.detach();
                             return .{ .blob = any_blob.toBlob(global) };
@@ -98,9 +98,10 @@ pub const BlobOrStringOrBuffer = union(enum) {
                     }
 
                     if (value.as(jsc.WebCore.Response)) |response| {
-                        response.body.value.toBlobIfPossible();
+                        const owner = if (response.this_jsvalue.tryGet()) |jsval| jsc.WebCore.ReadableStream.Ref.Owner{ .Response = jsval } else jsc.WebCore.ReadableStream.Ref.Owner{ .empty = {} };
+                        response.body.value.toBlobIfPossible(owner);
 
-                        if (response.body.value.tryUseAsAnyBlob()) |any_blob_| {
+                        if (response.body.value.tryUseAsAnyBlob(owner)) |any_blob_| {
                             var any_blob = any_blob_;
                             defer any_blob.detach();
                             return .{ .blob = any_blob.toBlob(global) };

src/bun.js/webcore/BakeResponse.zig

@@ -44,7 +44,7 @@ pub fn constructor(globalThis: *jsc.JSGlobalObject, callframe: *jsc.CallFrame, b
         }
     }
 
-    return Response.constructor(globalThis, callframe);
+    return Response.constructor(globalThis, callframe, .zero);
 }
 
 pub export fn BakeResponseClass__constructRedirect(globalObject: *jsc.JSGlobalObject, callFrame: *jsc.CallFrame) callconv(jsc.conv) jsc.JSValue {