fix(AbortSignal/fetch) fix AbortSignal.timeout, fetch lock behavior and fetch errors (#6390)

* fix abort signal and fetch error

* fix fetch error and lock behavior

9 files changed, 50 insertions, 23 deletions

diff --git a/src/bun.js/api/bun.zig b/src/bun.js/api/bun.zig
index 0862e41ec..21c2ecd0e 100644
--- a/src/bun.js/api/bun.zig
+++ b/src/bun.js/api/bun.zig
@@ -3604,7 +3604,7 @@ pub const Timer = struct {
 
             this.poll_ref.unref(vm);
 
-            this.timer.deinit();
+            this.timer.deinit(false);
 
             // balance double unreffing in doUnref
             vm.event_loop_handle.?.num_polls += @as(i32, @intFromBool(this.did_unref_timer));
diff --git a/src/bun.js/api/ffi.zig b/src/bun.js/api/ffi.zig
index 234b58888..a7a03e784 100644
--- a/src/bun.js/api/ffi.zig
+++ b/src/bun.js/api/ffi.zig
@@ -317,9 +317,9 @@ pub const FFI = struct {
                 };
             };
         };
-        
+
         var size = symbols.values().len;
-        if(size >= 63) {
+        if (size >= 63) {
             size = 0;
         }
         var obj = JSC.JSValue.createEmptyObject(global, size);
diff --git a/src/bun.js/bindings/ZigGlobalObject.cpp b/src/bun.js/bindings/ZigGlobalObject.cpp
index b9670f730..777e398d3 100644
--- a/src/bun.js/bindings/ZigGlobalObject.cpp
+++ b/src/bun.js/bindings/ZigGlobalObject.cpp
@@ -3717,6 +3717,7 @@ void GlobalObject::addBuiltinGlobals(JSC::VM& vm)
 
     putDirectBuiltinFunction(vm, this, builtinNames.createFIFOPrivateName(), streamInternalsCreateFIFOCodeGenerator(vm), PropertyAttribute::Builtin | PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly);
     putDirectBuiltinFunction(vm, this, builtinNames.createEmptyReadableStreamPrivateName(), readableStreamCreateEmptyReadableStreamCodeGenerator(vm), PropertyAttribute::Builtin | PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly);
+    putDirectBuiltinFunction(vm, this, builtinNames.createUsedReadableStreamPrivateName(), readableStreamCreateUsedReadableStreamCodeGenerator(vm), PropertyAttribute::Builtin | PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly);
     putDirectBuiltinFunction(vm, this, builtinNames.consumeReadableStreamPrivateName(), readableStreamConsumeReadableStreamCodeGenerator(vm), PropertyAttribute::Builtin | PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly);
     putDirectBuiltinFunction(vm, this, builtinNames.createNativeReadableStreamPrivateName(), readableStreamCreateNativeReadableStreamCodeGenerator(vm), PropertyAttribute::Builtin | PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly);
     putDirectBuiltinFunction(vm, this, builtinNames.requireESMPrivateName(), importMetaObjectRequireESMCodeGenerator(vm), PropertyAttribute::Builtin | PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly);
diff --git a/src/bun.js/bindings/bindings.cpp b/src/bun.js/bindings/bindings.cpp
index 9ba053a23..2efbde7b0 100644
--- a/src/bun.js/bindings/bindings.cpp
+++ b/src/bun.js/bindings/bindings.cpp
@@ -2165,6 +2165,14 @@ JSC__JSValue ReadableStream__empty(Zig::GlobalObject* globalObject)
     return JSValue::encode(JSC::call(globalObject, function, JSC::ArgList(), "ReadableStream.create"_s));
 }
 
