fix(runtime): improve IPC reliability + organization pass on that code (#6475)

* dfghj

* Handle messages that did not finish

* tidy

* ok

* a

* Merge remote-tracking branch 'origin/main' into dave/ipc-fixes

* test failures

---------

Co-authored-by: Jarred Sumner <jarred@jarredsumner.com>

15 files changed, 266 insertions, 98 deletions

diff --git a/packages/bun-types/tests/test.test-d.ts b/packages/bun-types/tests/test.test-d.ts
index 4aed19fb0..b385c3051 100644
--- a/packages/bun-types/tests/test.test-d.ts
+++ b/packages/bun-types/tests/test.test-d.ts
@@ -45,6 +45,7 @@ describe("bun:test", () => {
   test("expect()", () => {
     expect(1).toBe(1);
     expect(1).not.toBe(2);
+    // @ts-expect-error
     expect({ a: 1 }).toEqual({ a: 1, b: undefined });
     expect({ a: 1 }).toStrictEqual({ a: 1 });
     expect(new Set()).toHaveProperty("size");
diff --git a/src/baby_list.zig b/src/baby_list.zig
index 5881f2c50..a691ea1c7 100644
--- a/src/baby_list.zig
+++ b/src/baby_list.zig
@@ -279,5 +279,13 @@ pub fn BabyList(comptime Type: type) type {
 
             return this.len - initial;
         }
+
+        pub fn writeTypeAsBytesAssumeCapacity(this: *@This(), comptime Int: type, int: Int) void {
+            if (comptime Type != u8)
+                @compileError("Unsupported for type " ++ @typeName(Type));
+            std.debug.assert(this.cap >= this.len + @sizeOf(Int));
+            @as([*]align(1) Int, @ptrCast(this.ptr[this.len .. this.len + @sizeOf(Int)]))[0] = int;
+            this.len += @sizeOf(Int);
+        }
     };
 }
diff --git a/src/bun.js/WebKit b/src/bun.js/WebKit
-Subproject e1aa0a58e282b53fc20503d6e7ec93c621bc557
+Subproject 1a49a1f94bf42ab4f8c6b11d7bbbb21e491d2d6
diff --git a/src/bun.js/api/bun/subprocess.zig b/src/bun.js/api/bun/subprocess.zig
index 1bed014f4..f6d86d91a 100644
--- a/src/bun.js/api/bun/subprocess.zig
+++ b/src/bun.js/api/bun/subprocess.zig
@@ -62,11 +62,9 @@ pub const Subprocess = struct {
     is_sync: bool = false,
     this_jsvalue: JSC.JSValue = .zero,
 
-    ipc: IPCMode,
-    // this is only ever accessed when `ipc` is not `none`
-    ipc_socket: IPC.Socket = undefined,
+    ipc_mode: IPCMode,
     ipc_callback: JSC.Strong = .{},
-    ipc_buffer: bun.ByteList,
+    ipc: IPC.IPCData,
 
     has_pending_unref: bool = false,
     pub const SignalCode = bun.SignalCode;
@@ -83,7 +81,7 @@ pub const Subprocess = struct {
 
     pub fn updateHasPendingActivityFlag(this: *Subprocess) void {
         @fence(.SeqCst);
-        this.has_pending_activity.store(this.waitpid_err == null and this.exit_code == null and this.ipc == .none and this.has_pending_unref, .SeqCst);
+        this.has_pending_activity.store(this.waitpid_err == null and this.exit_code == null and this.ipc_mode == .none and this.has_pending_unref, .SeqCst);
     }
 
     pub fn hasPendingActivity(this: *Subprocess) callconv(.C) bool {
@@ -93,7 +91,7 @@ pub const Subprocess = struct {
 
     pub fn updateHasPendingActivity(this: *Subprocess) void {
         @fence(.Release);
-        this.has_pending_activity.store(this.waitpid_err == null and this.exit_code == null and this.ipc == .none and this.has_pending_unref, .Release);
+        this.has_pending_activity.store(this.waitpid_err == null and this.exit_code == null and this.ipc_mode == .none and this.has_pending_unref, .Release);
     }
 
     pub fn ref(this: *Subprocess) void {
@@ -428,7 +426,7 @@ pub const Subprocess = struct {
     }
 
     pub fn doSend(this: *Subprocess, global: *JSC.JSGlobalObject, callFrame: *JSC.CallFrame) callconv(.C) JSValue {
-        if (this.ipc == .none) {
+        if (this.ipc_mode == .none) {
             global.throw("Subprocess.send() can only be used if an IPC channel is open.", .{});
             return .zero;
         }
@@ -440,20 +438,16 @@ pub const Subprocess = struct {
 
         const value = callFrame.argument(0);
 
-        const success = IPC.serializeJSValueForSubprocess(
-            global,
-            value,
-            this.ipc_socket.fd(),
-        );
+        const success = this.ipc.serializeAndSend(global, value);
         if (!success) return .zero;
 
         return JSC.JSValue.jsUndefined();
     }
 
     pub fn disconnect(this: *Subprocess) void {
-        if (this.ipc == .none) return;
-        this.ipc_socket.close(0, null);
-        this.ipc = .none;
+        if (this.ipc_mode == .none) return;
+        this.ipc.socket.close(0, null);
+        this.ipc_mode = .none;
     }
 
     pub fn getPid(
@@ -1541,15 +1535,15 @@ pub const Subprocess = struct {
             .stderr = Readable.init(stdio[bun.STDERR_FD], stderr_pipe[0], jsc_vm.allocator, default_max_buffer_size),
             .on_exit_callback = if (on_exit_callback != .zero) JSC.Strong.create(on_exit_callback, globalThis) else .{},
             .is_sync = is_sync,
-            .ipc = ipc_mode,
+            .ipc_mode = ipc_mode,
             // will be assigned in the block below
-            .ipc_socket = socket,
-            .ipc_buffer = bun.ByteList{},
+            .ipc = .{ .socket = socket },
             .ipc_callback = if (ipc_callback != .zero) JSC.Strong.create(ipc_callback, globalThis) else undefined,
         };
         if (ipc_mode != .none) {
             var ptr = socket.ext(*Subprocess);
             ptr.?.* = subprocess;
+            subprocess.ipc.writeVersionPacket();
         }
 
         if (subprocess.stdin == .pipe) {
@@ -2073,7 +2067,7 @@ pub const Subprocess = struct {
 
     pub fn handleIPCClose(this: *Subprocess, _: IPC.Socket) void {
         // uSocket is already freed so calling .close() on the socket can segfault
-        this.ipc = .none;
+        this.ipc_mode = .none;
         this.updateHasPendingActivity();
     }
 
diff --git a/src/bun.js/bindings/Serialization.cpp b/src/bun.js/bindings/Serialization.cpp
index 89937ebbb..fdaf2ab75 100644
--- a/src/bun.js/bindings/Serialization.cpp
+++ b/src/bun.js/bindings/Serialization.cpp
@@ -8,9 +8,15 @@
 using namespace JSC;
 using namespace WebCore;
 
-/// This is used for Bun.spawn() IPC because otherwise we would have to copy the data once to get it to zig, then write it.
-/// Returns `true` on success, `false` on failure + throws a JS error.
-extern "C" bool Bun__serializeJSValueForSubprocess(JSGlobalObject* globalObject, EncodedJSValue encodedValue, int fd)
+// Must be synced with bindings.zig's JSValue.SerializedScriptValue.External
+struct SerializedValueSlice {
+    uint8_t* bytes;
+    size_t size;
+    WebCore::SerializedScriptValue* value;
+};
+
+/// Returns a "slice" that also contains a pointer to the SerializedScriptValue. Must be freed by the caller
+extern "C" SerializedValueSlice Bun__serializeJSValue(JSGlobalObject* globalObject, EncodedJSValue encodedValue)
 {
     JSValue value = JSValue::decode(encodedValue);
 
@@ -25,19 +31,23 @@ extern "C" bool Bun__serializeJSValueForSubprocess(JSGlobalObject* globalObject,
     if (serialized.hasException()) {
         WebCore::propagateException(*globalObject, scope,
             serialized.releaseException());
-        RELEASE_AND_RETURN(scope, false);
+        RELEASE_AND_RETURN(scope, { 0 });
     }
 
     auto serializedValue = serialized.releaseReturnValue();
-    auto bytes = serializedValue.ptr()->wireBytes();
 
-    uint8_t id = 2; // IPCMessageType.SerializedMessage
-    write(fd, &id, sizeof(uint8_t));
-    uint32_t size = bytes.size();
-    write(fd, &size, sizeof(uint32_t));
-    write(fd, bytes.data(), size);
+    auto bytes = serializedValue->wireBytes();
 
-    RELEASE_AND_RETURN(scope, true);
+    return {
+        bytes.data(),
+        bytes.size(),
+        &serializedValue.leakRef(),
+    };
+}
+
+extern "C" void Bun__SerializedScriptSlice__free(SerializedScriptValue* value)
+{
+    delete value;
 }
 
 extern "C" EncodedJSValue Bun__JSValue__deserialize(JSGlobalObject* globalObject, const uint8_t* bytes, size_t size)
diff --git a/src/bun.js/bindings/bindings.zig b/src/bun.js/bindings/bindings.zig
index 1ab1ca603..98e91c1ca 100644
--- a/src/bun.js/bindings/bindings.zig
+++ b/src/bun.js/bindings/bindings.zig
@@ -4935,6 +4935,34 @@ pub const JSValue = enum(JSValueReprInt) {
     pub inline fn deserialize(bytes: []const u8, global: *JSGlobalObject) JSValue {
         return Bun__JSValue__deserialize(global, bytes.ptr, @intCast(bytes.len));
     }
+
+    extern fn Bun__serializeJSValue(global: *JSC.JSGlobalObject, value: JSValue) SerializedScriptValue.External;
+    extern fn Bun__SerializedScriptSlice__free(*anyopaque) void;
+
+    pub const SerializedScriptValue = struct {
+        data: []const u8,
+        handle: *anyopaque,
+
+        const External = extern struct {
+            bytes: ?[*]const u8,
+            size: isize,
+            handle: ?*anyopaque,
+        };
+
+        pub inline fn deinit(self: @This()) void {
+            Bun__SerializedScriptSlice__free(self.handle);
+        }
+    };
+
+    /// Throws a JS exception and returns null if the serialization fails, otherwise returns a SerializedScriptValue.
+    /// Must be freed when you are done with the bytes.
+    pub inline fn serialize(this: JSValue, global: *JSGlobalObject) ?SerializedScriptValue {
+        const value = Bun__serializeJSValue(global, this);
+        return if (value.bytes) |bytes|
+            .{ .data = bytes[0..@intCast(value.size)], .handle = value.handle.? }
+        else
+            null;
+    }
 };
 
 extern "c" fn AsyncContextFrame__withAsyncContextIfNeeded(global: *JSGlobalObject, callback: JSValue) JSValue;
diff --git a/src/bun.js/bindings/exports.zig b/src/bun.js/bindings/exports.zig
index 784a6289c..617530ec6 100644
--- a/src/bun.js/bindings/exports.zig
+++ b/src/bun.js/bindings/exports.zig
@@ -969,7 +969,7 @@ pub const ZigConsoleClient = struct {
                 Output.prettyFmt("<r><red>Assertion failed<r>\n", true)
             else
                 "Assertion failed\n";
-            console.error_writer.unbuffered_writer.writeAll(text) catch unreachable;
+            console.error_writer.unbuffered_writer.writeAll(text) catch {};
             return;
         }
 
@@ -1024,7 +1024,7 @@ pub const ZigConsoleClient = struct {
             _ = console.writer.write("\n") catch 0;
             console.writer.flush() catch {};
         } else if (message_type != .Trace)
-            writer.writeAll("undefined\n") catch unreachable;
+            writer.writeAll("undefined\n") catch {};
 
         if (message_type == .Trace) {
             writeTrace(Writer, writer, global);
@@ -1046,14 +1046,14 @@ pub const ZigConsoleClient = struct {
                 writer,
                 exception.stack,
                 true,
-            ) catch unreachable
+            ) catch {}
         else
             JS.VirtualMachine.printStackTrace(
                 Writer,
                 writer,
                 exception.stack,
                 false,
-            ) catch unreachable;
+            ) catch {};
     }
 
     pub const FormatOptions = struct {
@@ -1102,7 +1102,7 @@ pub const ZigConsoleClient = struct {
             if (tag.tag == .String) {
                 if (options.enable_colors) {
                     if (level == .Error) {
-                        unbuffered_writer.writeAll(comptime Output.prettyFmt("<r><red>", true)) catch unreachable;
+                        unbuffered_writer.writeAll(comptime Output.prettyFmt("<r><red>", true)) catch {};
                     }
                     fmt.format(
                         tag,
@@ -1113,7 +1113,7 @@ pub const ZigConsoleClient = struct {
                         true,
                     );
                     if (level == .Error) {
-                        unbuffered_writer.writeAll(comptime Output.prettyFmt("<r>", true)) catch unreachable;
+                        unbuffered_writer.writeAll(comptime Output.prettyFmt("<r>", true)) catch {};
                     }
                 } else {
                     fmt.format(
@@ -1175,7 +1175,7 @@ pub const ZigConsoleClient = struct {
         var any = false;
         if (options.enable_colors) {
             if (level == .Error) {
-                writer.writeAll(comptime Output.prettyFmt("<r><red>", true)) catch unreachable;
+                writer.writeAll(comptime Output.prettyFmt("<r><red>", true)) catch {};
             }
             while (true) {
                 if (any) {
@@ -1197,7 +1197,7 @@ pub const ZigConsoleClient = struct {
                 fmt.remaining_values = fmt.remaining_values[1..];
             }
             if (level == .Error) {
-                writer.writeAll(comptime Output.prettyFmt("<r>", true)) catch unreachable;
+                writer.writeAll(comptime Output.prettyFmt("<r>", true)) catch {};
             }
         } else {
             while (true) {
diff --git a/src/bun.js/bindings/webcore/WebCoreTypedArrayController.cpp b/src/bun.js/bindings/webcore/WebCoreTypedArrayController.cpp
index ca8c79aef..5c8f4fbea 100644
--- a/src/bun.js/bindings/webcore/WebCoreTypedArrayController.cpp
+++ b/src/bun.js/bindings/webcore/WebCoreTypedArrayController.cpp
@@ -35,6 +35,16 @@
 
 #include "JavaScriptCore/JSArrayBuffer.h"
 
+extern "C" Zig::GlobalObject* Bun__getDefaultGlobal();
+static inline WebCore::JSDOMGlobalObject* getDefaultGlobal(JSC::JSGlobalObject* lexicalGlobalObject)
+{
+    if (auto* global = jsDynamicCast<WebCore::JSDOMGlobalObject*>(lexicalGlobalObject)) {
+        return global;
+    }
+
+    return Bun__getDefaultGlobal();
+}
+
 namespace WebCore {
 
 WebCoreTypedArrayController::WebCoreTypedArrayController(bool allowAtomicsWait)
@@ -46,7 +56,7 @@ WebCoreTypedArrayController::~WebCoreTypedArrayController() = default;
 
 JSC::JSArrayBuffer* WebCoreTypedArrayController::toJS(JSC::JSGlobalObject* lexicalGlobalObject, JSC::JSGlobalObject* globalObject, JSC::ArrayBuffer* buffer)
 {
-    return JSC::jsCast<JSC::JSArrayBuffer*>(WebCore::toJS(lexicalGlobalObject, JSC::jsCast<JSDOMGlobalObject*>(globalObject), buffer));
+    return JSC::jsCast<JSC::JSArrayBuffer*>(WebCore::toJS(lexicalGlobalObject, getDefaultGlobal(globalObject), buffer));
 }
 
 void WebCoreTypedArrayController::registerWrapper(JSC::JSGlobalObject* globalObject, JSC::ArrayBuffer* native, JSC::JSArrayBuffer* wrapper)
diff --git a/src/bun.js/ipc.zig b/src/bun.js/ipc.zig
index 8f0e0f406..a0742a0c4 100644
--- a/src/bun.js/ipc.zig
+++ b/src/bun.js/ipc.zig
@@ -35,6 +35,11 @@ pub const IPCMessageType = enum(u8) {
     _,
 };
 
+pub const IPCBuffer = struct {
+    list: bun.ByteList = .{},
+    cursor: u32 = 0,
+};
+
 /// Given potentially unfinished buffer `data`, attempt to decode and process a message from it.
 /// Returns `NotEnoughBytes` if there werent enough bytes
 /// Returns `InvalidFormat` if the message was invalid, probably close the socket in this case
@@ -89,12 +94,74 @@ pub fn decodeIPCMessage(
 
 pub const Socket = uws.NewSocketHandler(false);
 
+pub const IPCData = struct {
+    socket: Socket,
+    incoming: bun.ByteList = .{}, // Maybe we should use IPCBuffer here as well
+    outgoing: IPCBuffer = .{},
+
+    has_written_version: if (Environment.allow_assert) u1 else u0 = 0,
+
+    pub fn writeVersionPacket(this: *IPCData) void {
+        if (Environment.allow_assert) {
+            std.debug.assert(this.has_written_version == 0);
+        }
+        const VersionPacket = extern struct {
+            type: IPCMessageType align(1) = .Version,
+            version: u32 align(1) = ipcVersion,
+        };
+        const bytes = comptime std.mem.asBytes(&VersionPacket{});
+        const n = this.socket.write(bytes, false);
+        if (n != bytes.len) {
+            var list = this.outgoing.list.listManaged(bun.default_allocator);
+            list.appendSlice(bytes) catch @panic("OOM");
+        }
+        if (Environment.allow_assert) {
+            this.has_written_version = 1;
+        }
+    }
+
+    pub fn serializeAndSend(ipc_data: *IPCData, globalThis: *JSGlobalObject, value: JSValue) bool {
+        if (Environment.allow_assert) {
+            std.debug.assert(ipc_data.has_written_version == 1);
+        }
+
+        const serialized = value.serialize(globalThis) orelse return false;
+        defer serialized.deinit();
+
+        const size: u32 = @intCast(serialized.data.len);
+
+        const payload_length: usize = @sizeOf(IPCMessageType) + @sizeOf(u32) + size;
+
+        ipc_data.outgoing.list.ensureUnusedCapacity(bun.default_allocator, payload_length) catch @panic("OOM");
+        const start_offset = ipc_data.outgoing.list.len;
+
+        ipc_data.outgoing.list.writeTypeAsBytesAssumeCapacity(u8, @intFromEnum(IPCMessageType.SerializedMessage));
+        ipc_data.outgoing.list.writeTypeAsBytesAssumeCapacity(u32, size);
+        ipc_data.outgoing.list.appendSliceAssumeCapacity(serialized.data);
+
+        std.debug.assert(ipc_data.outgoing.list.len == start_offset + payload_length);
+
+        if (start_offset == 0) {
+            std.debug.assert(ipc_data.outgoing.cursor == 0);
+
+            const n = ipc_data.socket.write(ipc_data.outgoing.list.ptr[start_offset..payload_length], false);
+            if (n == payload_length) {
+                ipc_data.outgoing.list.len = 0;
+            } else if (n > 0) {
+                ipc_data.outgoing.cursor = @intCast(n);
+            }
+        }
+
+        return true;
+    }
+};
+
 /// This type is shared between VirtualMachine and Subprocess for their respective IPC handlers
 ///
 /// `Context` must be a struct that implements this interface:
 /// struct {
 ///     globalThis: ?*JSGlobalObject,
-///     ipc_buffer: bun.ByteList,
+///     ipc: IPCData,
 ///
 ///     fn handleIPCMessage(*Context, DecodedIPCMessage) void
 ///     fn handleIPCClose(*Context, Socket) void
@@ -102,18 +169,18 @@ pub const Socket = uws.NewSocketHandler(false);
 pub fn NewIPCHandler(comptime Context: type) type {
     return struct {
         pub fn onOpen(
-            _: *Context,
-            socket: Socket,
+            _: *anyopaque,
+            _: Socket,
         ) void {
-            // Write the version message
-            const Data = extern struct {
-                type: IPCMessageType align(1) = .Version,
-                version: u32 align(1) = ipcVersion,
-            };
-            const data: []const u8 = comptime @as([@sizeOf(Data)]u8, @bitCast(Data{}))[0..];
-            _ = socket.write(data, false);
-            socket.flush();
+            // it is NOT safe to use the first argument here because it has not been initialized yet.
+            // ideally we would call .ipc.writeVersionPacket() here, and we need that to handle the
+            // theoretical write failure, but since the .ipc.outgoing buffer isn't available, that
+            // data has nowhere to go.
+            //
+            // therefore, initializers of IPC handlers need to call .ipc.writeVersionPacket() themselves
+            // this is covered by an assertion.
         }
+
         pub fn onClose(
             this: *Context,
             socket: Socket,
@@ -124,7 +191,7 @@ pub fn NewIPCHandler(comptime Context: type) type {
             log("onClose\n", .{});
             this.handleIPCClose(socket);
         }
-        // extern fn getpid() i32;
+
         pub fn onData(
             this: *Context,
             socket: Socket,
@@ -133,10 +200,6 @@ pub fn NewIPCHandler(comptime Context: type) type {
             var data = data_;
             log("onData {}", .{std.fmt.fmtSliceHexLower(data)});
 
-            // if (comptime Context == bun.JSC.VirtualMachine.IPCInstance) {
-            //     logDataOnly("{d} -> '{}'", .{ getpid(), std.fmt.fmtSliceHexLower(data) });
-            // }
-
             // In the VirtualMachine case, `globalThis` is an optional, in case
             // the vm is freed before the socket closes.
             var globalThis = switch (@typeInfo(@TypeOf(this.globalThis))) {
@@ -154,11 +217,11 @@ pub fn NewIPCHandler(comptime Context: type) type {
 
             // Decode the message with just the temporary buffer, and if that
             // fails (not enough bytes) then we allocate to .ipc_buffer
-            if (this.ipc_buffer.len == 0) {
+            if (this.ipc.incoming.len == 0) {
                 while (true) {
                     const result = decodeIPCMessage(data, globalThis) catch |e| switch (e) {
                         error.NotEnoughBytes => {
-                            _ = this.ipc_buffer.write(bun.default_allocator, data) catch @panic("OOM");
+                            _ = this.ipc.incoming.write(bun.default_allocator, data) catch @panic("OOM");
                             log("hit NotEnoughBytes", .{});
                             return;
                         },
@@ -180,15 +243,15 @@ pub fn NewIPCHandler(comptime Context: type) type {
                 }
             }
 
-            _ = this.ipc_buffer.write(bun.default_allocator, data) catch @panic("OOM");
+            _ = this.ipc.incoming.write(bun.default_allocator, data) catch @panic("OOM");
 
-            var slice = this.ipc_buffer.slice();
+            var slice = this.ipc.incoming.slice();
             while (true) {
                 const result = decodeIPCMessage(slice, globalThis) catch |e| switch (e) {
                     error.NotEnoughBytes => {
                         // copy the remaining bytes to the start of the buffer
-                        std.mem.copyForwards(u8, this.ipc_buffer.ptr[0..slice.len], slice);
-                        this.ipc_buffer.len = @truncate(slice.len);
+                        bun.copy(u8, this.ipc.incoming.ptr[0..slice.len], slice);
+                        this.ipc.incoming.len = @truncate(slice.len);
                         log("hit NotEnoughBytes2", .{});
                         return;
                     },
@@ -206,34 +269,47 @@ pub fn NewIPCHandler(comptime Context: type) type {
                     slice = slice[result.bytes_consumed..];
                 } else {
                     // clear the buffer
-                    this.ipc_buffer.len = 0;
+                    this.ipc.incoming.len = 0;
                     return;
                 }
             }
         }
 
         pub fn onWritable(
-            _: *Context,
-            _: Socket,
-        ) void {}
+            context: *Context,
+            socket: Socket,
+        ) void {
+            const to_write = context.ipc.outgoing.list.ptr[context.ipc.outgoing.cursor..context.ipc.outgoing.list.len];
+            if (to_write.len == 0) {
+                context.ipc.outgoing.cursor = 0;
+                context.ipc.outgoing.list.len = 0;
+                return;
+            }
+            const n = socket.write(to_write, false);
+            if (n == to_write.len) {
+                context.ipc.outgoing.cursor = 0;
+                context.ipc.outgoing.list.len = 0;
+            } else if (n > 0) {
+                context.ipc.outgoing.cursor += @intCast(n);
+            }
+        }
+
         pub fn onTimeout(
             _: *Context,
             _: Socket,
         ) void {}
+
         pub fn onConnectError(
-            _: *Context,
+            _: *anyopaque,
             _: Socket,
             _: c_int,
-        ) void {}
+        ) void {
+            // context has not been initialized
+        }
+
         pub fn onEnd(
             _: *Context,
             _: Socket,
         ) void {}
     };
 }
-
-/// This is used for Bun.spawn() IPC because otherwise we would have to copy the data once to get it to zig, then write it.
-/// Returns `true` on success, `false` on failure + throws a JS error.
-extern fn Bun__serializeJSValueForSubprocess(global: *JSC.JSGlobalObject, value: JSValue, fd: bun.FileDescriptor) bool;
-
-pub const serializeJSValueForSubprocess = Bun__serializeJSValueForSubprocess;
diff --git a/src/bun.js/javascript.zig b/src/bun.js/javascript.zig
index d1b87e34e..e93e65399 100644
--- a/src/bun.js/javascript.zig
+++ b/src/bun.js/javascript.zig
@@ -294,13 +294,8 @@ pub export fn Bun__Process__send(
         return .zero;
     }
     var vm = globalObject.bunVM();
-    if (vm.ipc) |ipc| {
-        const fd = ipc.socket.fd();
-        const success = IPC.serializeJSValueForSubprocess(
-            globalObject,
-            callFrame.argument(0),
-            fd,
-        );
+    if (vm.ipc) |ipc_instance| {
+        const success = ipc_instance.ipc.serializeAndSend(globalObject, callFrame.argument(0));
         return if (success) .undefined else .zero;
     } else {
         globalObject.throw("IPC Socket is no longer open.", .{});
@@ -373,6 +368,14 @@ pub export fn Bun__onDidAppendPlugin(jsc_vm: *VirtualMachine, globalObject: *JSG
     jsc_vm.bundler.linker.plugin_runner = &jsc_vm.plugin_runner.?;
 }
 
+// pub fn getGlobalExitCodeForPipeFailure() u8 {
+//     if (VirtualMachine.is_main_thread_vm) {
+//         return VirtualMachine.get().exit_handler.exit_code;
+//     }
+
+//     return 0;
+// }
+
 pub const ExitHandler = struct {
     exit_code: u8 = 0,
 
@@ -2803,9 +2806,8 @@ pub const VirtualMachine = struct {
 
     pub const IPCInstance = struct {
         globalThis: ?*JSGlobalObject,
-        socket: IPC.Socket,
         uws_context: *uws.SocketContext,
-        ipc_buffer: bun.ByteList,
+        ipc: IPC.IPCData,
 
         pub fn handleIPCMessage(
             this: *IPCInstance,
@@ -2855,13 +2857,13 @@ pub const VirtualMachine = struct {
         var instance = bun.default_allocator.create(IPCInstance) catch @panic("OOM");
         instance.* = .{
             .globalThis = this.global,
-            .socket = socket,
             .uws_context = context,
-            .ipc_buffer = bun.ByteList{},
+            .ipc = .{ .socket = socket },
         };
         var ptr = socket.ext(*IPCInstance);
         ptr.?.* = instance;
         this.ipc = instance;
+        instance.ipc.writeVersionPacket();
     }
     comptime {
         if (!JSC.is_bindgen)
diff --git a/src/bun.js/test/jest.zig b/src/bun.js/test/jest.zig
index 283f369ff..5490b3472 100644
--- a/src/bun.js/test/jest.zig
+++ b/src/bun.js/test/jest.zig
@@ -808,9 +808,17 @@ pub const DescribeScope = struct {
         return false;
     }
 
+    fn isWithinTodoScope(this: *const DescribeScope) bool {
+        if (this.tag == .todo) return true;
+        if (this.parent != null) return this.parent.?.isWithinTodoScope();
+        return false;
+    }
+
     pub fn shouldEvaluateScope(this: *const DescribeScope) bool {
-        if (this.isWithinSkipScope()) return false;
-        if (Jest.runner.?.only and this.isWithinOnlyScope()) return true;
+        if (this.tag == .skip or
+            this.tag == .todo) return false;
+        if (Jest.runner.?.only and this.tag == .only) return true;
+        if (this.parent != null) return this.parent.?.shouldEvaluateScope();
         return true;
     }
 
diff --git a/src/js/_codegen/client-js.ts b/src/js/_codegen/client-js.ts
index bd9ed63f4..4dfa6acf6 100644
--- a/src/js/_codegen/client-js.ts
+++ b/src/js/_codegen/client-js.ts
@@ -16,9 +16,10 @@ let $debug_log_enabled = ((env) => (
     .join("_")
     .toUpperCase()})
 ))(Bun.env);
+let $debug_pid_prefix = Bun.env.SHOW_PID === '1';
 let $debug_log = $debug_log_enabled ? (...args) => {
   // warn goes to stderr without colorizing
-  console.warn(Bun.enableANSIColors ? '\\x1b[90m[${publicName}]\\x1b[0m' : '[${publicName}]', ...args);
+  console.warn(($debug_pid_prefix ? \`[\${process.pid}] \` : '') + (Bun.enableANSIColors ? '\\x1b[90m[${publicName}]\\x1b[0m' : '[${publicName}]'), ...args);
 } : () => {};
 `;
 }
diff --git a/src/js/node/async_hooks.ts b/src/js/node/async_hooks.ts
index 83b313912..4a2fd1936 100644
--- a/src/js/node/async_hooks.ts
+++ b/src/js/node/async_hooks.ts
@@ -51,11 +51,13 @@ function assertValidAsyncContextArray(array: unknown): array is ReadonlyArray<an
 
 // Only run during debug
 function debugFormatContextValue(value: ReadonlyArray<any> | undefined) {
-  if (value === undefined) return "{}";
+  if (value === undefined) return "undefined";
   let str = "{\n";
   for (var i = 0; i < value.length; i += 2) {
-    str += `  ${value[i].__id__}: ${Bun.inspect(value[i + 1], { depth: 1, colors: Bun.enableANSIColors })}\n`;
+    str += `  ${value[i].__id__}: typeof = ${typeof value[i + 1]}\n`;
   }
+  str += "}";
+  return str;
 }
 
 function get(): ReadonlyArray<any> | undefined {
@@ -77,10 +79,12 @@ class AsyncLocalStorage {
 
     // In debug mode assign every AsyncLocalStorage a unique ID
     if (IS_BUN_DEVELOPMENT) {
-      (this as any).__id__ =
-        Math.random().toString(36).slice(2, 8) +
-        "@" +
-        require("node:path").basename(require("bun:jsc").callerSourceOrigin());
+      const uid = Math.random().toString(36).slice(2, 8);
+      const source = require("bun:jsc").callerSourceOrigin();
+
+      (this as any).__id__ = uid + "@" + require("node:path").basename(source);
+
+      $debug("new AsyncLocalStorage uid=", (this as any).__id__, source);
     }
   }
 
@@ -133,6 +137,7 @@ class AsyncLocalStorage {
   // This function is literred with $asserts to ensure that everything that
   // is assumed to be true is *actually* true.
   run(store_value, callback, ...args) {
+    $debug("run " + (this as any).__id__);
     var context = get() as any[]; // we make sure to .slice() before mutating
     var hasPrevious = false;
     var previous_value;
@@ -215,6 +220,7 @@ class AsyncLocalStorage {
   }
 
   getStore() {
+    $debug("getStore " + (this as any).__id__);
     var context = get();
     if (!context) return;
     var { length } = context;
diff --git a/src/js/node/child_process.js b/src/js/node/child_process.js
index c46a50bc0..adc479bba 100644
--- a/src/js/node/child_process.js
+++ b/src/js/node/child_process.js
@@ -1174,6 +1174,7 @@ class ChildProcess extends EventEmitter {
       env: env || process.env,
       detached: typeof detachedOption !== "undefined" ? !!detachedOption : false,
       onExit: (handle, exitCode, signalCode, err) => {
+        $debug("ChildProcess: onExit", exitCode, signalCode, err, this.pid);
         this.#handle = handle;
         this.pid = this.#handle.pid;
 
@@ -1189,6 +1190,8 @@ class ChildProcess extends EventEmitter {
     });
     this.pid = this.#handle.pid;
 
+    $debug("ChildProcess: spawn", this.pid, spawnargs);
+
     onSpawnNT(this);
 
     if (ipc) {
diff --git a/src/report.zig b/src/report.zig
index c78c5cb0d..75c63a7af 100644
--- a/src/report.zig
+++ b/src/report.zig
@@ -247,10 +247,21 @@ pub fn fatal(err_: ?anyerror, msg_: ?string) void {
 }
 
 var globalError_ranOnce = false;
+var error_return_trace: ?*std.builtin.StackTrace = null;
 
 export fn Bun__crashReportWrite(ctx: *CrashReportWriter, bytes_ptr: [*]const u8, len: usize) void {
-    if (len > 0)
+    if (error_return_trace) |trace| {
+        if (len > 0) {
+            ctx.print("{s}\n{}", .{ bytes_ptr[0..len], trace });
+        } else {
+            ctx.print("{}\n", .{trace});
+        }
+        return;
+    }
+
+    if (len > 0) {
         ctx.print("{s}\n", .{bytes_ptr[0..len]});
+    }
 }
 
 extern "C" fn Bun__crashReportDumpStackTrace(ctx: *anyopaque) void;
@@ -274,6 +285,10 @@ pub noinline fn handleCrash(signal: i32, addr: usize) void {
         .{ @errorName(name), bun.fmt.hexIntUpper(addr) },
     );
     printMetadata();
+    if (comptime Environment.isDebug) {
+        error_return_trace = @errorReturnTrace();
+    }
+
     if (comptime !@import("root").bun.JSC.is_bindgen) {
         std.mem.doNotOptimizeAway(&Bun__crashReportWrite);
         Bun__crashReportDumpStackTrace(&crash_report_writer);
@@ -290,10 +305,8 @@ pub noinline fn handleCrash(signal: i32, addr: usize) void {
     }
 
     crash_report_writer.file = null;
-    if (comptime Environment.isDebug) {
-        if (@errorReturnTrace()) |stack| {
-            std.debug.dumpStackTrace(stack.*);
-        }
+    if (error_return_trace) |trace| {
+        std.debug.dumpStackTrace(trace.*);
     }
 
     std.c._exit(128 + @as(u8, @truncate(@as(u8, @intCast(@max(signal, 0))))));
@@ -302,11 +315,19 @@ pub noinline fn handleCrash(signal: i32, addr: usize) void {
 pub noinline fn globalError(err: anyerror, trace_: @TypeOf(@errorReturnTrace())) noreturn {
     @setCold(true);
 
+    error_return_trace = trace_;
+
     if (@atomicRmw(bool, &globalError_ranOnce, .Xchg, true, .Monotonic)) {
         Global.exit(1);
     }
 
     switch (err) {
+        // error.BrokenPipe => {
+        //     if (comptime Environment.isNative) {
+        //         // if stdout/stderr was closed, we don't need to print anything
+        //         std.c._exit(bun.JSC.getGlobalExitCodeForPipeFailure());
+        //     }
+        // },
         error.SyntaxError => {
             Output.prettyError(
                 "\n<r><red>SyntaxError<r><d>:<r> An error occurred while parsing code",