Make `Bun.spawn`, FileSink and FileBlobLoader a little more reliable

14 files changed, 550 insertions, 209 deletions

diff --git a/src/bun.js/api/bun.classes.ts b/src/bun.js/api/bun.classes.ts
index 3a74549d2..5a6ae47cc 100644
--- a/src/bun.js/api/bun.classes.ts
+++ b/src/bun.js/api/bun.classes.ts
@@ -42,8 +42,12 @@ export default [
         getter: "getKilled",
       },
 
-      exitStatus: {
-        getter: "getExitStatus",
+      exitCode: {
+        getter: "getExitCode",
+      },
+
+      exited: {
+        getter: "getExited",
         cache: true,
       },
     },
diff --git a/src/bun.js/api/bun.zig b/src/bun.js/api/bun.zig
index d62b1e9b8..0b4f62c39 100644
--- a/src/bun.js/api/bun.zig
+++ b/src/bun.js/api/bun.zig
@@ -3384,7 +3384,8 @@ pub const Subprocess = struct {
     stderr: Readable,
 
     killed: bool = false,
-    has_ref: bool = false,
+    reffer: JSC.Ref = JSC.Ref.init(),
+    poll_ref: JSC.PollRef = JSC.PollRef.init(),
 
     exit_promise: JSValue = JSValue.zero,
     this_jsvalue: JSValue = JSValue.zero,
@@ -3402,6 +3403,16 @@ pub const Subprocess = struct {
 
     globalThis: *JSC.JSGlobalObject,
 
+    pub fn ref(this: *Subprocess) void {
+        this.reffer.ref(this.globalThis.bunVM());
+        this.poll_ref.ref(this.globalThis.bunVM());
+    }
+
+    pub fn unref(this: *Subprocess) void {
+        this.reffer.unref(this.globalThis.bunVM());
+        this.poll_ref.unref(this.globalThis.bunVM());
+    }
+
     pub fn constructor(
         _: *JSC.JSGlobalObject,
         _: *JSC.CallFrame,
@@ -3425,8 +3436,9 @@ pub const Subprocess = struct {
                     defer blob.detach();
 
                     var stream = JSC.WebCore.ReadableStream.fromBlob(globalThis, &blob, 0);
-
-                    break :brk Readable{ .pipe = JSC.WebCore.ReadableStream.fromJS(stream, globalThis).? };
+                    var out = JSC.WebCore.ReadableStream.fromJS(stream, globalThis).?;
+                    out.ptr.File.stored_global_this_ = globalThis;
+                    break :brk Readable{ .pipe = out };
                 },
                 .callback, .fd, .path, .blob => Readable{ .fd = @intCast(JSC.Node.FileDescriptor, fd) },
             };
@@ -3559,20 +3571,6 @@ pub const Subprocess = struct {
         this.stderr.close();
     }
 
-    pub fn unref(this: *Subprocess) void {
-        if (!this.has_ref)
-            return;
-        this.has_ref = false;
-        this.globalThis.bunVM().active_tasks -= 1;
-    }
-
-    pub fn ref(this: *Subprocess) void {
-        if (this.has_ref)
-            return;
-        this.has_ref = true;
-        this.globalThis.bunVM().active_tasks += 1;
-    }
-
     pub fn doRef(this: *Subprocess, _: *JSC.JSGlobalObject, _: *JSC.CallFrame) callconv(.C) JSValue {
         this.ref();
         return JSC.JSValue.jsUndefined();
@@ -3608,7 +3606,7 @@ pub const Subprocess = struct {
                 .path, .pipe, .callback => {
                     var sink = try globalThis.bunVM().allocator.create(JSC.WebCore.FileSink);
                     sink.* = .{
-                        .opened_fd = fd,
+                        .fd = fd,
                         .buffer = bun.ByteList.init(&.{}),
                         .allocator = globalThis.bunVM().allocator,
                     };
@@ -3659,7 +3657,7 @@ pub const Subprocess = struct {
             bun.default_allocator.destroy(this);
     }
 
-    pub fn getExitStatus(
+    pub fn getExited(
         this: *Subprocess,
         globalThis: *JSGlobalObject,
     ) callconv(.C) JSValue {
@@ -3674,6 +3672,16 @@ pub const Subprocess = struct {
         return this.exit_promise;
     }
 
+    pub fn getExitCode(
+        this: *Subprocess,
+        _: *JSGlobalObject,
+    ) callconv(.C) JSValue {
+        if (this.exit_code) |code| {
+            return JSC.JSValue.jsNumber(code);
+        }
+        return JSC.JSValue.jsNull();
+    }
+
     pub fn spawn(globalThis: *JSC.JSGlobalObject, args: JSValue) JSValue {
         var arena = std.heap.ArenaAllocator.init(bun.default_allocator);
         defer arena.deinit();
diff --git a/src/bun.js/base.zig b/src/bun.js/base.zig
index 6a4cc7469..09a84450e 100644
--- a/src/bun.js/base.zig
+++ b/src/bun.js/base.zig
@@ -22,7 +22,7 @@ const Request = WebCore.Request;
 const Router = @import("./api/router.zig");
 const FetchEvent = WebCore.FetchEvent;
 const IdentityContext = @import("../identity_context.zig").IdentityContext;
-
+const uws = @import("uws");
 const Body = WebCore.Body;
 const TaggedPointerTypes = @import("../tagged_pointer.zig");
 const TaggedPointerUnion = TaggedPointerTypes.TaggedPointerUnion;
@@ -370,21 +370,6 @@ pub const To = struct {
             };
         }
     };
-
-    pub const Ref = struct {
-        pub inline fn str(ref: anytype) js.JSStringRef {
-            return @as(js.JSStringRef, ref);
-        }
-    };
-
-    pub const Zig = struct {
-        pub inline fn str(ref: anytype, buf: anytype) string {
-            return buf[0..js.JSStringGetUTF8CString(Ref.str(ref), buf.ptr, buf.len)];
-        }
-        pub inline fn ptr(comptime StructType: type, obj: js.JSObjectRef) *StructType {
-            return GetJSPrivateData(StructType, obj).?;
-        }
-    };
 };
 
 pub const Properties = struct {
@@ -3864,3 +3849,130 @@ pub fn cachedBoundFunction(comptime name: [:0]const u8, comptime callback: anyty
         }
     }.getter;
 }
+
+/// Track whether an object should keep the event loop alive
+pub const Ref = struct {
+    has: bool = false,
+
+    pub fn init() Ref {
+        return .{};
+    }
+
+    pub fn unref(this: *Ref, vm: *JSC.VirtualMachine) void {
+        if (!this.has)
+            return;
+        this.has = false;
+        vm.active_tasks -= 1;
+    }
+
+    pub fn ref(this: *Ref, vm: *JSC.VirtualMachine) void {
+        if (this.has)
+            return;
+        this.has = true;
+        vm.active_tasks += 1;
+    }
+};
+
+/// Track if an object whose file descriptor is being watched should keep the event loop alive.
+/// This is not reference counted. It only tracks active or inactive.
+pub const PollRef = struct {
+    status: Status = .inactive,
+
+    const Status = enum { active, inactive, done };
+
+    /// Make calling ref() on this poll into a no-op.
+    pub fn disable(this: *PollRef) void {
+        this.unref();
+        this.status = .done;
+    }
+
+    /// Only intended to be used from EventLoop.Poller
+    pub fn deactivate(this: *PollRef, loop: *uws.Loop) void {
+        if (this.status != .active)
+            return;
+
+        this.status = .inactive;
+        loop.num_polls -= 1;
+        loop.active -= 1;
+    }
+
+    /// Only intended to be used from EventLoop.Poller
+    pub fn activate(this: *PollRef, loop: *uws.Loop) void {
+        if (this.status != .inactive)
+            return;
+
+        this.status = .active;
+        loop.num_polls += 1;
+        loop.active += 1;
+    }
+
+    pub fn init() PollRef {
+        return .{};
+    }
+
+    /// Prevent a poll from keeping the process alive.
+    pub fn unref(this: *PollRef, vm: *JSC.VirtualMachine) void {
+        if (this.status != .active)
+            return;
+        this.status = .inactive;
+        vm.uws_event_loop.?.num_polls -= 1;
+        vm.uws_event_loop.?.active -= 1;
+    }
+
+    /// Allow a poll to keep the process alive.
+    pub fn ref(this: *PollRef, vm: *JSC.VirtualMachine) void {
+        if (this.status != .inactive)
+            return;
+        this.status = .active;
+        vm.uws_event_loop.?.num_polls += 1;
+        vm.uws_event_loop.?.active += 1;
+    }
+};
+
+pub const Strong = struct {
+    ref: ?*JSC.napi.Ref = null,
+
+    pub fn init() Strong {
+        return .{};
+    }
+
+    pub fn get(this: *Strong) ?JSValue {
+        var ref = this.ref orelse return null;
+        const result = ref.get();
+        if (result == .zero) {
+            return null;
+        }
+
+        return result;
+    }
+
+    pub fn swap(this: *Strong) JSValue {
+        var ref = this.ref orelse return .zero;
+        const result = ref.get();
+        if (result == .zero) {
+            return .zero;
+        }
+
+        ref.set(.zero);
+        return result;
+    }
+
+    pub fn set(this: *Strong, globalThis: *JSC.JSGlobalObject, value: JSValue) void {
+        var ref: *JSC.napi.Ref = this.ref orelse {
+            this.ref = JSC.napi.Ref.create(globalThis, value);
+            return;
+        };
+        ref.set(value);
+    }
+
+    pub fn clear(this: *Strong) void {
+        var ref: *JSC.napi.Ref = this.ref orelse return;
+        ref.set(JSC.JSValue.zero);
+    }
+
+    pub fn deinit(this: *Strong) void {
+        var ref: *JSC.napi.Ref = this.ref orelse return;
+        this.ref = null;
+        ref.destroy();
+    }
+};
diff --git a/src/bun.js/bindings/ZigGeneratedClasses.cpp b/src/bun.js/bindings/ZigGeneratedClasses.cpp
index 925814c2d..aefc1f43c 100644
--- a/src/bun.js/bindings/ZigGeneratedClasses.cpp
+++ b/src/bun.js/bindings/ZigGeneratedClasses.cpp
@@ -27,8 +27,12 @@ extern "C" void* SubprocessClass__construct(JSC::JSGlobalObject*, JSC::CallFrame
 JSC_DECLARE_CUSTOM_GETTER(jsSubprocessConstructor);
 extern "C" void SubprocessClass__finalize(void*);
 
-extern "C" JSC::EncodedJSValue SubprocessPrototype__getExitStatus(void* ptr, JSC::JSGlobalObject* lexicalGlobalObject);
-JSC_DECLARE_CUSTOM_GETTER(SubprocessPrototype__exitStatusGetterWrap);
+extern "C" JSC::EncodedJSValue SubprocessPrototype__getExitCode(void* ptr, JSC::JSGlobalObject* lexicalGlobalObject);
+JSC_DECLARE_CUSTOM_GETTER(SubprocessPrototype__exitCodeGetterWrap);
+
+
+extern "C" JSC::EncodedJSValue SubprocessPrototype__getExited(void* ptr, JSC::JSGlobalObject* lexicalGlobalObject);
+JSC_DECLARE_CUSTOM_GETTER(SubprocessPrototype__exitedGetterWrap);
 
 
 extern "C" EncodedJSValue SubprocessPrototype__kill(void* ptr, JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame);
@@ -67,7 +71,8 @@ STATIC_ASSERT_ISO_SUBSPACE_SHARABLE(JSSubprocessPrototype, JSSubprocessPrototype
 
 
   static const HashTableValue JSSubprocessPrototypeTableValues[] = {
-{ "exitStatus"_s, static_cast<unsigned>(JSC::PropertyAttribute::ReadOnly | JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::DOMAttribute), NoIntrinsic, { HashTableValue::GetterSetterType, SubprocessPrototype__exitStatusGetterWrap, 0 } }  ,
+{ "exitCode"_s, static_cast<unsigned>(JSC::PropertyAttribute::ReadOnly | JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::DOMAttribute), NoIntrinsic, { HashTableValue::GetterSetterType, SubprocessPrototype__exitCodeGetterWrap, 0 } }  ,
+{ "exited"_s, static_cast<unsigned>(JSC::PropertyAttribute::ReadOnly | JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::DOMAttribute), NoIntrinsic, { HashTableValue::GetterSetterType, SubprocessPrototype__exitedGetterWrap, 0 } }  ,
 { "kill"_s, static_cast<unsigned>(JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, SubprocessPrototype__killCallback, 1 } }  ,
 { "killed"_s, static_cast<unsigned>(JSC::PropertyAttribute::ReadOnly | JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::DOMAttribute), NoIntrinsic, { HashTableValue::GetterSetterType, SubprocessPrototype__killedGetterWrap, 0 } }  ,
 { "pid"_s, static_cast<unsigned>(JSC::PropertyAttribute::ReadOnly | JSC::PropertyAttribute::CustomAccessor | JSC::PropertyAttribute::DOMAttribute), NoIntrinsic, { HashTableValue::GetterSetterType, SubprocessPrototype__pidGetterWrap, 0 } }  ,
@@ -97,25 +102,31 @@ JSC_DEFINE_CUSTOM_GETTER(jsSubprocessConstructor, (JSGlobalObject * lexicalGloba
     
 
 
-JSC_DEFINE_CUSTOM_GETTER(SubprocessPrototype__exitStatusGetterWrap, (JSGlobalObject * lexicalGlobalObject, EncodedJSValue thisValue, PropertyName attributeName))
+JSC_DEFINE_CUSTOM_GETTER(SubprocessPrototype__exitCodeGetterWrap, (JSGlobalObject * lexicalGlobalObject, EncodedJSValue thisValue, PropertyName attributeName))
 {
     auto& vm = lexicalGlobalObject->vm();
     Zig::GlobalObject *globalObject = reinterpret_cast<Zig::GlobalObject*>(lexicalGlobalObject);
     auto throwScope = DECLARE_THROW_SCOPE(vm);
     JSSubprocess* thisObject = jsCast<JSSubprocess*>(JSValue::decode(thisValue));
-      JSC::EnsureStillAliveScope thisArg = JSC::EnsureStillAliveScope(thisObject);
-    
-    if (JSValue cachedValue = thisObject->m_exitStatus.get())
-        return JSValue::encode(cachedValue);
-    
-    JSC::JSValue result = JSC::JSValue::decode(
-        SubprocessPrototype__getExitStatus(thisObject->wrapped(), globalObject)
-    );
+    JSC::EnsureStillAliveScope thisArg = JSC::EnsureStillAliveScope(thisObject);
+    JSC::EncodedJSValue result = SubprocessPrototype__getExitCode(thisObject->wrapped(), globalObject);
     RETURN_IF_EXCEPTION(throwScope, {});
-    thisObject->m_exitStatus.set(vm, thisObject, result);
-    RELEASE_AND_RETURN(throwScope, JSValue::encode(result));
+    RELEASE_AND_RETURN(throwScope, result);
 }
+        
 
+JSC_DEFINE_CUSTOM_GETTER(SubprocessPrototype__exitedGetterWrap, (JSGlobalObject * lexicalGlobalObject, EncodedJSValue thisValue, PropertyName attributeName))
+{
+    auto& vm = lexicalGlobalObject->vm();
+    Zig::GlobalObject *globalObject = reinterpret_cast<Zig::GlobalObject*>(lexicalGlobalObject);
+    auto throwScope = DECLARE_THROW_SCOPE(vm);
+    JSSubprocess* thisObject = jsCast<JSSubprocess*>(JSValue::decode(thisValue));
+    JSC::EnsureStillAliveScope thisArg = JSC::EnsureStillAliveScope(thisObject);
+    JSC::EncodedJSValue result = SubprocessPrototype__getExited(thisObject->wrapped(), globalObject);
+    RETURN_IF_EXCEPTION(throwScope, {});
+    RELEASE_AND_RETURN(throwScope, result);
+}
+        
 
 JSC_DEFINE_HOST_FUNCTION(SubprocessPrototype__killCallback, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
 {
@@ -394,7 +405,6 @@ void JSSubprocess::visitChildrenImpl(JSCell* cell, Visitor& visitor)
     JSSubprocess* thisObject = jsCast<JSSubprocess*>(cell);
     ASSERT_GC_OBJECT_INHERITS(thisObject, info());
     Base::visitChildren(thisObject, visitor);
-    visitor.append(thisObject->m_exitStatus);
     visitor.append(thisObject->m_stderr);
     visitor.append(thisObject->m_stdin);
     visitor.append(thisObject->m_stdout);
diff --git a/src/bun.js/bindings/ZigGeneratedClasses.h b/src/bun.js/bindings/ZigGeneratedClasses.h
index ad85dd3f9..22ffe60d7 100644
--- a/src/bun.js/bindings/ZigGeneratedClasses.h
+++ b/src/bun.js/bindings/ZigGeneratedClasses.h
@@ -5,6 +5,8 @@
 
 #include "root.h"
 
+#include "JSEventEmitter.h"
+
 namespace Zig {
 }
 
@@ -15,9 +17,9 @@ namespace WebCore {
 using namespace Zig;
 using namespace JSC;
 
-class JSSubprocess final : public JSC::JSDestructibleObject {
+class JSSubprocess final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSSubprocess* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -66,8 +68,7 @@ class JSSubprocess final : public JSC::JSDestructibleObject {
 
         DECLARE_VISIT_CHILDREN;
 
-        mutable JSC::WriteBarrier<JSC::Unknown> m_exitStatus;
-mutable JSC::WriteBarrier<JSC::Unknown> m_stderr;
+        mutable JSC::WriteBarrier<JSC::Unknown> m_stderr;
 mutable JSC::WriteBarrier<JSC::Unknown> m_stdin;
 mutable JSC::WriteBarrier<JSC::Unknown> m_stdout;
     };
@@ -142,9 +143,9 @@ class JSSubprocessPrototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSSubprocessPrototype* prototype);
     };
-class JSSHA1 final : public JSC::JSDestructibleObject {
+class JSSHA1 final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSSHA1* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -266,9 +267,9 @@ class JSSHA1Prototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSSHA1Prototype* prototype);
     };
-class JSMD5 final : public JSC::JSDestructibleObject {
+class JSMD5 final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSMD5* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -390,9 +391,9 @@ class JSMD5Prototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSMD5Prototype* prototype);
     };
-class JSMD4 final : public JSC::JSDestructibleObject {
+class JSMD4 final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSMD4* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -514,9 +515,9 @@ class JSMD4Prototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSMD4Prototype* prototype);
     };
-class JSSHA224 final : public JSC::JSDestructibleObject {
+class JSSHA224 final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSSHA224* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -638,9 +639,9 @@ class JSSHA224Prototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSSHA224Prototype* prototype);
     };
-class JSSHA512 final : public JSC::JSDestructibleObject {
+class JSSHA512 final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSSHA512* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -762,9 +763,9 @@ class JSSHA512Prototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSSHA512Prototype* prototype);
     };
-class JSSHA384 final : public JSC::JSDestructibleObject {
+class JSSHA384 final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSSHA384* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -886,9 +887,9 @@ class JSSHA384Prototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSSHA384Prototype* prototype);
     };
-class JSSHA256 final : public JSC::JSDestructibleObject {
+class JSSHA256 final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSSHA256* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -1010,9 +1011,9 @@ class JSSHA256Prototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSSHA256Prototype* prototype);
     };
-class JSSHA512_256 final : public JSC::JSDestructibleObject {
+class JSSHA512_256 final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSSHA512_256* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -1134,9 +1135,9 @@ class JSSHA512_256Prototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSSHA512_256Prototype* prototype);
     };
-class JSTextDecoder final : public JSC::JSDestructibleObject {
+class JSTextDecoder final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSTextDecoder* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -1258,9 +1259,9 @@ class JSTextDecoderPrototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSTextDecoderPrototype* prototype);
     };
-class JSRequest final : public JSC::JSDestructibleObject {
+class JSRequest final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSRequest* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -1384,9 +1385,9 @@ class JSRequestPrototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSRequestPrototype* prototype);
     };
-class JSResponse final : public JSC::JSDestructibleObject {
+class JSResponse final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSResponse* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
@@ -1511,9 +1512,9 @@ class JSResponsePrototype final : public JSC::JSNonFinalObject {
     
         void finishCreation(JSC::VM&, JSC::JSGlobalObject* globalObject, JSResponsePrototype* prototype);
     };
-class JSBlob final : public JSC::JSDestructibleObject {
+class JSBlob final : public JSDestructibleObject {
     public:
-        using Base = JSC::JSDestructibleObject;
+        using Base = JSDestructibleObject;
         static JSBlob* create(JSC::VM& vm, JSC::JSGlobalObject* globalObject, JSC::Structure* structure, void* ctx);
     
         DECLARE_EXPORT_INFO;
diff --git a/src/bun.js/bindings/generated_classes.zig b/src/bun.js/bindings/generated_classes.zig
index 7230de74a..a05d6a958 100644
--- a/src/bun.js/bindings/generated_classes.zig
+++ b/src/bun.js/bindings/generated_classes.zig
@@ -61,8 +61,11 @@ pub const JSSubprocess = struct {
             @compileLog("Subprocess.finalize is not a finalizer");
         }
 
-        if (@TypeOf(Subprocess.getExitStatus) != GetterType)
-            @compileLog("Expected Subprocess.getExitStatus to be a getter");
+        if (@TypeOf(Subprocess.getExitCode) != GetterType)
+            @compileLog("Expected Subprocess.getExitCode to be a getter");
+
+        if (@TypeOf(Subprocess.getExited) != GetterType)
+            @compileLog("Expected Subprocess.getExited to be a getter");
 
         if (@TypeOf(Subprocess.kill) != CallbackType)
             @compileLog("Expected Subprocess.kill to be a callback");
@@ -90,7 +93,8 @@ pub const JSSubprocess = struct {
             @export(Subprocess.doRef, .{ .name = "SubprocessPrototype__doRef" });
             @export(Subprocess.doUnref, .{ .name = "SubprocessPrototype__doUnref" });
             @export(Subprocess.finalize, .{ .name = "SubprocessClass__finalize" });
-            @export(Subprocess.getExitStatus, .{ .name = "SubprocessPrototype__getExitStatus" });
+            @export(Subprocess.getExitCode, .{ .name = "SubprocessPrototype__getExitCode" });
+            @export(Subprocess.getExited, .{ .name = "SubprocessPrototype__getExited" });
             @export(Subprocess.getKilled, .{ .name = "SubprocessPrototype__getKilled" });
             @export(Subprocess.getPid, .{ .name = "SubprocessPrototype__getPid" });
             @export(Subprocess.getStderr, .{ .name = "SubprocessPrototype__getStderr" });
diff --git a/src/bun.js/bindings/napi.cpp b/src/bun.js/bindings/napi.cpp
index b4bef19f9..abd603c41 100644
--- a/src/bun.js/bindings/napi.cpp
+++ b/src/bun.js/bindings/napi.cpp
@@ -826,6 +826,12 @@ extern "C" napi_status napi_delete_reference(napi_env env, napi_ref ref)
     return napi_ok;
 }
 
+extern "C" void napi_delete_reference_internal(napi_ref ref)
+{
+    NapiRef* napiRef = toJS(ref);
+    napiRef->~NapiRef();
+}
+
 extern "C" napi_status napi_is_detached_arraybuffer(napi_env env,
     napi_value arraybuffer,
     bool* result)
diff --git a/src/bun.js/builtins/cpp/ReadableStreamInternalsBuiltins.cpp b/src/bun.js/builtins/cpp/ReadableStreamInternalsBuiltins.cpp
index 79db2b727..80945d27e 100644
--- a/src/bun.js/builtins/cpp/ReadableStreamInternalsBuiltins.cpp
+++ b/src/bun.js/builtins/cpp/ReadableStreamInternalsBuiltins.cpp
@@ -2253,7 +2253,7 @@ const char* const s_readableStreamInternalsReadableStreamDefaultControllerCanClo
 const JSC::ConstructAbility s_readableStreamInternalsLazyLoadStreamCodeConstructAbility = JSC::ConstructAbility::CannotConstruct;
 const JSC::ConstructorKind s_readableStreamInternalsLazyLoadStreamCodeConstructorKind = JSC::ConstructorKind::None;
 const JSC::ImplementationVisibility s_readableStreamInternalsLazyLoadStreamCodeImplementationVisibility = JSC::ImplementationVisibility::Public;
-const int s_readableStreamInternalsLazyLoadStreamCodeLength = 2512;
+const int s_readableStreamInternalsLazyLoadStreamCodeLength = 2614;
 static const JSC::Intrinsic s_readableStreamInternalsLazyLoadStreamCodeIntrinsic = JSC::NoIntrinsic;
 const char* const s_readableStreamInternalsLazyLoadStreamCode =
     "(function (stream, autoAllocateChunkSize) {\n" \
@@ -2286,12 +2286,14 @@ const char* const s_readableStreamInternalsLazyLoadStreamCode =
     "          }),\n" \
     "          (err) => controller.error(err)\n" \
     "        );\n" \
-    "      } else if (result !== false) {\n" \
+    "      } else if (typeof result === 'number') {\n" \
     "        if (view && view.byteLength === result) {\n" \
     "          controller.byobRequest.respondWithNewView(view);\n" \
     "        } else {\n" \
     "          controller.byobRequest.respond(result);\n" \
     "        }\n" \
+    "      } else if (result.constructor === @Uint8Array) {\n" \
+    "        controller.enqueue(result);\n" \
     "      }\n" \
     "\n" \
     "      if (closer[0] || result === false) {\n" \
@@ -2317,6 +2319,7 @@ const char* const s_readableStreamInternalsLazyLoadStreamCode =
     "\n" \
     "      pull_(controller) {\n" \
     "        closer[0] = false;\n" \
+    "\n" \
     "        var result;\n" \
     "\n" \
     "        const view = controller.byobRequest.view;\n" \
diff --git a/src/bun.js/builtins/js/ReadableStreamInternals.js b/src/bun.js/builtins/js/ReadableStreamInternals.js
index 3d14535ca..067d10366 100644
--- a/src/bun.js/builtins/js/ReadableStreamInternals.js
+++ b/src/bun.js/builtins/js/ReadableStreamInternals.js
@@ -1864,12 +1864,14 @@ function lazyLoadStream(stream, autoAllocateChunkSize) {
           }),
           (err) => controller.error(err)
         );
-      } else if (result !== false) {
+      } else if (typeof result === 'number') {
         if (view && view.byteLength === result) {
           controller.byobRequest.respondWithNewView(view);
         } else {
           controller.byobRequest.respond(result);
         }
+      } else if (result.constructor === @Uint8Array) {
+        controller.enqueue(result);
       }
 
       if (closer[0] || result === false) {
@@ -1895,6 +1897,7 @@ function lazyLoadStream(stream, autoAllocateChunkSize) {
 
       pull_(controller) {
         closer[0] = false;
+
         var result;
 
         const view = controller.byobRequest.view;
diff --git a/src/bun.js/event_loop.zig b/src/bun.js/event_loop.zig
index 747bf01e0..e6b6477a3 100644
--- a/src/bun.js/event_loop.zig
+++ b/src/bun.js/event_loop.zig
@@ -462,16 +462,14 @@ pub const Poller = struct {
         switch (ptr.tag()) {
             @field(Pollable.Tag, "FileBlobLoader") => {
                 var loader = ptr.as(FileBlobLoader);
-                loop.active -= 1;
-                loop.num_polls -= 1;
+                loader.poll_ref.deactivate(loop);
 
                 loader.onPoll(@bitCast(i64, kqueue_event.data), kqueue_event.flags);
             },
             @field(Pollable.Tag, "Subprocess") => {
                 var loader = ptr.as(JSC.Subprocess);
 
-                loop.num_polls -= 1;
-                loop.active -= 1;
+                loader.poll_ref.deactivate(loop);
 
                 // kqueue sends the same notification multiple times in the same tick potentially
                 // so we have to dedupe it
@@ -479,9 +477,7 @@ pub const Poller = struct {
             },
             @field(Pollable.Tag, "FileSink") => {
                 var loader = ptr.as(JSC.WebCore.FileSink);
-
-                loop.num_polls -= 1;
-                loop.active -= 1;
+                loader.poll_ref.deactivate(loop);
 
                 loader.onPoll(0, 0);
             },
@@ -499,16 +495,13 @@ pub const Poller = struct {
         switch (ptr.tag()) {
             @field(Pollable.Tag, "FileBlobLoader") => {
                 var loader = ptr.as(FileBlobLoader);
-                loop.active -= 1;
-                loop.num_polls -= 1;
+                loader.poll_ref.deactivate(loop);
 
                 loader.onPoll(0, 0);
             },
             @field(Pollable.Tag, "Subprocess") => {
                 var loader = ptr.as(JSC.Subprocess);
-
-                loop.num_polls -= 1;
-                loop.active -= 1;
+                loader.poll_ref.deactivate(loop);
 
                 // kqueue sends the same notification multiple times in the same tick potentially
                 // so we have to dedupe it
@@ -516,9 +509,7 @@ pub const Poller = struct {
             },
             @field(Pollable.Tag, "FileSink") => {
                 var loader = ptr.as(JSC.WebCore.FileSink);
-
-                loop.num_polls -= 1;
-                loop.active -= 1;
+                loader.poll_ref.deactivate(loop);
 
                 loader.onPoll(0, 0);
             },
@@ -570,8 +561,7 @@ pub const Poller = struct {
                 return errno;
             }
 
-            this.loop.?.num_polls += 1;
-            this.loop.?.active += 1;
+            ctx.poll_ref.activate(this.loop.?);
 
             return JSC.Maybe(void).success;
         } else if (comptime Environment.isMac) {
@@ -635,8 +625,7 @@ pub const Poller = struct {
             const errno = std.c.getErrno(rc);
 
             if (errno == .SUCCESS) {
-                this.loop.?.num_polls += 1;
-                this.loop.?.active += 1;
+                ctx.poll_ref.activate(this.loop.?);
                 return JSC.Maybe(void).success;
             }
 
@@ -668,6 +657,8 @@ pub const Poller = struct {
                 return errno;
             }
 
+            ctx.poll_ref.deactivate(this.loop.?);
+
             return JSC.Maybe(void).success;
         } else if (comptime Environment.isMac) {
             var changelist = std.mem.zeroes([2]std.os.system.kevent64_s);
@@ -730,6 +721,7 @@ pub const Poller = struct {
             const errno = std.c.getErrno(rc);
 
             if (errno == .SUCCESS) {
+                ctx.poll_ref.deactivate(this.loop.?);
                 return JSC.Maybe(void).success;
             }
 
diff --git a/src/bun.js/webcore/response.zig b/src/bun.js/webcore/response.zig
index 64dfe3dc7..84721653d 100644
--- a/src/bun.js/webcore/response.zig
+++ b/src/bun.js/webcore/response.zig
@@ -565,7 +565,7 @@ pub const Fetch = struct {
 
             var ref = this.ref;
             const promise_value = ref.get();
-            defer ref.destroy(globalThis);
+            defer ref.destroy();
 
             if (promise_value.isEmptyOrUndefinedOrNull()) {
                 this.clearData();
@@ -4084,7 +4084,6 @@ pub const Body = struct {
 
         pub const empty = Value{ .Empty = .{} };
 
-
         pub fn toReadableStream(this: *Value, globalThis: *JSGlobalObject) JSValue {
             JSC.markBinding();
 
@@ -4489,7 +4488,6 @@ pub const Body = struct {
         if (body.value == .Blob)
             std.debug.assert(body.value.Blob.allocator == null); // owned by Body
 
-
         return body;
     }
 };
@@ -4712,7 +4710,7 @@ pub const Request = struct {
 
                     if (urlOrObject.fastGet(globalThis, .body)) |body_| {
                         if (Body.Value.fromJS(globalThis, body_)) |body| {
-                            request.body = body.value;
+                            request.body = body;
                         } else {
                             return null;
                         }
@@ -4729,7 +4727,7 @@ pub const Request = struct {
 
                 if (arguments[1].fastGet(globalThis, .body)) |body_| {
                     if (Body.Value.fromJS(globalThis, body_)) |body| {
-                        request.body = body.value;
+                        request.body = body;
                     } else {
                         return null;
                     }
diff --git a/src/bun.js/webcore/streams.zig b/src/bun.js/webcore/streams.zig
index 681ff6172..07e792fe0 100644
--- a/src/bun.js/webcore/streams.zig
+++ b/src/bun.js/webcore/streams.zig
@@ -948,7 +948,7 @@ pub const FileSink = struct {
     next: ?Sink = null,
     auto_close: bool = false,
     auto_truncate: bool = false,
-    opened_fd: JSC.Node.FileDescriptor = std.math.maxInt(JSC.Node.FileDescriptor),
+    fd: JSC.Node.FileDescriptor = std.math.maxInt(JSC.Node.FileDescriptor),
     mode: JSC.Node.Mode = 0,
     chunk_size: usize = 0,
     pending: StreamResult.Writable.Pending = StreamResult.Writable.Pending{
@@ -963,6 +963,9 @@ pub const FileSink = struct {
 
     prevent_process_exit: bool = false,
     reachable_from_js: bool = true,
+    poll_ref: JSC.PollRef = .{},
+
+    pub usingnamespace NewReadyWatcher(@This(), .write, ready);
 
     pub fn prepare(this: *FileSink, input_path: PathOrFileDescriptor, mode: JSC.Node.Mode) JSC.Node.Maybe(void) {
         var file_buf: [std.fs.MAX_PATH_BYTES]u8 = undefined;
@@ -987,7 +990,9 @@ pub const FileSink = struct {
         };
 
         this.mode = stat.mode;
-        this.opened_fd = fd;
+        this.fd = fd;
+
+        this.auto_truncate = this.auto_truncate and (std.os.S.ISREG(this.mode));
 
         return .{ .result = {} };
     }
@@ -998,9 +1003,9 @@ pub const FileSink = struct {
     }
 
     pub fn start(this: *FileSink, stream_start: StreamStart) JSC.Node.Maybe(void) {
-        if (this.opened_fd != std.math.maxInt(JSC.Node.FileDescriptor)) {
-            _ = JSC.Node.Syscall.close(this.opened_fd);
-            this.opened_fd = std.math.maxInt(JSC.Node.FileDescriptor);
+        if (this.fd != std.math.maxInt(JSC.Node.FileDescriptor)) {
+            _ = JSC.Node.Syscall.close(this.fd);
+            this.fd = std.math.maxInt(JSC.Node.FileDescriptor);
         }
 
         this.done = false;
@@ -1034,16 +1039,17 @@ pub const FileSink = struct {
     }
 
     pub fn flush(this: *FileSink) StreamResult.Writable {
-        std.debug.assert(this.opened_fd != std.math.maxInt(JSC.Node.FileDescriptor));
+        std.debug.assert(this.fd != std.math.maxInt(JSC.Node.FileDescriptor));
 
         var total: usize = this.written;
         const initial = total;
         defer this.written = total;
-        const fd = this.opened_fd;
+        const fd = this.fd;
         var remain = this.buffer.slice();
         remain = remain[@minimum(this.head, remain.len)..];
+        const initial_remain = remain;
         defer {
-            std.debug.assert(total - initial == @ptrToInt(remain.ptr) - @ptrToInt(this.buffer.ptr));
+            std.debug.assert(total - initial == @ptrToInt(remain.ptr) - @ptrToInt(initial_remain.ptr));
 
             if (remain.len == 0) {
                 this.head = 0;
@@ -1060,7 +1066,7 @@ pub const FileSink = struct {
 
                 switch (res.err.getErrno()) {
                     retry => {
-                        this.watch();
+                        this.watch(fd);
                         return .{
                             .pending = &this.pending,
                         };
@@ -1085,11 +1091,11 @@ pub const FileSink = struct {
         if (this.requested_end) {
             this.done = true;
             if (this.auto_truncate)
-                std.os.ftruncate(this.opened_fd, total) catch {};
+                std.os.ftruncate(this.fd, total) catch {};
 
             if (this.auto_close) {
-                _ = JSC.Node.Syscall.close(this.opened_fd);
-                this.opened_fd = std.math.maxInt(JSC.Node.FileDescriptor);
+                _ = JSC.Node.Syscall.close(this.fd);
+                this.fd = std.math.maxInt(JSC.Node.FileDescriptor);
             }
         }
         this.pending.run();
@@ -1097,6 +1103,9 @@ pub const FileSink = struct {
     }
 
     pub fn flushFromJS(this: *FileSink, globalThis: *JSGlobalObject, _: bool) JSC.Node.Maybe(JSValue) {
+        if (this.isPending()) {
+            return .{ .result = JSC.JSValue.jsUndefined() };
+        }
         const result = this.flush();
 
         if (result == .err) {
@@ -1109,9 +1118,11 @@ pub const FileSink = struct {
     }
 
     fn cleanup(this: *FileSink) void {
-        if (this.opened_fd != std.math.maxInt(JSC.Node.FileDescriptor)) {
-            _ = JSC.Node.Syscall.close(this.opened_fd);
-            this.opened_fd = std.math.maxInt(JSC.Node.FileDescriptor);
+        this.unwatch(this.fd);
+
+        if (this.fd != std.math.maxInt(JSC.Node.FileDescriptor)) {
+            _ = JSC.Node.Syscall.close(this.fd);
+            this.fd = std.math.maxInt(JSC.Node.FileDescriptor);
         }
 
         if (this.buffer.cap > 0) {
@@ -1120,6 +1131,9 @@ pub const FileSink = struct {
             this.done = true;
             this.head = 0;
         }
+
+        this.pending.result = .done;
+        this.pending.run();
     }
 
     pub fn finalize(this: *FileSink) void {
@@ -1151,25 +1165,11 @@ pub const FileSink = struct {
         };
     }
 
-    pub fn watch(this: *FileSink) void {
-        std.debug.assert(this.opened_fd != std.math.maxInt(JSC.Node.FileDescriptor));
-        _ = JSC.VirtualMachine.vm.poller.watch(this.opened_fd, .write, FileSink, this);
-        this.scheduled_count += 1;
-    }
-
-    pub fn unwatch(this: *FileSink) void {
-        std.debug.assert(this.scheduled_count > 0);
-        std.debug.assert(this.opened_fd != std.math.maxInt(JSC.Node.FileDescriptor));
-        _ = JSC.VirtualMachine.vm.poller.unwatch(this.opened_fd, .write, FileSink, this);
-        this.scheduled_count -= 1;
-    }
-
     pub fn toJS(this: *FileSink, globalThis: *JSGlobalObject) JSValue {
         return JSSink.createObject(globalThis, this);
     }
 
-    pub fn onPoll(this: *FileSink, _: i64, _: u16) void {
-        this.scheduled_count -= 1;
+    pub fn ready(this: *FileSink, _: i64) void {
         _ = this.flush();
     }
 
@@ -1246,7 +1246,7 @@ pub const FileSink = struct {
     }
 
     fn isPending(this: *const FileSink) bool {
-        return this.scheduled_count > 0;
+        return this.poll_ref.status == .active;
     }
 
     pub fn end(this: *FileSink, err: ?Syscall.Error) JSC.Node.Maybe(void) {
@@ -2531,13 +2531,16 @@ pub fn ReadableStreamSource(
             }
 
             pub fn processResult(globalThis: *JSGlobalObject, callFrame: *JSC.CallFrame, result: StreamResult) JSC.JSValue {
+                const arguments = callFrame.arguments(2);
+                var array = arguments.ptr[1].asObjectRef();
+
                 switch (result) {
                     .err => |err| {
                         globalThis.vm().throwError(globalThis, err.toJSC(globalThis));
                         return JSValue.jsUndefined();
                     },
                     .temporary_and_done, .owned_and_done, .into_array_and_done => {
-                        JSC.C.JSObjectSetPropertyAtIndex(globalThis.ref(), callFrame.argument(2).asObjectRef(), 0, JSValue.jsBoolean(true).asObjectRef(), null);
+                        JSC.C.JSObjectSetPropertyAtIndex(globalThis.ref(), array, 0, JSValue.jsBoolean(true).asObjectRef(), null);
                         return result.toJS(globalThis);
                     },
                     else => return result.toJS(globalThis),
@@ -2691,7 +2694,6 @@ pub const ByteStream = struct {
     has_received_last_chunk: bool = false,
     pending: StreamResult.Pending = StreamResult.Pending{
         .frame = undefined,
-        .used = false,
         .result = .{ .done = {} },
     },
     done: bool = false,
@@ -2794,7 +2796,7 @@ pub const ByteStream = struct {
 
         const chunk = stream.slice();
 
-        if (!this.pending.used) {
+        if (this.pending.state != .pending) {
             std.debug.assert(this.buffer.items.len == 0);
             var to_copy = this.pending_buffer[0..@minimum(chunk.len, this.pending_buffer.len)];
             const pending_buffer_len = this.pending_buffer.len;
@@ -2941,15 +2943,13 @@ pub const ByteStream = struct {
         this.done = true;
         if (this.pending_value) |ref| {
             this.pending_value = null;
-            ref.destroy(undefined);
+            ref.destroy();
         }
 
         if (view != .zero) {
             this.pending_buffer = &.{};
             this.pending.result = .{ .done = {} };
-            if (!this.pending.used) {
-                resume this.pending.frame;
-            }
+            this.pending.run();
         }
     }
 
@@ -2959,17 +2959,14 @@ pub const ByteStream = struct {
 
         if (this.pending_value) |ref| {
             this.pending_value = null;
-            ref.destroy(undefined);
+            ref.destroy();
         }
         if (!this.done) {
             this.done = true;
 
             this.pending_buffer = &.{};
             this.pending.result = .{ .done = {} };
-
-            if (!this.pending.used) {
-                resume this.pending.frame;
-            }
+            this.pending.run();
         }
 
         bun.default_allocator.destroy(this.parent());
@@ -2980,7 +2977,7 @@ pub const ByteStream = struct {
 
 pub const FileBlobLoader = struct {
     buf: []u8 = &[_]u8{},
-    protected_view: JSC.JSValue = JSC.JSValue.zero,
+    view: JSC.Strong = .{},
     fd: JSC.Node.FileDescriptor = 0,
     auto_close: bool = false,
     loop: *JSC.EventLoop = undefined,
@@ -3000,6 +2997,14 @@ pub const FileBlobLoader = struct {
     concurrent: Concurrent = Concurrent{},
     input_tag: StreamResult.Tag = StreamResult.Tag.done,
     started: bool = false,
+    stored_global_this_: ?*JSC.JSGlobalObject = null,
+    poll_ref: JSC.PollRef = .{},
+
+    pub usingnamespace NewReadyWatcher(@This(), .read, ready);
+
+    pub inline fn globalThis(this: *FileBlobLoader) *JSC.JSGlobalObject {
+        return this.stored_global_this_ orelse @fieldParentPtr(Source, "context", this).globalThis;
+    }
 
     const FileReader = @This();
 
@@ -3017,15 +3022,6 @@ pub const FileBlobLoader = struct {
         };
     }
 
-    pub fn watch(this: *FileReader) ?JSC.Node.Syscall.Error {
-        switch (JSC.VirtualMachine.vm.poller.watch(this.fd, .read, FileBlobLoader, this)) {
-            .err => |err| return err,
-            else => {},
-        }
-        this.scheduled_count += 1;
-        return null;
-    }
-
     const Concurrent = struct {
         read: Blob.SizeType = 0,
         task: NetworkThread.Task = .{ .callback = Concurrent.taskCallback },
@@ -3126,11 +3122,10 @@ pub const FileBlobLoader = struct {
 
         pub fn onJSThread(task_ctx: *anyopaque) void {
             var this: *FileBlobLoader = bun.cast(*FileBlobLoader, task_ctx);
-            const protected_view = this.protected_view;
-            defer protected_view.unprotect();
-            this.protected_view = JSC.JSValue.zero;
+            const view = this.view.get().?;
+            defer this.view.clear();
 
-            if (this.finalized and this.scheduled_count == 0) {
+            if (this.finalized and this.scheduled_count > 0) {
                 this.pending.run();
                 this.scheduled_count -= 1;
 
@@ -3154,7 +3149,7 @@ pub const FileBlobLoader = struct {
                 return;
             }
 
-            this.pending.result = this.handleReadChunk(@as(usize, this.concurrent.read), protected_view);
+            this.pending.result = this.handleReadChunk(@as(usize, this.concurrent.read), view, false, this.buf);
             this.pending.run();
             this.scheduled_count -= 1;
             if (this.pending.result.isDone()) {
@@ -3309,8 +3304,7 @@ pub const FileBlobLoader = struct {
                 return .{ .done = {} };
             },
             run_on_different_thread_size...std.math.maxInt(@TypeOf(chunk_size)) => {
-                this.protected_view = view;
-                this.protected_view.protect();
+                this.view.set(this.globalThis(), view);
                 // should never be reached
                 this.pending.result = .{
                     .err = Syscall.Error.todo,
@@ -3324,7 +3318,7 @@ pub const FileBlobLoader = struct {
             else => {},
         }
 
-        return this.read(buffer, view);
+        return this.read(buffer, view, null);
     }
 
     fn maybeAutoClose(this: *FileBlobLoader) void {
@@ -3334,7 +3328,7 @@ pub const FileBlobLoader = struct {
         }
     }
 
-    fn handleReadChunk(this: *FileBlobLoader, result: usize, view: JSC.JSValue) StreamResult {
+    fn handleReadChunk(this: *FileBlobLoader, result: usize, view: JSC.JSValue, owned: bool, buf: []u8) StreamResult {
         std.debug.assert(this.started);
 
         this.total_read += @intCast(Blob.SizeType, result);
@@ -3355,9 +3349,17 @@ pub const FileBlobLoader = struct {
         const has_more = remaining > 0;
 
         if (!has_more) {
+            if (owned) {
+                return .{ .owned_and_done = bun.ByteList.init(buf) };
+            }
+
             return .{ .into_array_and_done = .{ .len = @truncate(Blob.SizeType, result), .value = view } };
         }
 
+        if (owned) {
+            return .{ .owned = bun.ByteList.init(buf) };
+        }
+
         return .{ .into_array = .{ .len = @truncate(Blob.SizeType, result), .value = view } };
     }
 
@@ -3365,27 +3367,68 @@ pub const FileBlobLoader = struct {
         this: *FileBlobLoader,
         read_buf: []u8,
         view: JSC.JSValue,
+        /// provided via kqueue(), only on macOS
+        available_to_read: ?c_int,
     ) StreamResult {
         std.debug.assert(this.started);
+        std.debug.assert(read_buf.len > 0);
+
+        var buf_to_use = read_buf;
+        var free_buffer_on_error: bool = false;
+
+        // if it's a pipe, we really don't know what to expect what the max size will be
+        // if the pipe is sending us WAY bigger data than what we can fit in the buffer
+        // we allocate a new buffer of up to 4 MB
+        if (std.os.S.ISFIFO(this.mode)) {
+            outer: {
+                var len: c_int = available_to_read orelse 0;
+
+                // macOS FIONREAD doesn't seem to work here
+                // but we can get this information from the kqueue callback so we don't need to
+                if (len == 0) {
+                    const FIONREAD = if (Environment.isLinux) std.os.FIONREAD else bun.C.FIONREAD;
+                    const rc: c_int = std.c.ioctl(this.fd, FIONREAD, &len);
+                    if (rc != 0) {
+                        len = 0;
+                    }
+                }
 
-        const rc =
-            Syscall.read(this.fd, read_buf);
+                if (len > read_buf.len * 10 and read_buf.len < std.mem.page_size) {
+                    // then we need to allocate a buffer
+                    // to read into
+                    // this
+                    buf_to_use = bun.default_allocator.alloc(
+                        u8,
+                        @intCast(
+                            usize,
+                            @minimum(
+                                len,
+                                1024 * 1024 * 4,
+                            ),
+                        ),
+                    ) catch break :outer;
+                    free_buffer_on_error = true;
+                }
+            }
+        }
+
+        const rc = Syscall.read(this.fd, buf_to_use);
 
         switch (rc) {
             .err => |err| {
-                const retry =
-                    std.os.E.AGAIN;
+                const retry = std.os.E.AGAIN;
 
                 switch (err.getErrno()) {
                     retry => {
-                        this.protected_view = view;
-                        this.protected_view.protect();
-                        this.buf = read_buf;
-                        if (this.watch()) |watch_fail| {
-                            this.finalize();
-                            return .{ .err = watch_fail };
+                        if (free_buffer_on_error) {
+                            bun.default_allocator.free(buf_to_use);
+                            buf_to_use = read_buf;
                         }
 
+                        this.view.set(this.globalThis(), view);
+                        this.buf = read_buf;
+                        this.watch(this.fd);
+
                         return .{
                             .pending = &this.pending,
                         };
@@ -3401,19 +3444,28 @@ pub const FileBlobLoader = struct {
                 return .{ .err = sys };
             },
             .result => |result| {
-                return this.handleReadChunk(result, view);
+                if (result == 0 and free_buffer_on_error) {
+                    bun.default_allocator.free(buf_to_use);
+                    buf_to_use = read_buf;
+
+                    return this.handleReadChunk(result, view, true, buf_to_use);
+                } else if (free_buffer_on_error) {
+                    this.view.clear();
+                    this.buf = &.{};
+                    return this.handleReadChunk(result, view, true, buf_to_use);
+                }
+
+                return this.handleReadChunk(result, view, false, buf_to_use);
             },
         }
     }
 
     /// Called from Poller
-    pub fn onPoll(this: *FileBlobLoader, sizeOrOffset: i64, _: u16) void {
+    pub fn ready(this: *FileBlobLoader, sizeOrOffset: i64) void {
         std.debug.assert(this.started);
-        this.scheduled_count -= 1;
-        const protected_view = this.protected_view;
-        defer protected_view.unprotect();
 
-        this.protected_view = JSValue.zero;
+        const view = this.view.get() orelse .zero;
+        defer this.view.clear();
 
         var available_to_read: usize = std.math.maxInt(usize);
         if (comptime Environment.isMac) {
@@ -3446,35 +3498,29 @@ pub const FileBlobLoader = struct {
             this.buf.len = @minimum(this.buf.len, available_to_read);
         }
 
-        this.pending.result = this.read(this.buf, protected_view);
+        this.pending.result = this.read(
+            this.buf,
+            view,
+            if (available_to_read == std.math.maxInt(usize))
+                null
+            else
+                @truncate(c_int, @intCast(isize, available_to_read)),
+        );
         this.pending.run();
     }
 
-    pub fn unwatch(this: *FileBlobLoader) void {
-        std.debug.assert(this.scheduled_count > 0);
-        std.debug.assert(this.fd != std.math.maxInt(JSC.Node.FileDescriptor));
-        _ = JSC.VirtualMachine.vm.poller.unwatch(this.fd, .read, FileBlobLoader, this);
-        this.scheduled_count -= 1;
-    }
-
     pub fn finalize(this: *FileBlobLoader) void {
         if (this.finalized)
             return;
-        this.finalized = true;
 
-        if (this.scheduled_count > 0) {
-            this.unwatch();
-        }
+        this.unwatch(this.fd);
+        this.finalized = true;
 
         this.pending.result = .{ .done = {} };
         this.pending.run();
 
-        if (this.protected_view != .zero) {
-            this.protected_view.unprotect();
-            this.protected_view = .zero;
-
-            this.buf = &.{};
-        }
+        this.view.deinit();
+        this.buf = &.{};
 
         this.maybeAutoClose();
 
@@ -3483,7 +3529,6 @@ pub const FileBlobLoader = struct {
 
     pub fn onCancel(this: *FileBlobLoader) void {
         this.cancelled = true;
-
         this.deinit();
     }
 
@@ -3501,6 +3546,33 @@ pub const FileBlobLoader = struct {
     pub const Source = ReadableStreamSource(@This(), "FileBlobLoader", onStart, onPullInto, onCancel, deinit);
 };
 
+pub fn NewReadyWatcher(
+    comptime Context: type,
+    comptime flag_: JSC.Poller.Flag,
+    comptime onReady: anytype,
+) type {
+    return struct {
+        const flag = flag_;
+        const ready = onReady;
+
+        const Watcher = @This();
+
+        pub fn onPoll(this: *Context, sizeOrOffset: i64, _: u16) void {
+            ready(this, sizeOrOffset);
+        }
+
+        pub fn unwatch(this: *Context, fd: JSC.Node.FileDescriptor) void {
+            std.debug.assert(fd != std.math.maxInt(JSC.Node.FileDescriptor));
+            _ = JSC.VirtualMachine.vm.poller.unwatch(fd, flag, Context, this);
+        }
+
+        pub fn watch(this: *Context, fd: JSC.Node.FileDescriptor) void {
+            std.debug.assert(fd != std.math.maxInt(JSC.Node.FileDescriptor));
+            _ = JSC.VirtualMachine.vm.poller.watch(fd, flag, Context, this);
+        }
+    };
+}
+
 // pub const HTTPRequest = RequestBodyStreamer(false);
 // pub const HTTPSRequest = RequestBodyStreamer(true);
 // pub fn ResponseBodyStreamer(comptime is_ssl: bool) type {
diff --git a/src/darwin_c.zig b/src/darwin_c.zig
index 45e6f88d3..94d7f12ea 100644
--- a/src/darwin_c.zig
+++ b/src/darwin_c.zig
@@ -578,3 +578,130 @@ pub fn get_release(buf: []u8) []const u8 {
 
     return std.mem.span(std.meta.assumeSentinel(buf.ptr, 0));
 }
+
+const IO_CTL_RELATED = struct {
+    pub const IOCPARM_MASK = @as(c_int, 0x1fff);
+    pub inline fn IOCPARM_LEN(x: anytype) @TypeOf((x >> @as(c_int, 16)) & IOCPARM_MASK) {
+        return (x >> @as(c_int, 16)) & IOCPARM_MASK;
+    }
+    pub inline fn IOCBASECMD(x: anytype) @TypeOf(x & ~(IOCPARM_MASK << @as(c_int, 16))) {
+        return x & ~(IOCPARM_MASK << @as(c_int, 16));
+    }
+    pub inline fn IOCGROUP(x: anytype) @TypeOf((x >> @as(c_int, 8)) & @as(c_int, 0xff)) {
+        return (x >> @as(c_int, 8)) & @as(c_int, 0xff);
+    }
+    pub const IOCPARM_MAX = IOCPARM_MASK + @as(c_int, 1);
+    pub const IOC_VOID = @import("std").zig.c_translation.cast(u32, @import("std").zig.c_translation.promoteIntLiteral(c_int, 0x20000000, .hexadecimal));
+    pub const IOC_OUT = @import("std").zig.c_translation.cast(u32, @import("std").zig.c_translation.promoteIntLiteral(c_int, 0x40000000, .hexadecimal));
+    pub const IOC_IN = @import("std").zig.c_translation.cast(u32, @import("std").zig.c_translation.promoteIntLiteral(c_int, 0x80000000, .hexadecimal));
+    pub const IOC_INOUT = IOC_IN | IOC_OUT;
+    pub const IOC_DIRMASK = @import("std").zig.c_translation.cast(u32, @import("std").zig.c_translation.promoteIntLiteral(c_int, 0xe0000000, .hexadecimal));
+    pub inline fn _IOC(inout: anytype, group: anytype, num: anytype, len: anytype) @TypeOf(((inout | ((len & IOCPARM_MASK) << @as(c_int, 16))) | (group << @as(c_int, 8))) | num) {
+        return ((inout | ((len & IOCPARM_MASK) << @as(c_int, 16))) | (group << @as(c_int, 8))) | num;
+    }
+    pub inline fn _IO(g: anytype, n: anytype) @TypeOf(_IOC(IOC_VOID, g, n, @as(c_int, 0))) {
+        return _IOC(IOC_VOID, g, n, @as(c_int, 0));
+    }
+    pub inline fn _IOR(g: anytype, n: anytype, t: anytype) @TypeOf(_IOC(IOC_OUT, g, n, @import("std").zig.c_translation.sizeof(t))) {
+        _ = t;
+        return _IOC(IOC_OUT, g, n, @import("std").zig.c_translation.sizeof(t));
+    }
+    pub inline fn _IOW(g: anytype, n: anytype, t: anytype) @TypeOf(_IOC(IOC_IN, g, n, @import("std").zig.c_translation.sizeof(t))) {
+        _ = t;
+        return _IOC(IOC_IN, g, n, @import("std").zig.c_translation.sizeof(t));
+    }
+    pub inline fn _IOWR(g: anytype, n: anytype, t: anytype) @TypeOf(_IOC(IOC_INOUT, g, n, @import("std").zig.c_translation.sizeof(t))) {
+        _ = t;
+        return _IOC(IOC_INOUT, g, n, @import("std").zig.c_translation.sizeof(t));
+    }
+    pub const TIOCMODG = _IOR('t', @as(c_int, 3), c_int);
+    pub const TIOCMODS = _IOW('t', @as(c_int, 4), c_int);
+    pub const TIOCM_LE = @as(c_int, 0o001);
+    pub const TIOCM_DTR = @as(c_int, 0o002);
+    pub const TIOCM_RTS = @as(c_int, 0o004);
+    pub const TIOCM_ST = @as(c_int, 0o010);
+    pub const TIOCM_SR = @as(c_int, 0o020);
+    pub const TIOCM_CTS = @as(c_int, 0o040);
+    pub const TIOCM_CAR = @as(c_int, 0o100);
+    pub const TIOCM_CD = TIOCM_CAR;
+    pub const TIOCM_RNG = @as(c_int, 0o200);
+    pub const TIOCM_RI = TIOCM_RNG;
+    pub const TIOCM_DSR = @as(c_int, 0o400);
+    pub const TIOCEXCL = _IO('t', @as(c_int, 13));
+    pub const TIOCNXCL = _IO('t', @as(c_int, 14));
+    pub const TIOCFLUSH = _IOW('t', @as(c_int, 16), c_int);
+    pub const TIOCGETD = _IOR('t', @as(c_int, 26), c_int);
+    pub const TIOCSETD = _IOW('t', @as(c_int, 27), c_int);
+    pub const TIOCIXON = _IO('t', @as(c_int, 129));
+    pub const TIOCIXOFF = _IO('t', @as(c_int, 128));
+    pub const TIOCSBRK = _IO('t', @as(c_int, 123));
+    pub const TIOCCBRK = _IO('t', @as(c_int, 122));
+    pub const TIOCSDTR = _IO('t', @as(c_int, 121));
+    pub const TIOCCDTR = _IO('t', @as(c_int, 120));
+    pub const TIOCGPGRP = _IOR('t', @as(c_int, 119), c_int);
+    pub const TIOCSPGRP = _IOW('t', @as(c_int, 118), c_int);
+    pub const TIOCOUTQ = _IOR('t', @as(c_int, 115), c_int);
+    pub const TIOCSTI = _IOW('t', @as(c_int, 114), u8);
+    pub const TIOCNOTTY = _IO('t', @as(c_int, 113));
+    pub const TIOCPKT = _IOW('t', @as(c_int, 112), c_int);
+    pub const TIOCPKT_DATA = @as(c_int, 0x00);
+    pub const TIOCPKT_FLUSHREAD = @as(c_int, 0x01);
+    pub const TIOCPKT_FLUSHWRITE = @as(c_int, 0x02);
+    pub const TIOCPKT_STOP = @as(c_int, 0x04);
+    pub const TIOCPKT_START = @as(c_int, 0x08);
+    pub const TIOCPKT_NOSTOP = @as(c_int, 0x10);
+    pub const TIOCPKT_DOSTOP = @as(c_int, 0x20);
+    pub const TIOCPKT_IOCTL = @as(c_int, 0x40);
+    pub const TIOCSTOP = _IO('t', @as(c_int, 111));
+    pub const TIOCSTART = _IO('t', @as(c_int, 110));
+    pub const TIOCMSET = _IOW('t', @as(c_int, 109), c_int);
+    pub const TIOCMBIS = _IOW('t', @as(c_int, 108), c_int);
+    pub const TIOCMBIC = _IOW('t', @as(c_int, 107), c_int);
+    pub const TIOCMGET = _IOR('t', @as(c_int, 106), c_int);
+    // pub const TIOCGWINSZ = _IOR('t', @as(c_int, 104), struct_winsize);
+    // pub const TIOCSWINSZ = _IOW('t', @as(c_int, 103), struct_winsize);
+    pub const TIOCUCNTL = _IOW('t', @as(c_int, 102), c_int);
+    pub const TIOCSTAT = _IO('t', @as(c_int, 101));
+    pub inline fn UIOCCMD(n: anytype) @TypeOf(_IO('u', n)) {
+        return _IO('u', n);
+    }
+    pub const TIOCSCONS = _IO('t', @as(c_int, 99));
+    pub const TIOCCONS = _IOW('t', @as(c_int, 98), c_int);
+    pub const TIOCSCTTY = _IO('t', @as(c_int, 97));
+    pub const TIOCEXT = _IOW('t', @as(c_int, 96), c_int);
+    pub const TIOCSIG = _IO('t', @as(c_int, 95));
+    pub const TIOCDRAIN = _IO('t', @as(c_int, 94));
+    pub const TIOCMSDTRWAIT = _IOW('t', @as(c_int, 91), c_int);
+    pub const TIOCMGDTRWAIT = _IOR('t', @as(c_int, 90), c_int);
+    pub const TIOCSDRAINWAIT = _IOW('t', @as(c_int, 87), c_int);
+    pub const TIOCGDRAINWAIT = _IOR('t', @as(c_int, 86), c_int);
+    pub const TIOCDSIMICROCODE = _IO('t', @as(c_int, 85));
+    pub const TIOCPTYGRANT = _IO('t', @as(c_int, 84));
+    pub const TIOCPTYGNAME = _IOC(IOC_OUT, 't', @as(c_int, 83), @as(c_int, 128));
+    pub const TIOCPTYUNLK = _IO('t', @as(c_int, 82));
+    pub const TTYDISC = @as(c_int, 0);
+    pub const TABLDISC = @as(c_int, 3);
+    pub const SLIPDISC = @as(c_int, 4);
+    pub const PPPDISC = @as(c_int, 5);
+    // pub const TIOCGSIZE = TIOCGWINSZ;
+    // pub const TIOCSSIZE = TIOCSWINSZ;
+    pub const FIOCLEX = _IO('f', @as(c_int, 1));
+    pub const FIONCLEX = _IO('f', @as(c_int, 2));
+    pub const FIONREAD = _IOR('f', @as(c_int, 127), c_int);
+    pub const FIONBIO = _IOW('f', @as(c_int, 126), c_int);
+    pub const FIOASYNC = _IOW('f', @as(c_int, 125), c_int);
+    pub const FIOSETOWN = _IOW('f', @as(c_int, 124), c_int);
+    pub const FIOGETOWN = _IOR('f', @as(c_int, 123), c_int);
+    pub const FIODTYPE = _IOR('f', @as(c_int, 122), c_int);
+    pub const SIOCSHIWAT = _IOW('s', @as(c_int, 0), c_int);
+    pub const SIOCGHIWAT = _IOR('s', @as(c_int, 1), c_int);
+    pub const SIOCSLOWAT = _IOW('s', @as(c_int, 2), c_int);
+    pub const SIOCGLOWAT = _IOR('s', @as(c_int, 3), c_int);
+    pub const SIOCATMARK = _IOR('s', @as(c_int, 7), c_int);
+    pub const SIOCSPGRP = _IOW('s', @as(c_int, 8), c_int);
+    pub const SIOCGPGRP = _IOR('s', @as(c_int, 9), c_int);
+    // pub const SIOCSETVLAN = SIOCSIFVLAN;
+    // pub const SIOCGETVLAN = SIOCGIFVLAN;
+};
+
+pub usingnamespace IO_CTL_RELATED;
diff --git a/src/napi/napi.zig b/src/napi/napi.zig
index 6b191167f..570ac9cc8 100644
--- a/src/napi/napi.zig
+++ b/src/napi/napi.zig
@@ -33,9 +33,9 @@ pub const Ref = opaque {
         return napi_get_reference_value_internal(ref);
     }
 
-    pub fn destroy(ref: *Ref, globalThis: *JSC.JSGlobalObject) void {
+    pub fn destroy(ref: *Ref) void {
         JSC.markBinding();
-        std.debug.assert(napi_delete_reference(globalThis, ref) == .ok);
+        napi_delete_reference_internal(ref);
     }
 
     pub fn set(this: *Ref, value: JSC.JSValue) void {
@@ -43,6 +43,7 @@ pub const Ref = opaque {
         napi_set_ref(this, value);
     }
 
+    extern fn napi_delete_reference_internal(ref: *Ref) void;
     extern fn napi_set_ref(ref: *Ref, value: JSC.JSValue) void;
 };
 pub const napi_handle_scope = napi_env;