+JSC__JSValue ReadableStream__used(Zig::GlobalObject* globalObject)
+{
+    auto& vm = globalObject->vm();
+    auto clientData = WebCore::clientData(vm);
+    auto* function = globalObject->getDirect(vm, clientData->builtinNames().createUsedReadableStreamPrivateName()).getObject();
+    return JSValue::encode(JSC::call(globalObject, function, JSC::ArgList(), "ReadableStream.create"_s));
+}
+
 JSC__JSValue JSC__JSValue__createRangeError(const ZigString* message, const ZigString* arg1,
     JSC__JSGlobalObject* globalObject)
 {
diff --git a/src/bun.js/event_loop.zig b/src/bun.js/event_loop.zig
index 278e62327..8bc6a771e 100644
--- a/src/bun.js/event_loop.zig
+++ b/src/bun.js/event_loop.zig
@@ -1148,7 +1148,7 @@ pub const EventLoop = struct {
 
     pub fn callTask(timer: *uws.Timer) callconv(.C) void {
         var task = Task.from(timer.as(*anyopaque));
-        timer.deinit();
+        defer timer.deinit(true);
 
         JSC.VirtualMachine.get().enqueueTask(task);
     }
diff --git a/src/bun.js/node/node_fs_stat_watcher.zig b/src/bun.js/node/node_fs_stat_watcher.zig
index 3bf262f07..158a08ff7 100644
--- a/src/bun.js/node/node_fs_stat_watcher.zig
+++ b/src/bun.js/node/node_fs_stat_watcher.zig
@@ -97,7 +97,7 @@ pub const StatWatcherScheduler = struct {
                 prev = next;
             } else {
                 if (this.head.load(.Monotonic) == null) {
-                    this.timer.?.deinit();
+                    this.timer.?.deinit(false);
                     this.timer = null;
                     // The scheduler is not deinit here, but it will get reused.
                 }
diff --git a/src/bun.js/webcore/body.zig b/src/bun.js/webcore/body.zig
index bbdf21d5d..76cda1bad 100644
--- a/src/bun.js/webcore/body.zig
+++ b/src/bun.js/webcore/body.zig
@@ -466,7 +466,10 @@ pub const Body = struct {
             JSC.markBinding(@src());
 
             switch (this.*) {
-                .Used, .Empty => {
+                .Used => {
+                    return JSC.WebCore.ReadableStream.used(globalThis);
+                },
+                .Empty => {
                     return JSC.WebCore.ReadableStream.empty(globalThis);
                 },
                 .Null => {
@@ -493,6 +496,9 @@ pub const Body = struct {
                     if (locked.readable) |readable| {
                         return readable.value;
                     }
+                    if (locked.promise != null) {
+                        return JSC.WebCore.ReadableStream.used(globalThis);
+                    }
                     var drain_result: JSC.WebCore.DrainResult = .{
                         .estimated_size = 0,
                     };
@@ -1104,8 +1110,7 @@ pub fn BodyMixin(comptime Type: type) type {
             var body: *Body.Value = this.getBodyValue();
 
             if (body.* == .Used) {
-                // TODO: make this closed
-                return JSC.WebCore.ReadableStream.empty(globalThis);
+                return JSC.WebCore.ReadableStream.used(globalThis);
             }
 
             return body.toReadableStream(globalThis);
diff --git a/src/bun.js/webcore/response.zig b/src/bun.js/webcore/response.zig
index 0e80adfc4..9a2ec48ba 100644
--- a/src/bun.js/webcore/response.zig
+++ b/src/bun.js/webcore/response.zig
@@ -778,27 +778,27 @@ pub const Fetch = struct {
             if (!success) {
                 const err = this.onReject();
                 err.ensureStillAlive();
+                // if we are streaming update with error
+                if (this.readable_stream_ref.get()) |readable| {
+                    readable.ptr.Bytes.onData(
+                        .{
+                            .err = .{ .JSValue = err },
+                        },
+                        bun.default_allocator,
+                    );
+                }
+                // if we are buffering resolve the promise
                 if (this.response.get()) |response_js| {
                     if (response_js.as(Response)) |response| {
                         const body = response.body;
-                        if (body.value == .Locked) {
-                            if (body.value.Locked.readable) |readable| {
-                                readable.ptr.Bytes.onData(
-                                    .{
-                                        .err = .{ .JSValue = err },
-                                    },
-                                    bun.default_allocator,
-                                );
-                                return;
-                            }
+                        if (body.value.Locked.promise) |promise_| {
+                            const promise = promise_.asAnyPromise().?;
+                            promise.reject(globalThis, err);
                         }
 
                         response.body.value.toErrorInstance(err, globalThis);
-                        return;
                     }
                 }
-
-                globalThis.throwValue(err);
                 return;
             }
 
@@ -1708,7 +1708,7 @@ pub const Fetch = struct {
                             if (decompress.isBoolean()) {
                                 disable_decompression = !decompress.asBoolean();
                             } else if (decompress.isNumber()) {
-                                disable_keepalive = decompress.to(i32) == 0;
+                                disable_decompression = decompress.to(i32) == 0;
                             }
                         }
 
@@ -1901,7 +1901,7 @@ pub const Fetch = struct {
                             if (decompress.isBoolean()) {
                                 disable_decompression = !decompress.asBoolean();
                             } else if (decompress.isNumber()) {
-                                disable_keepalive = decompress.to(i32) == 0;
+                                disable_decompression = decompress.to(i32) == 0;
                             }
                         }
 
diff --git a/src/bun.js/webcore/streams.zig b/src/bun.js/webcore/streams.zig
index 2e6bbdce2..1f95659e2 100644
--- a/src/bun.js/webcore/streams.zig
+++ b/src/bun.js/webcore/streams.zig
@@ -225,6 +225,7 @@ pub const ReadableStream = struct {
     extern fn ReadableStream__isDisturbed(possibleReadableStream: JSValue, globalObject: *JSGlobalObject) bool;
     extern fn ReadableStream__isLocked(possibleReadableStream: JSValue, globalObject: *JSGlobalObject) bool;
     extern fn ReadableStream__empty(*JSGlobalObject) JSC.JSValue;
+    extern fn ReadableStream__used(*JSGlobalObject) JSC.JSValue;
     extern fn ReadableStream__cancel(stream: JSValue, *JSGlobalObject) void;
     extern fn ReadableStream__abort(stream: JSValue, *JSGlobalObject) void;
     extern fn ReadableStream__detach(stream: JSValue, *JSGlobalObject) void;
@@ -367,6 +368,12 @@ pub const ReadableStream = struct {
         return ReadableStream__empty(globalThis);
     }
 
+    pub fn used(globalThis: *JSGlobalObject) JSC.JSValue {
+        JSC.markBinding(@src());
+
+        return ReadableStream__used(globalThis);
+    }
+
     const Base = @import("../../ast/base.zig");
     pub const StreamTag = enum(usize) {
         invalid = 0,
@@ -3596,6 +3603,9 @@ pub const ByteStream = struct {
                     this.buffer = try std.ArrayList(u8).initCapacity(bun.default_allocator, chunk.len);
                     this.buffer.appendSliceAssumeCapacity(chunk);
                 },
+                .err => {
+                    this.pending.result = .{ .err = stream.err };
+                },
                 else => unreachable,
             }
             return;
@@ -3605,6 +3615,9 @@ pub const ByteStream = struct {
             .temporary_and_done, .temporary => {
                 try this.buffer.appendSlice(chunk);
             },
+            .err => {
+                this.pending.result = .{ .err = stream.err };
+            },
             // We don't support the rest of these yet
             else => unreachable,
         }