Merge branch 'main' into jarred/split-subprocess

src/bun.js/api/bun/ResourceUsage.zig

@@ -0,0 +1,110 @@
+const ResourceUsage = @This();
+
+rusage: Rusage,
+
+pub fn create(rusage: *const Rusage, globalObject: *JSGlobalObject) JSValue {
+    const resource_usage = ResourceUsage{
+        .rusage = rusage.*,
+    };
+
+    var result = bun.default_allocator.create(ResourceUsage) catch {
+        return globalObject.throwOutOfMemoryValue();
+    };
+    result.* = resource_usage;
+    return result.toJS(globalObject);
+}
+
+pub fn getCPUTime(
+    this: *ResourceUsage,
+    globalObject: *JSGlobalObject,
+) JSValue {
+    var cpu = JSC.JSValue.createEmptyObjectWithNullPrototype(globalObject);
+    const rusage = this.rusage;
+
+    const usrTime = JSValue.fromTimevalNoTruncate(globalObject, rusage.utime.usec, rusage.utime.sec);
+    const sysTime = JSValue.fromTimevalNoTruncate(globalObject, rusage.stime.usec, rusage.stime.sec);
+
+    cpu.put(globalObject, JSC.ZigString.static("user"), usrTime);
+    cpu.put(globalObject, JSC.ZigString.static("system"), sysTime);
+    cpu.put(globalObject, JSC.ZigString.static("total"), JSValue.bigIntSum(globalObject, usrTime, sysTime));
+
+    return cpu;
+}
+
+pub fn getMaxRSS(
+    this: *ResourceUsage,
+    _: *JSGlobalObject,
+) JSValue {
+    return JSC.JSValue.jsNumber(this.rusage.maxrss);
+}
+
+pub fn getSharedMemorySize(
+    this: *ResourceUsage,
+    _: *JSGlobalObject,
+) JSValue {
+    return JSC.JSValue.jsNumber(this.rusage.ixrss);
+}
+
+pub fn getSwapCount(
+    this: *ResourceUsage,
+    _: *JSGlobalObject,
+) JSValue {
+    return JSC.JSValue.jsNumber(this.rusage.nswap);
+}
+
+pub fn getOps(
+    this: *ResourceUsage,
+    globalObject: *JSGlobalObject,
+) JSValue {
+    var ops = JSC.JSValue.createEmptyObjectWithNullPrototype(globalObject);
+    ops.put(globalObject, JSC.ZigString.static("in"), JSC.JSValue.jsNumber(this.rusage.inblock));
+    ops.put(globalObject, JSC.ZigString.static("out"), JSC.JSValue.jsNumber(this.rusage.oublock));
+    return ops;
+}
+
+pub fn getMessages(
+    this: *ResourceUsage,
+    globalObject: *JSGlobalObject,
+) JSValue {
+    var msgs = JSC.JSValue.createEmptyObjectWithNullPrototype(globalObject);
+    msgs.put(globalObject, JSC.ZigString.static("sent"), JSC.JSValue.jsNumber(this.rusage.msgsnd));
+    msgs.put(globalObject, JSC.ZigString.static("received"), JSC.JSValue.jsNumber(this.rusage.msgrcv));
+    return msgs;
+}
+
+pub fn getSignalCount(
+    this: *ResourceUsage,
+    _: *JSGlobalObject,
+) JSValue {
+    return JSC.JSValue.jsNumber(this.rusage.nsignals);
+}
+
+pub fn getContextSwitches(
+    this: *ResourceUsage,
+    globalObject: *JSGlobalObject,
+) JSValue {
+    var ctx = JSC.JSValue.createEmptyObjectWithNullPrototype(globalObject);
+    ctx.put(globalObject, JSC.ZigString.static("voluntary"), JSC.JSValue.jsNumber(this.rusage.nvcsw));
+    ctx.put(globalObject, JSC.ZigString.static("involuntary"), JSC.JSValue.jsNumber(this.rusage.nivcsw));
+    return ctx;
+}
+
+pub fn finalize(this: *ResourceUsage) callconv(.C) void {
+    bun.default_allocator.destroy(this);
+}
+
+pub const js = JSC.Codegen.JSResourceUsage;
+pub const toJS = ResourceUsage.js.toJS;
+pub const fromJS = ResourceUsage.js.fromJS;
+pub const fromJSDirect = ResourceUsage.js.fromJSDirect;
+
+const std = @import("std");
+const bun = @import("bun");
+const JSC = bun.JSC;
+
+const Rusage = bun.spawn.Rusage;
+const JSGlobalObject = JSC.JSGlobalObject;
+const JSValue = JSC.JSValue;
+const Subprocess = JSC.API.Subprocess;
+const Environment = bun.Environment;
+const PosixSpawn = bun.spawn;

src/bun.js/api/bun/subprocess.zig

@@ -71,103 +71,7 @@ pub const WaitThreadPoll = struct {
     poll_ref: Async.KeepAlive = .{},
 };
 
-pub inline fn assertStdioResult(result: StdioResult) void {
-    if (comptime Environment.allow_assert) {
-        if (Environment.isPosix) {
-            if (result) |fd| {
-                bun.assert(fd != bun.invalid_fd);
-            }
-        }
-    }
-}
-
-pub const ResourceUsage = struct {
-    pub const js = JSC.Codegen.JSResourceUsage;
-    pub const toJS = ResourceUsage.js.toJS;
-    pub const fromJS = ResourceUsage.js.fromJS;
-    pub const fromJSDirect = ResourceUsage.js.fromJSDirect;
-
-    rusage: Rusage,
-
-    pub fn getCPUTime(
-        this: *ResourceUsage,
-        globalObject: *JSGlobalObject,
-    ) JSValue {
-        var cpu = JSC.JSValue.createEmptyObjectWithNullPrototype(globalObject);
-        const rusage = this.rusage;
-
-        const usrTime = JSValue.fromTimevalNoTruncate(globalObject, rusage.utime.usec, rusage.utime.sec);
-        const sysTime = JSValue.fromTimevalNoTruncate(globalObject, rusage.stime.usec, rusage.stime.sec);
-
-        cpu.put(globalObject, JSC.ZigString.static("user"), usrTime);
-        cpu.put(globalObject, JSC.ZigString.static("system"), sysTime);
-        cpu.put(globalObject, JSC.ZigString.static("total"), JSValue.bigIntSum(globalObject, usrTime, sysTime));
-
-        return cpu;
-    }
-
-    pub fn getMaxRSS(
-        this: *ResourceUsage,
-        _: *JSGlobalObject,
-    ) JSValue {
-        return JSC.JSValue.jsNumber(this.rusage.maxrss);
-    }
-
-    pub fn getSharedMemorySize(
-        this: *ResourceUsage,
-        _: *JSGlobalObject,
-    ) JSValue {
-        return JSC.JSValue.jsNumber(this.rusage.ixrss);
-    }
-
-    pub fn getSwapCount(
-        this: *ResourceUsage,
-        _: *JSGlobalObject,
-    ) JSValue {
-        return JSC.JSValue.jsNumber(this.rusage.nswap);
-    }
-
-    pub fn getOps(
-        this: *ResourceUsage,
-        globalObject: *JSGlobalObject,
-    ) JSValue {
-        var ops = JSC.JSValue.createEmptyObjectWithNullPrototype(globalObject);
-        ops.put(globalObject, JSC.ZigString.static("in"), JSC.JSValue.jsNumber(this.rusage.inblock));
-        ops.put(globalObject, JSC.ZigString.static("out"), JSC.JSValue.jsNumber(this.rusage.oublock));
-        return ops;
-    }
-
-    pub fn getMessages(
-        this: *ResourceUsage,
-        globalObject: *JSGlobalObject,
-    ) JSValue {
-        var msgs = JSC.JSValue.createEmptyObjectWithNullPrototype(globalObject);
-        msgs.put(globalObject, JSC.ZigString.static("sent"), JSC.JSValue.jsNumber(this.rusage.msgsnd));
-        msgs.put(globalObject, JSC.ZigString.static("received"), JSC.JSValue.jsNumber(this.rusage.msgrcv));
-        return msgs;
-    }
-
-    pub fn getSignalCount(
-        this: *ResourceUsage,
-        _: *JSGlobalObject,
-    ) JSValue {
-        return JSC.JSValue.jsNumber(this.rusage.nsignals);
-    }
-
-    pub fn getContextSwitches(
-        this: *ResourceUsage,
-        globalObject: *JSGlobalObject,
-    ) JSValue {
-        var ctx = JSC.JSValue.createEmptyObjectWithNullPrototype(globalObject);
-        ctx.put(globalObject, JSC.ZigString.static("voluntary"), JSC.JSValue.jsNumber(this.rusage.nvcsw));
-        ctx.put(globalObject, JSC.ZigString.static("involuntary"), JSC.JSValue.jsNumber(this.rusage.nivcsw));
-        return ctx;
-    }
-
-    pub fn finalize(this: *ResourceUsage) callconv(.C) void {
-        bun.default_allocator.destroy(this);
-    }
-};
+pub const ResourceUsage = @import("./ResourceUsage.zig");
 
 pub fn appendEnvpFromJS(globalThis: *JSC.JSGlobalObject, object: *JSC.JSObject, envp: *std.ArrayList(?[*:0]const u8), PATH: *[]const u8) bun.JSError!void {
     var object_iter = try JSC.JSPropertyIterator(.{ .skip_empty_name = false, .include_value = true }).init(globalThis, object);
@@ -229,26 +133,19 @@ pub fn resourceUsage(
 }
 
 pub fn createResourceUsageObject(this: *Subprocess, globalObject: *JSGlobalObject) JSValue {
-    const pid_rusage = this.pid_rusage orelse brk: {
-        if (Environment.isWindows) {
-            if (this.process.poller == .uv) {
-                this.pid_rusage = PosixSpawn.process.uv_getrusage(&this.process.poller.uv);
-                break :brk this.pid_rusage.?;
+    return ResourceUsage.create(
+        this.pid_rusage orelse brk: {
+            if (Environment.isWindows) {
+                if (this.process.poller == .uv) {
+                    this.pid_rusage = PosixSpawn.process.uv_getrusage(&this.process.poller.uv);
+                    break :brk this.pid_rusage.?;
+                }
             }
-        }
 
-        return JSValue.jsUndefined();
-    };
-
-    const resource_usage = ResourceUsage{
-        .rusage = pid_rusage,
-    };
-
-    var result = bun.default_allocator.create(ResourceUsage) catch {
-        return globalObject.throwOutOfMemoryValue();
-    };
-    result.* = resource_usage;
-    return result.toJS(globalObject);
+            return JSValue.jsUndefined();
+        },
+        globalObject,
+    );
 }
 
 pub fn hasExited(this: *const Subprocess) bool {
@@ -378,184 +275,15 @@ pub fn constructor(globalObject: *JSC.JSGlobalObject, _: *JSC.CallFrame) bun.JSE
     return globalObject.throw("Cannot construct Subprocess", .{});
 }
 
-const Readable = union(enum) {
-    fd: bun.FileDescriptor,
-    memfd: bun.FileDescriptor,
-    pipe: *PipeReader,
-    inherit: void,
-    ignore: void,
-    closed: void,
-    /// Eventually we will implement Readables created from blobs and array buffers.
-    /// When we do that, `buffer` will be borrowed from those objects.
-    ///
-    /// When a buffered `pipe` finishes reading from its file descriptor,
-    /// the owning `Readable` will be convered into this variant and the pipe's
-    /// buffer will be taken as an owned `CowString`.
-    buffer: CowString,
-
-    pub fn memoryCost(this: *const Readable) usize {
-        return switch (this.*) {
-            .pipe => @sizeOf(PipeReader) + this.pipe.memoryCost(),
-            .buffer => this.buffer.length(),
-            else => 0,
-        };
-    }
-
-    pub fn hasPendingActivity(this: *const Readable) bool {
-        return switch (this.*) {
-            .pipe => this.pipe.hasPendingActivity(),
-            else => false,
-        };
-    }
-
-    pub fn ref(this: *Readable) void {
-        switch (this.*) {
-            .pipe => {
-                this.pipe.updateRef(true);
-            },
-            else => {},
-        }
-    }
-
-    pub fn unref(this: *Readable) void {
-        switch (this.*) {
-            .pipe => {
-                this.pipe.updateRef(false);
-            },
-            else => {},
-        }
-    }
-
-    pub fn init(stdio: Stdio, event_loop: *JSC.EventLoop, process: *Subprocess, result: StdioResult, allocator: std.mem.Allocator, max_size: ?*MaxBuf, is_sync: bool) Readable {
-        _ = allocator; // autofix
-        _ = is_sync; // autofix
-        assertStdioResult(result);
-
-        if (comptime Environment.isPosix) {
-            if (stdio == .pipe) {
-                _ = bun.sys.setNonblocking(result.?);
+pub inline fn assertStdioResult(result: StdioResult) void {
+    if (comptime Environment.allow_assert) {
+        if (Environment.isPosix) {
+            if (result) |fd| {
+                bun.assert(fd != bun.invalid_fd);
             }
         }
-
-        return switch (stdio) {
-            .inherit => Readable{ .inherit = {} },
-            .ignore, .ipc, .path => Readable{ .ignore = {} },
-            .fd => |fd| if (Environment.isPosix) Readable{ .fd = result.? } else Readable{ .fd = fd },
-            .memfd => if (Environment.isPosix) Readable{ .memfd = stdio.memfd } else Readable{ .ignore = {} },
-            .dup2 => |dup2| if (Environment.isPosix) Output.panic("TODO: implement dup2 support in Stdio readable", .{}) else Readable{ .fd = dup2.out.toFd() },
-            .pipe => Readable{ .pipe = PipeReader.create(event_loop, process, result, max_size) },
-            .array_buffer, .blob => Output.panic("TODO: implement ArrayBuffer & Blob support in Stdio readable", .{}),
-            .capture => Output.panic("TODO: implement capture support in Stdio readable", .{}),
-        };
     }
-
-    pub fn onClose(this: *Readable, _: ?bun.sys.Error) void {
-        this.* = .closed;
-    }
-
-    pub fn onReady(_: *Readable, _: ?JSC.WebCore.Blob.SizeType, _: ?JSC.WebCore.Blob.SizeType) void {}
-
-    pub fn onStart(_: *Readable) void {}
-
-    pub fn close(this: *Readable) void {
-        switch (this.*) {
-            .memfd => |fd| {
-                this.* = .{ .closed = {} };
-                fd.close();
-            },
-            .fd => |_| {
-                this.* = .{ .closed = {} };
-            },
-            .pipe => {
-                this.pipe.close();
-            },
-            else => {},
-        }
-    }
-
-    pub fn finalize(this: *Readable) void {
-        switch (this.*) {
-            .memfd => |fd| {
-                this.* = .{ .closed = {} };
-                fd.close();
-            },
-            .fd => {
-                this.* = .{ .closed = {} };
-            },
-            .pipe => |pipe| {
-                defer pipe.detach();
-                this.* = .{ .closed = {} };
-            },
-            .buffer => |*buf| {
-                buf.deinit(bun.default_allocator);
-            },
-            else => {},
-        }
-    }
-
-    pub fn toJS(this: *Readable, globalThis: *JSC.JSGlobalObject, exited: bool) JSValue {
-        _ = exited; // autofix
-        switch (this.*) {
-            // should only be reachable when the entire output is buffered.
-            .memfd => return this.toBufferedValue(globalThis) catch .zero,
-
-            .fd => |fd| {
-                return fd.toJS(globalThis);
-            },
-            .pipe => |pipe| {
-                defer pipe.detach();
-                this.* = .{ .closed = {} };
-                return pipe.toJS(globalThis);
-            },
-            .buffer => |*buffer| {
-                defer this.* = .{ .closed = {} };
-
-                if (buffer.length() == 0) {
-                    return JSC.WebCore.ReadableStream.empty(globalThis);
-                }
-
-                const own = buffer.takeSlice(bun.default_allocator) catch {
-                    globalThis.throwOutOfMemory() catch return .zero;
-                };
-                return JSC.WebCore.ReadableStream.fromOwnedSlice(globalThis, own, 0);
-            },
-            else => {
-                return JSValue.jsUndefined();
-            },
-        }
-    }
-
-    pub fn toBufferedValue(this: *Readable, globalThis: *JSC.JSGlobalObject) bun.JSError!JSValue {
-        switch (this.*) {
-            .fd => |fd| {
-                return fd.toJS(globalThis);
-            },
-            .memfd => |fd| {
-                if (comptime !Environment.isPosix) {
-                    Output.panic("memfd is only supported on Linux", .{});
-                }
-                this.* = .{ .closed = {} };
-                return JSC.ArrayBuffer.toJSBufferFromMemfd(fd, globalThis);
-            },
-            .pipe => |pipe| {
-                defer pipe.detach();
-                this.* = .{ .closed = {} };
-                return pipe.toBuffer(globalThis);
-            },
-            .buffer => |*buf| {
-                defer this.* = .{ .closed = {} };
-                const own = buf.takeSlice(bun.default_allocator) catch {
-                    return globalThis.throwOutOfMemory();
-                };
-
-                return JSC.MarkedArrayBuffer.fromBytes(own, bun.default_allocator, .Uint8Array).toNodeBuffer(globalThis);
-            },
-            else => {
-                return JSValue.jsUndefined();
-            },
-        }
-    }
-};
+}
 
 pub fn getStderr(
     this: *Subprocess,
@@ -845,637 +573,11 @@ pub const Source = union(enum) {
     }
 };
 
+pub const PipeReader = @import("./subprocess/PipeReader.zig");
+pub const NewStaticPipeWriter = @import("./subprocess/PipeWriter.zig").NewStaticPipeWriter;
 pub const StaticPipeWriter = NewStaticPipeWriter(Subprocess);
-
-pub fn NewStaticPipeWriter(comptime ProcessType: type) type {
-    return struct {
-        const This = @This();
-
-        ref_count: WriterRefCount,
-        writer: IOWriter = .{},
-        stdio_result: StdioResult,
-        source: Source = .{ .detached = {} },
-        process: *ProcessType = undefined,
-        event_loop: JSC.EventLoopHandle,
-        buffer: []const u8 = "",
-
-        // It seems there is a bug in the Zig compiler. We'll get back to this one later
-        const WriterRefCount = bun.ptr.RefCount(@This(), "ref_count", _deinit, .{});
-        pub const ref = WriterRefCount.ref;
-        pub const deref = WriterRefCount.deref;
-
-        const print = bun.Output.scoped(.StaticPipeWriter, false);
-
-        pub const IOWriter = bun.io.BufferedWriter(@This(), struct {
-            pub const onWritable = null;
-            pub const getBuffer = This.getBuffer;
-            pub const onClose = This.onClose;
-            pub const onError = This.onError;
-            pub const onWrite = This.onWrite;
-        });
-        pub const Poll = IOWriter;
-
-        pub fn updateRef(this: *This, add: bool) void {
-            this.writer.updateRef(this.event_loop, add);
-        }
-
-        pub fn getBuffer(this: *This) []const u8 {
-            return this.buffer;
-        }
-
-        pub fn close(this: *This) void {
-            log("StaticPipeWriter(0x{x}) close()", .{@intFromPtr(this)});
-            this.writer.close();
-        }
-
-        pub fn flush(this: *This) void {
-            if (this.buffer.len > 0)
-                this.writer.write();
-        }
-
-        pub fn create(event_loop: anytype, subprocess: *ProcessType, result: StdioResult, source: Source) *This {
-            const this = bun.new(This, .{
-                .ref_count = .init(),
-                .event_loop = JSC.EventLoopHandle.init(event_loop),
-                .process = subprocess,
-                .stdio_result = result,
-                .source = source,
-            });
-            if (Environment.isWindows) {
-                this.writer.setPipe(this.stdio_result.buffer);
-            }
-            this.writer.setParent(this);
-            return this;
-        }
-
-        pub fn start(this: *This) JSC.Maybe(void) {
-            log("StaticPipeWriter(0x{x}) start()", .{@intFromPtr(this)});
-            this.ref();
-            this.buffer = this.source.slice();
-            if (Environment.isWindows) {
-                return this.writer.startWithCurrentPipe();
-            }
-            switch (this.writer.start(this.stdio_result.?, true)) {
-                .err => |err| {
-                    return .{ .err = err };
-                },
-                .result => {
-                    if (comptime Environment.isPosix) {
-                        const poll = this.writer.handle.poll;
-                        poll.flags.insert(.socket);
-                    }
-
-                    return .{ .result = {} };
-                },
-            }
-        }
-
-        pub fn onWrite(this: *This, amount: usize, status: bun.io.WriteStatus) void {
-            log("StaticPipeWriter(0x{x}) onWrite(amount={d} {})", .{ @intFromPtr(this), amount, status });
-            this.buffer = this.buffer[@min(amount, this.buffer.len)..];
-            if (status == .end_of_file or this.buffer.len == 0) {
-                this.writer.close();
-            }
-        }
-
-        pub fn onError(this: *This, err: bun.sys.Error) void {
-            log("StaticPipeWriter(0x{x}) onError(err={any})", .{ @intFromPtr(this), err });
-            this.source.detach();
-        }
-
-        pub fn onClose(this: *This) void {
-            log("StaticPipeWriter(0x{x}) onClose()", .{@intFromPtr(this)});
-            this.source.detach();
-            this.process.onCloseIO(.stdin);
-        }
-
-        fn _deinit(this: *This) void {
-            this.writer.end();
-            this.source.detach();
-            bun.destroy(this);
-        }
-
-        pub fn memoryCost(this: *const This) usize {
-            return @sizeOf(@This()) + this.source.memoryCost() + this.writer.memoryCost();
-        }
-
-        pub fn loop(this: *This) *uws.Loop {
-            return this.event_loop.loop();
-        }
-
-        pub fn watch(this: *This) void {
-            if (this.buffer.len > 0) {
-                this.writer.watch();
-            }
-        }
-
-        pub fn eventLoop(this: *This) JSC.EventLoopHandle {
-            return this.event_loop;
-        }
-    };
-}
-
-pub const PipeReader = struct {
-    const RefCount = bun.ptr.RefCount(@This(), "ref_count", PipeReader.deinit, .{});
-    pub const ref = PipeReader.RefCount.ref;
-    pub const deref = PipeReader.RefCount.deref;
-
-    reader: IOReader = undefined,
-    process: ?*Subprocess = null,
-    event_loop: *JSC.EventLoop = undefined,
-    ref_count: PipeReader.RefCount,
-    state: union(enum) {
-        pending: void,
-        done: []u8,
-        err: bun.sys.Error,
-    } = .{ .pending = {} },
-    stdio_result: StdioResult,
-    pub const IOReader = bun.io.BufferedReader;
-    pub const Poll = IOReader;
-
-    pub fn memoryCost(this: *const PipeReader) usize {
-        return this.reader.memoryCost();
-    }
-
-    pub fn hasPendingActivity(this: *const PipeReader) bool {
-        if (this.state == .pending)
-            return true;
-
-        return this.reader.hasPendingActivity();
-    }
-
-    pub fn detach(this: *PipeReader) void {
-        this.process = null;
-        this.deref();
-    }
-
-    pub fn create(event_loop: *JSC.EventLoop, process: *Subprocess, result: StdioResult, limit: ?*MaxBuf) *PipeReader {
-        var this = bun.new(PipeReader, .{
-            .ref_count = .init(),
-            .process = process,
-            .reader = IOReader.init(@This()),
-            .event_loop = event_loop,
-            .stdio_result = result,
-        });
-        MaxBuf.addToPipereader(limit, &this.reader.maxbuf);
-        if (Environment.isWindows) {
-            this.reader.source = .{ .pipe = this.stdio_result.buffer };
-        }
-        this.reader.setParent(this);
-        return this;
-    }
-
-    pub fn readAll(this: *PipeReader) void {
-        if (this.state == .pending)
-            this.reader.read();
-    }
-
-    pub fn start(this: *PipeReader, process: *Subprocess, event_loop: *JSC.EventLoop) JSC.Maybe(void) {
-        this.ref();
-        this.process = process;
-        this.event_loop = event_loop;
-        if (Environment.isWindows) {
-            return this.reader.startWithCurrentPipe();
-        }
-
-        switch (this.reader.start(this.stdio_result.?, true)) {
-            .err => |err| {
-                return .{ .err = err };
-            },
-            .result => {
-                if (comptime Environment.isPosix) {
-                    const poll = this.reader.handle.poll;
-                    poll.flags.insert(.socket);
-                    this.reader.flags.socket = true;
-                }
-
-                return .{ .result = {} };
-            },
-        }
-    }
-
-    pub const toJS = toReadableStream;
-
-    pub fn onReaderDone(this: *PipeReader) void {
-        const owned = this.toOwnedSlice();
-        this.state = .{ .done = owned };
-        if (this.process) |process| {
-            this.process = null;
-            process.onCloseIO(this.kind(process));
-            this.deref();
-        }
-    }
-
-    pub fn kind(reader: *const PipeReader, process: *const Subprocess) StdioKind {
-        if (process.stdout == .pipe and process.stdout.pipe == reader) {
-            return .stdout;
-        }
-
-        if (process.stderr == .pipe and process.stderr.pipe == reader) {
-            return .stderr;
-        }
-
-        @panic("We should be either stdout or stderr");
-    }
-
-    pub fn toOwnedSlice(this: *PipeReader) []u8 {
-        if (this.state == .done) {
-            return this.state.done;
-        }
-        // we do not use .toOwnedSlice() because we don't want to reallocate memory.
-        const out = this.reader._buffer;
-        this.reader._buffer.items = &.{};
-        this.reader._buffer.capacity = 0;
-
-        if (out.capacity > 0 and out.items.len == 0) {
-            out.deinit();
-            return &.{};
-        }
-
-        return out.items;
-    }
-
-    pub fn updateRef(this: *PipeReader, add: bool) void {
-        this.reader.updateRef(add);
-    }
-
-    pub fn watch(this: *PipeReader) void {
-        if (!this.reader.isDone())
-            this.reader.watch();
-    }
-
-    pub fn toReadableStream(this: *PipeReader, globalObject: *JSC.JSGlobalObject) JSC.JSValue {
-        defer this.detach();
-
-        switch (this.state) {
-            .pending => {
-                const stream = JSC.WebCore.ReadableStream.fromPipe(globalObject, this, &this.reader);
-                this.state = .{ .done = &.{} };
-                return stream;
-            },
-            .done => |bytes| {
-                this.state = .{ .done = &.{} };
-                return JSC.WebCore.ReadableStream.fromOwnedSlice(globalObject, bytes, 0);
-            },
-            .err => |err| {
-                _ = err; // autofix
-                const empty = JSC.WebCore.ReadableStream.empty(globalObject);
-                JSC.WebCore.ReadableStream.cancel(&JSC.WebCore.ReadableStream.fromJS(empty, globalObject).?, globalObject);
-                return empty;
-            },
-        }
-    }
-
-    pub fn toBuffer(this: *PipeReader, globalThis: *JSC.JSGlobalObject) JSC.JSValue {
-        switch (this.state) {
-            .done => |bytes| {
-                defer this.state = .{ .done = &.{} };
-                return JSC.MarkedArrayBuffer.fromBytes(bytes, bun.default_allocator, .Uint8Array).toNodeBuffer(globalThis);
-            },
-            else => {
-                return JSC.JSValue.undefined;
-            },
-        }
-    }
-
-    pub fn onReaderError(this: *PipeReader, err: bun.sys.Error) void {
-        if (this.state == .done) {
-            bun.default_allocator.free(this.state.done);
-        }
-        this.state = .{ .err = err };
-        if (this.process) |process|
-            process.onCloseIO(this.kind(process));
-    }
-
-    pub fn close(this: *PipeReader) void {
-        switch (this.state) {
-            .pending => {
-                this.reader.close();
-            },
-            .done => {},
-            .err => {},
-        }
-    }
-
-    pub fn eventLoop(this: *PipeReader) *JSC.EventLoop {
-        return this.event_loop;
-    }
-
-    pub fn loop(this: *PipeReader) *uws.Loop {
-        return this.event_loop.virtual_machine.uwsLoop();
-    }
-
-    fn deinit(this: *PipeReader) void {
-        if (comptime Environment.isPosix) {
-            bun.assert(this.reader.isDone());
-        }
-
-        if (comptime Environment.isWindows) {
-            bun.assert(this.reader.source == null or this.reader.source.?.isClosed());
-        }
-
-        if (this.state == .done) {
-            bun.default_allocator.free(this.state.done);
-        }
-
-        this.reader.deinit();
-        bun.destroy(this);
-    }
-};
-
-const Writable = union(enum) {
-    pipe: *JSC.WebCore.FileSink,
-    fd: bun.FileDescriptor,
-    buffer: *StaticPipeWriter,
-    memfd: bun.FileDescriptor,
-    inherit: void,
-    ignore: void,
-
-    pub fn memoryCost(this: *const Writable) usize {
-        return switch (this.*) {
-            .pipe => |pipe| pipe.memoryCost(),
-            .buffer => |buffer| buffer.memoryCost(),
-            // TODO: memfd
-            else => 0,
-        };
-    }
-
-    pub fn hasPendingActivity(this: *const Writable) bool {
-        return switch (this.*) {
-            .pipe => false,
-
-            // we mark them as .ignore when they are closed, so this must be true
-            .buffer => true,
-            else => false,
-        };
-    }
-
-    pub fn ref(this: *Writable) void {
-        switch (this.*) {
-            .pipe => {
-                this.pipe.updateRef(true);
-            },
-            .buffer => {
-                this.buffer.updateRef(true);
-            },
-            else => {},
-        }
-    }
-
-    pub fn unref(this: *Writable) void {
-        switch (this.*) {
-            .pipe => {
-                this.pipe.updateRef(false);
-            },
-            .buffer => {
-                this.buffer.updateRef(false);
-            },
-            else => {},
-        }
-    }
-
-    // When the stream has closed we need to be notified to prevent a use-after-free
-    // We can test for this use-after-free by enabling hot module reloading on a file and then saving it twice
-    pub fn onClose(this: *Writable, _: ?bun.sys.Error) void {
-        const process: *Subprocess = @fieldParentPtr("stdin", this);
-
-        if (process.this_jsvalue != .zero) {
-            if (js.stdinGetCached(process.this_jsvalue)) |existing_value| {
-                JSC.WebCore.FileSink.JSSink.setDestroyCallback(existing_value, 0);
-            }
-        }
-
-        switch (this.*) {
-            .buffer => {
-                this.buffer.deref();
-            },
-            .pipe => {
-                this.pipe.deref();
-            },
-            else => {},
-        }
-
-        process.onStdinDestroyed();
-
-        this.* = .{
-            .ignore = {},
-        };
-    }
-    pub fn onReady(_: *Writable, _: ?JSC.WebCore.Blob.SizeType, _: ?JSC.WebCore.Blob.SizeType) void {}
-    pub fn onStart(_: *Writable) void {}
-
-    pub fn init(
-        stdio: Stdio,
-        event_loop: *JSC.EventLoop,
-        subprocess: *Subprocess,
-        result: StdioResult,
-    ) !Writable {
-        assertStdioResult(result);
-
-        if (Environment.isWindows) {
-            switch (stdio) {
-                .pipe => {
-                    if (result == .buffer) {
-                        const pipe = JSC.WebCore.FileSink.createWithPipe(event_loop, result.buffer);
-
-                        switch (pipe.writer.startWithCurrentPipe()) {
-                            .result => {},
-                            .err => |err| {
-                                _ = err; // autofix
-                                pipe.deref();
-                                return error.UnexpectedCreatingStdin;
-                            },
-                        }
-                        pipe.writer.setParent(pipe);
-                        subprocess.weak_file_sink_stdin_ptr = pipe;
-                        subprocess.ref();
-                        subprocess.flags.deref_on_stdin_destroyed = true;
-                        subprocess.flags.has_stdin_destructor_called = false;
-
-                        return Writable{
-                            .pipe = pipe,
-                        };
-                    }
-                    return Writable{ .inherit = {} };
-                },
-
-                .blob => |blob| {
-                    return Writable{
-                        .buffer = StaticPipeWriter.create(event_loop, subprocess, result, .{ .blob = blob }),
-                    };
-                },
-                .array_buffer => |array_buffer| {
-                    return Writable{
-                        .buffer = StaticPipeWriter.create(event_loop, subprocess, result, .{ .array_buffer = array_buffer }),
-                    };
-                },
-                .fd => |fd| {
-                    return Writable{ .fd = fd };
-                },
-                .dup2 => |dup2| {
-                    return Writable{ .fd = dup2.to.toFd() };
-                },
-                .inherit => {
-                    return Writable{ .inherit = {} };
-                },
-                .memfd, .path, .ignore => {
-                    return Writable{ .ignore = {} };
-                },
-                .ipc, .capture => {
-                    return Writable{ .ignore = {} };
-                },
-            }
-        }
-
-        if (comptime Environment.isPosix) {
-            if (stdio == .pipe) {
-                _ = bun.sys.setNonblocking(result.?);
-            }
-        }
-
-        switch (stdio) {
-            .dup2 => @panic("TODO dup2 stdio"),
-            .pipe => {
-                const pipe = JSC.WebCore.FileSink.create(event_loop, result.?);
-
-                switch (pipe.writer.start(pipe.fd, true)) {
-                    .result => {},
-                    .err => |err| {
-                        _ = err; // autofix
-                        pipe.deref();
-                        return error.UnexpectedCreatingStdin;
-                    },
-                }
-
-                subprocess.weak_file_sink_stdin_ptr = pipe;
-                subprocess.ref();
-                subprocess.flags.has_stdin_destructor_called = false;
-                subprocess.flags.deref_on_stdin_destroyed = true;
-
-                pipe.writer.handle.poll.flags.insert(.socket);
-
-                return Writable{
-                    .pipe = pipe,
-                };
-            },
-
-            .blob => |blob| {
-                return Writable{
-                    .buffer = StaticPipeWriter.create(event_loop, subprocess, result, .{ .blob = blob }),
-                };
-            },
-            .array_buffer => |array_buffer| {
-                return Writable{
-                    .buffer = StaticPipeWriter.create(event_loop, subprocess, result, .{ .array_buffer = array_buffer }),
-                };
-            },
-            .memfd => |memfd| {
-                bun.assert(memfd != bun.invalid_fd);
-                return Writable{ .memfd = memfd };
-            },
-            .fd => {
-                return Writable{ .fd = result.? };
-            },
-            .inherit => {
-                return Writable{ .inherit = {} };
-            },
-            .path, .ignore => {
-                return Writable{ .ignore = {} };
-            },
-            .ipc, .capture => {
-                return Writable{ .ignore = {} };
-            },
-        }
-    }
-
-    pub fn toJS(this: *Writable, globalThis: *JSC.JSGlobalObject, subprocess: *Subprocess) JSValue {
-        return switch (this.*) {
-            .fd => |fd| fd.toJS(globalThis),
-            .memfd, .ignore => JSValue.jsUndefined(),
-            .buffer, .inherit => JSValue.jsUndefined(),
-            .pipe => |pipe| {
-                this.* = .{ .ignore = {} };
-                if (subprocess.process.hasExited() and !subprocess.flags.has_stdin_destructor_called) {
-                    // onAttachedProcessExit() can call deref on the
-                    // subprocess. Since we never called ref(), it would be
-                    // unbalanced to do so, leading to a use-after-free.
-                    // So, let's not do that.
-                    // https://github.com/oven-sh/bun/pull/14092
-                    bun.debugAssert(!subprocess.flags.deref_on_stdin_destroyed);
-                    const debug_ref_count = if (Environment.isDebug) subprocess.ref_count else 0;
-                    pipe.onAttachedProcessExit();
-                    if (Environment.isDebug) {
-                        bun.debugAssert(subprocess.ref_count.active_counts == debug_ref_count.active_counts);
-                    }
-                    return pipe.toJS(globalThis);
-                } else {
-                    subprocess.flags.has_stdin_destructor_called = false;
-                    subprocess.weak_file_sink_stdin_ptr = pipe;
-                    subprocess.ref();
-                    subprocess.flags.deref_on_stdin_destroyed = true;
-                    if (@intFromPtr(pipe.signal.ptr) == @intFromPtr(subprocess)) {
-                        pipe.signal.clear();
-                    }
-                    return pipe.toJSWithDestructor(
-                        globalThis,
-                        JSC.WebCore.Sink.DestructorPtr.init(subprocess),
-                    );
-                }
-            },
-        };
-    }
-
-    pub fn finalize(this: *Writable) void {
-        const subprocess: *Subprocess = @fieldParentPtr("stdin", this);
-        if (subprocess.this_jsvalue != .zero) {
-            if (JSC.Codegen.JSSubprocess.stdinGetCached(subprocess.this_jsvalue)) |existing_value| {
-                JSC.WebCore.FileSink.JSSink.setDestroyCallback(existing_value, 0);
-            }
-        }
-
-        return switch (this.*) {
-            .pipe => |pipe| {
-                if (pipe.signal.ptr == @as(*anyopaque, @ptrCast(this))) {
-                    pipe.signal.clear();
-                }
-
-                pipe.deref();
-
-                this.* = .{ .ignore = {} };
-            },
-            .buffer => {
-                this.buffer.updateRef(false);
-                this.buffer.deref();
-            },
-            .memfd => |fd| {
-                fd.close();
-                this.* = .{ .ignore = {} };
-            },
-            .ignore => {},
-            .fd, .inherit => {},
-        };
-    }
-
-    pub fn close(this: *Writable) void {
-        switch (this.*) {
-            .pipe => |pipe| {
-                _ = pipe.end(null);
-            },
-            .memfd => |fd| {
-                fd.close();
-                this.* = .{ .ignore = {} };
-            },
-            .fd => {
-                this.* = .{ .ignore = {} };
-            },
-            .buffer => {
-                this.buffer.close();
-            },
-            .ignore => {},
-            .inherit => {},
-        }
-    }
-};
+pub const Writable = @import("./subprocess/Writable.zig").Writable;
+pub const Readable = @import("./subprocess/Readable.zig").Readable;
 
 pub fn memoryCost(this: *const Subprocess) usize {
     return @sizeOf(@This()) +
@@ -2697,7 +1799,7 @@ const PosixSpawn = bun.spawn;
 const Rusage = bun.spawn.Rusage;
 const Process = bun.spawn.Process;
 const Stdio = bun.spawn.Stdio;
-const StdioResult = if (Environment.isWindows) bun.spawn.WindowsSpawnResult.StdioResult else ?bun.FileDescriptor;
+pub const StdioResult = if (Environment.isWindows) bun.spawn.WindowsSpawnResult.StdioResult else ?bun.FileDescriptor;
 
 const Subprocess = @This();
 pub const MaxBuf = bun.io.MaxBuf;

src/bun.js/api/bun/subprocess/PipeReader.zig

@@ -0,0 +1,220 @@
+const PipeReader = @This();
+
+reader: IOReader = undefined,
+process: ?*Subprocess = null,
+event_loop: *JSC.EventLoop = undefined,
+ref_count: PipeReader.RefCount,
+state: union(enum) {
+    pending: void,
+    done: []u8,
+    err: bun.sys.Error,
+} = .{ .pending = {} },
+stdio_result: StdioResult,
+
+pub const ref = PipeReader.RefCount.ref;
+pub const deref = PipeReader.RefCount.deref;
+pub const Poll = IOReader;
+
+pub fn memoryCost(this: *const PipeReader) usize {
+    return this.reader.memoryCost();
+}
+
+pub fn hasPendingActivity(this: *const PipeReader) bool {
+    if (this.state == .pending)
+        return true;
+
+    return this.reader.hasPendingActivity();
+}
+
+pub fn detach(this: *PipeReader) void {
+    this.process = null;
+    this.deref();
+}
+
+pub fn create(event_loop: *JSC.EventLoop, process: *Subprocess, result: StdioResult, limit: ?*MaxBuf) *PipeReader {
+    var this = bun.new(PipeReader, .{
+        .ref_count = .init(),
+        .process = process,
+        .reader = IOReader.init(@This()),
+        .event_loop = event_loop,
+        .stdio_result = result,
+    });
+    MaxBuf.addToPipereader(limit, &this.reader.maxbuf);
+    if (Environment.isWindows) {
+        this.reader.source = .{ .pipe = this.stdio_result.buffer };
+    }
+    this.reader.setParent(this);
+    return this;
+}
+
+pub fn readAll(this: *PipeReader) void {
+    if (this.state == .pending)
+        this.reader.read();
+}
+
+pub fn start(this: *PipeReader, process: *Subprocess, event_loop: *JSC.EventLoop) JSC.Maybe(void) {
+    this.ref();
+    this.process = process;
+    this.event_loop = event_loop;
+    if (Environment.isWindows) {
+        return this.reader.startWithCurrentPipe();
+    }
+
+    switch (this.reader.start(this.stdio_result.?, true)) {
+        .err => |err| {
+            return .{ .err = err };
+        },
+        .result => {
+            if (comptime Environment.isPosix) {
+                const poll = this.reader.handle.poll;
+                poll.flags.insert(.socket);
+                this.reader.flags.socket = true;
+            }
+
+            return .{ .result = {} };
+        },
+    }
+}
+
+pub const toJS = toReadableStream;
+
+pub fn onReaderDone(this: *PipeReader) void {
+    const owned = this.toOwnedSlice();
+    this.state = .{ .done = owned };
+    if (this.process) |process| {
+        this.process = null;
+        process.onCloseIO(this.kind(process));
+        this.deref();
+    }
+}
+
+pub fn kind(reader: *const PipeReader, process: *const Subprocess) StdioKind {
+    if (process.stdout == .pipe and process.stdout.pipe == reader) {
+        return .stdout;
+    }
+
+    if (process.stderr == .pipe and process.stderr.pipe == reader) {
+        return .stderr;
+    }
+
+    @panic("We should be either stdout or stderr");
+}
+
+pub fn toOwnedSlice(this: *PipeReader) []u8 {
+    if (this.state == .done) {
+        return this.state.done;
+    }
+    // we do not use .toOwnedSlice() because we don't want to reallocate memory.
+    const out = this.reader._buffer;
+    this.reader._buffer.items = &.{};
+    this.reader._buffer.capacity = 0;
+
+    if (out.capacity > 0 and out.items.len == 0) {
+        out.deinit();
+        return &.{};
+    }
+
+    return out.items;
+}
+
+pub fn updateRef(this: *PipeReader, add: bool) void {
+    this.reader.updateRef(add);
+}
+
+pub fn watch(this: *PipeReader) void {
+    if (!this.reader.isDone())
+        this.reader.watch();
+}
+
+pub fn toReadableStream(this: *PipeReader, globalObject: *JSC.JSGlobalObject) JSC.JSValue {
+    defer this.detach();
+
+    switch (this.state) {
+        .pending => {
+            const stream = JSC.WebCore.ReadableStream.fromPipe(globalObject, this, &this.reader);
+            this.state = .{ .done = &.{} };
+            return stream;
+        },
+        .done => |bytes| {
+            this.state = .{ .done = &.{} };
+            return JSC.WebCore.ReadableStream.fromOwnedSlice(globalObject, bytes, 0);
+        },
+        .err => |err| {
+            _ = err; // autofix
+            const empty = JSC.WebCore.ReadableStream.empty(globalObject);
+            JSC.WebCore.ReadableStream.cancel(&JSC.WebCore.ReadableStream.fromJS(empty, globalObject).?, globalObject);
+            return empty;
+        },
+    }
+}
+
+pub fn toBuffer(this: *PipeReader, globalThis: *JSC.JSGlobalObject) JSC.JSValue {
+    switch (this.state) {
+        .done => |bytes| {
+            defer this.state = .{ .done = &.{} };
+            return JSC.MarkedArrayBuffer.fromBytes(bytes, bun.default_allocator, .Uint8Array).toNodeBuffer(globalThis);
+        },
+        else => {
+            return JSC.JSValue.undefined;
+        },
+    }
+}
+
+pub fn onReaderError(this: *PipeReader, err: bun.sys.Error) void {
+    if (this.state == .done) {
+        bun.default_allocator.free(this.state.done);
+    }
+    this.state = .{ .err = err };
+    if (this.process) |process|
+        process.onCloseIO(this.kind(process));
+}
+
+pub fn close(this: *PipeReader) void {
+    switch (this.state) {
+        .pending => {
+            this.reader.close();
+        },
+        .done => {},
+        .err => {},
+    }
+}
+
+pub fn eventLoop(this: *PipeReader) *JSC.EventLoop {
+    return this.event_loop;
+}
+
+pub fn loop(this: *PipeReader) *uws.Loop {
+    return this.event_loop.virtual_machine.uwsLoop();
+}
+
+fn deinit(this: *PipeReader) void {
+    if (comptime Environment.isPosix) {
+        bun.assert(this.reader.isDone());
+    }
+
+    if (comptime Environment.isWindows) {
+        bun.assert(this.reader.source == null or this.reader.source.?.isClosed());
+    }
+
+    if (this.state == .done) {
+        bun.default_allocator.free(this.state.done);
+    }
+
+    this.reader.deinit();
+    bun.destroy(this);
+}
+
+const std = @import("std");
+const bun = @import("bun");
+const JSC = bun.JSC;
+const Subprocess = JSC.API.Subprocess;
+const Stdio = bun.spawn.Stdio;
+const StdioResult = Subprocess.StdioResult;
+const Environment = bun.Environment;
+const Output = bun.Output;
+const JSValue = JSC.JSValue;
+const RefCount = bun.ptr.RefCount(@This(), "ref_count", PipeReader.deinit, .{});
+const MaxBuf = Subprocess.MaxBuf;
+const uws = bun.uws;
+const IOReader = bun.io.BufferedReader;
+const StdioKind = Subprocess.StdioKind;

src/bun.js/api/bun/subprocess/PipeWriter.zig

@@ -0,0 +1,139 @@
+pub fn NewStaticPipeWriter(comptime ProcessType: type) type {
+    return struct {
+        const This = @This();
+
+        ref_count: WriterRefCount,
+        writer: IOWriter = .{},
+        stdio_result: StdioResult,
+        source: Source = .{ .detached = {} },
+        process: *ProcessType = undefined,
+        event_loop: JSC.EventLoopHandle,
+        buffer: []const u8 = "",
+
+        // It seems there is a bug in the Zig compiler. We'll get back to this one later
+        const WriterRefCount = bun.ptr.RefCount(@This(), "ref_count", _deinit, .{});
+        pub const ref = WriterRefCount.ref;
+        pub const deref = WriterRefCount.deref;
+
+        const print = bun.Output.scoped(.StaticPipeWriter, false);
+
+        pub const IOWriter = bun.io.BufferedWriter(@This(), struct {
+            pub const onWritable = null;
+            pub const getBuffer = This.getBuffer;
+            pub const onClose = This.onClose;
+            pub const onError = This.onError;
+            pub const onWrite = This.onWrite;
+        });
+        pub const Poll = IOWriter;
+
+        pub fn updateRef(this: *This, add: bool) void {
+            this.writer.updateRef(this.event_loop, add);
+        }
+
+        pub fn getBuffer(this: *This) []const u8 {
+            return this.buffer;
+        }
+
+        pub fn close(this: *This) void {
+            log("StaticPipeWriter(0x{x}) close()", .{@intFromPtr(this)});
+            this.writer.close();
+        }
+
+        pub fn flush(this: *This) void {
+            if (this.buffer.len > 0)
+                this.writer.write();
+        }
+
+        pub fn create(event_loop: anytype, subprocess: *ProcessType, result: StdioResult, source: Source) *This {
+            const this = bun.new(This, .{
+                .ref_count = .init(),
+                .event_loop = JSC.EventLoopHandle.init(event_loop),
+                .process = subprocess,
+                .stdio_result = result,
+                .source = source,
+            });
+            if (Environment.isWindows) {
+                this.writer.setPipe(this.stdio_result.buffer);
+            }
+            this.writer.setParent(this);
+            return this;
+        }
+
+        pub fn start(this: *This) JSC.Maybe(void) {
+            log("StaticPipeWriter(0x{x}) start()", .{@intFromPtr(this)});
+            this.ref();
+            this.buffer = this.source.slice();
+            if (Environment.isWindows) {
+                return this.writer.startWithCurrentPipe();
+            }
+            switch (this.writer.start(this.stdio_result.?, true)) {
+                .err => |err| {
+                    return .{ .err = err };
+                },
+                .result => {
+                    if (comptime Environment.isPosix) {
+                        const poll = this.writer.handle.poll;
+                        poll.flags.insert(.socket);
+                    }
+
+                    return .{ .result = {} };
+                },
+            }
+        }
+
+        pub fn onWrite(this: *This, amount: usize, status: bun.io.WriteStatus) void {
+            log("StaticPipeWriter(0x{x}) onWrite(amount={d} {})", .{ @intFromPtr(this), amount, status });
+            this.buffer = this.buffer[@min(amount, this.buffer.len)..];
+            if (status == .end_of_file or this.buffer.len == 0) {
+                this.writer.close();
+            }
+        }
+
+        pub fn onError(this: *This, err: bun.sys.Error) void {
+            log("StaticPipeWriter(0x{x}) onError(err={any})", .{ @intFromPtr(this), err });
+            this.source.detach();
+        }
+
+        pub fn onClose(this: *This) void {
+            log("StaticPipeWriter(0x{x}) onClose()", .{@intFromPtr(this)});
+            this.source.detach();
+            this.process.onCloseIO(.stdin);
+        }
+
+        fn _deinit(this: *This) void {
+            this.writer.end();
+            this.source.detach();
+            bun.destroy(this);
+        }
+
+        pub fn memoryCost(this: *const This) usize {
+            return @sizeOf(@This()) + this.source.memoryCost() + this.writer.memoryCost();
+        }
+
+        pub fn loop(this: *This) *uws.Loop {
+            return this.event_loop.loop();
+        }
+
+        pub fn watch(this: *This) void {
+            if (this.buffer.len > 0) {
+                this.writer.watch();
+            }
+        }
+
+        pub fn eventLoop(this: *This) JSC.EventLoopHandle {
+            return this.event_loop;
+        }
+    };
+}
+
+const std = @import("std");
+const bun = @import("bun");
+const JSC = bun.JSC;
+const Subprocess = JSC.API.Subprocess;
+const Stdio = bun.spawn.Stdio;
+const StdioResult = Subprocess.StdioResult;
+const Environment = bun.Environment;
+const Output = bun.Output;
+const Source = bun.io.Source;
+const log = bun.Output.scoped(.StaticPipeWriter, false);
+const uws = bun.uws;

src/bun.js/api/bun/subprocess/Readable.zig

@@ -0,0 +1,192 @@
+pub const Readable = union(enum) {
+    fd: bun.FileDescriptor,
+    memfd: bun.FileDescriptor,
+    pipe: *PipeReader,
+    inherit: void,
+    ignore: void,
+    closed: void,
+    /// Eventually we will implement Readables created from blobs and array buffers.
+    /// When we do that, `buffer` will be borrowed from those objects.
+    ///
+    /// When a buffered `pipe` finishes reading from its file descriptor,
+    /// the owning `Readable` will be convered into this variant and the pipe's
+    /// buffer will be taken as an owned `CowString`.
+    buffer: CowString,
+
+    pub fn memoryCost(this: *const Readable) usize {
+        return switch (this.*) {
+            .pipe => @sizeOf(PipeReader) + this.pipe.memoryCost(),
+            .buffer => this.buffer.length(),
+            else => 0,
+        };
+    }
+
+    pub fn hasPendingActivity(this: *const Readable) bool {
+        return switch (this.*) {
+            .pipe => this.pipe.hasPendingActivity(),
+            else => false,
+        };
+    }
+
+    pub fn ref(this: *Readable) void {
+        switch (this.*) {
+            .pipe => {
+                this.pipe.updateRef(true);
+            },
+            else => {},
+        }
+    }
+
+    pub fn unref(this: *Readable) void {
+        switch (this.*) {
+            .pipe => {
+                this.pipe.updateRef(false);
+            },
+            else => {},
+        }
+    }
+
+    pub fn init(stdio: Stdio, event_loop: *JSC.EventLoop, process: *Subprocess, result: StdioResult, allocator: std.mem.Allocator, max_size: ?*MaxBuf, is_sync: bool) Readable {
+        _ = allocator; // autofix
+        _ = is_sync; // autofix
+        assertStdioResult(result);
+
+        if (comptime Environment.isPosix) {
+            if (stdio == .pipe) {
+                _ = bun.sys.setNonblocking(result.?);
+            }
+        }
+
+        return switch (stdio) {
+            .inherit => Readable{ .inherit = {} },
+            .ignore, .ipc, .path => Readable{ .ignore = {} },
+            .fd => |fd| if (Environment.isPosix) Readable{ .fd = result.? } else Readable{ .fd = fd },
+            .memfd => if (Environment.isPosix) Readable{ .memfd = stdio.memfd } else Readable{ .ignore = {} },
+            .dup2 => |dup2| if (Environment.isPosix) Output.panic("TODO: implement dup2 support in Stdio readable", .{}) else Readable{ .fd = dup2.out.toFd() },
+            .pipe => Readable{ .pipe = PipeReader.create(event_loop, process, result, max_size) },
+            .array_buffer, .blob => Output.panic("TODO: implement ArrayBuffer & Blob support in Stdio readable", .{}),
+            .capture => Output.panic("TODO: implement capture support in Stdio readable", .{}),
+        };
+    }
+
+    pub fn onClose(this: *Readable, _: ?bun.sys.Error) void {
+        this.* = .closed;
+    }
+
+    pub fn onReady(_: *Readable, _: ?JSC.WebCore.Blob.SizeType, _: ?JSC.WebCore.Blob.SizeType) void {}
+
+    pub fn onStart(_: *Readable) void {}
+
+    pub fn close(this: *Readable) void {
+        switch (this.*) {
+            .memfd => |fd| {
+                this.* = .{ .closed = {} };
+                fd.close();
+            },
+            .fd => |_| {
+                this.* = .{ .closed = {} };
+            },
+            .pipe => {
+                this.pipe.close();
+            },
+            else => {},
+        }
+    }
+
+    pub fn finalize(this: *Readable) void {
+        switch (this.*) {
+            .memfd => |fd| {
+                this.* = .{ .closed = {} };
+                fd.close();
+            },
+            .fd => {
+                this.* = .{ .closed = {} };
+            },
+            .pipe => |pipe| {
+                defer pipe.detach();
+                this.* = .{ .closed = {} };
+            },
+            .buffer => |*buf| {
+                buf.deinit(bun.default_allocator);
+            },
+            else => {},
+        }
+    }
+
+    pub fn toJS(this: *Readable, globalThis: *JSC.JSGlobalObject, exited: bool) JSValue {
+        _ = exited; // autofix
+        switch (this.*) {
+            // should only be reachable when the entire output is buffered.
+            .memfd => return this.toBufferedValue(globalThis) catch .zero,
+
+            .fd => |fd| {
+                return fd.toJS(globalThis);
+            },
+            .pipe => |pipe| {
+                defer pipe.detach();
+                this.* = .{ .closed = {} };
+                return pipe.toJS(globalThis);
+            },
+            .buffer => |*buffer| {
+                defer this.* = .{ .closed = {} };
+
+                if (buffer.length() == 0) {
+                    return JSC.WebCore.ReadableStream.empty(globalThis);
+                }
+
+                const own = buffer.takeSlice(bun.default_allocator) catch {
+                    globalThis.throwOutOfMemory() catch return .zero;
+                };
+                return JSC.WebCore.ReadableStream.fromOwnedSlice(globalThis, own, 0);
+            },
+            else => {
+                return JSValue.jsUndefined();
+            },
+        }
+    }
+
+    pub fn toBufferedValue(this: *Readable, globalThis: *JSC.JSGlobalObject) bun.JSError!JSValue {
+        switch (this.*) {
+            .fd => |fd| {
+                return fd.toJS(globalThis);
+            },
+            .memfd => |fd| {
+                if (comptime !Environment.isPosix) {
+                    Output.panic("memfd is only supported on Linux", .{});
+                }
+                this.* = .{ .closed = {} };
+                return JSC.ArrayBuffer.toJSBufferFromMemfd(fd, globalThis);
+            },
+            .pipe => |pipe| {
+                defer pipe.detach();
+                this.* = .{ .closed = {} };
+                return pipe.toBuffer(globalThis);
+            },
+            .buffer => |*buf| {
+                defer this.* = .{ .closed = {} };
+                const own = buf.takeSlice(bun.default_allocator) catch {
+                    return globalThis.throwOutOfMemory();
+                };
+
+                return JSC.MarkedArrayBuffer.fromBytes(own, bun.default_allocator, .Uint8Array).toNodeBuffer(globalThis);
+            },
+            else => {
+                return JSValue.jsUndefined();
+            },
+        }
+    }
+};
+
+const std = @import("std");
+const bun = @import("bun");
+const JSC = bun.JSC;
+const Subprocess = JSC.API.Subprocess;
+const Stdio = bun.spawn.Stdio;
+const StdioResult = Subprocess.StdioResult;
+const MaxBuf = Subprocess.MaxBuf;
+const Environment = bun.Environment;
+const Output = bun.Output;
+const JSValue = JSC.JSValue;
+const PipeReader = @import("./PipeReader.zig");
+const CowString = bun.ptr.CowString;
+const assertStdioResult = Subprocess.assertStdioResult;

src/bun.js/api/bun/subprocess/Writable.zig

@@ -0,0 +1,306 @@
+pub const Writable = union(enum) {
+    pipe: *JSC.WebCore.FileSink,
+    fd: bun.FileDescriptor,
+    buffer: *StaticPipeWriter,
+    memfd: bun.FileDescriptor,
+    inherit: void,
+    ignore: void,
+
+    pub fn memoryCost(this: *const Writable) usize {
+        return switch (this.*) {
+            .pipe => |pipe| pipe.memoryCost(),
+            .buffer => |buffer| buffer.memoryCost(),
+            // TODO: memfd
+            else => 0,
+        };
+    }
+
+    pub fn hasPendingActivity(this: *const Writable) bool {
+        return switch (this.*) {
+            .pipe => false,
+
+            // we mark them as .ignore when they are closed, so this must be true
+            .buffer => true,
+            else => false,
+        };
+    }
+
+    pub fn ref(this: *Writable) void {
+        switch (this.*) {
+            .pipe => {
+                this.pipe.updateRef(true);
+            },
+            .buffer => {
+                this.buffer.updateRef(true);
+            },
+            else => {},
+        }
+    }
+
+    pub fn unref(this: *Writable) void {
+        switch (this.*) {
+            .pipe => {
+                this.pipe.updateRef(false);
+            },
+            .buffer => {
+                this.buffer.updateRef(false);
+            },
+            else => {},
+        }
+    }
+
+    // When the stream has closed we need to be notified to prevent a use-after-free
+    // We can test for this use-after-free by enabling hot module reloading on a file and then saving it twice
+    pub fn onClose(this: *Writable, _: ?bun.sys.Error) void {
+        const process: *Subprocess = @fieldParentPtr("stdin", this);
+
+        if (process.this_jsvalue != .zero) {
+            if (js.stdinGetCached(process.this_jsvalue)) |existing_value| {
+                JSC.WebCore.FileSink.JSSink.setDestroyCallback(existing_value, 0);
+            }
+        }
+
+        switch (this.*) {
+            .buffer => {
+                this.buffer.deref();
+            },
+            .pipe => {
+                this.pipe.deref();
+            },
+            else => {},
+        }
+
+        process.onStdinDestroyed();
+
+        this.* = .{
+            .ignore = {},
+        };
+    }
+    pub fn onReady(_: *Writable, _: ?JSC.WebCore.Blob.SizeType, _: ?JSC.WebCore.Blob.SizeType) void {}
+    pub fn onStart(_: *Writable) void {}
+
+    pub fn init(
+        stdio: Stdio,
+        event_loop: *JSC.EventLoop,
+        subprocess: *Subprocess,
+        result: StdioResult,
+    ) !Writable {
+        assertStdioResult(result);
+
+        if (Environment.isWindows) {
+            switch (stdio) {
+                .pipe => {
+                    if (result == .buffer) {
+                        const pipe = JSC.WebCore.FileSink.createWithPipe(event_loop, result.buffer);
+
+                        switch (pipe.writer.startWithCurrentPipe()) {
+                            .result => {},
+                            .err => |err| {
+                                _ = err; // autofix
+                                pipe.deref();
+                                return error.UnexpectedCreatingStdin;
+                            },
+                        }
+                        pipe.writer.setParent(pipe);
+                        subprocess.weak_file_sink_stdin_ptr = pipe;
+                        subprocess.ref();
+                        subprocess.flags.deref_on_stdin_destroyed = true;
+                        subprocess.flags.has_stdin_destructor_called = false;
+
+                        return Writable{
+                            .pipe = pipe,
+                        };
+                    }
+                    return Writable{ .inherit = {} };
+                },
+
+                .blob => |blob| {
+                    return Writable{
+                        .buffer = StaticPipeWriter.create(event_loop, subprocess, result, .{ .blob = blob }),
+                    };
+                },
+                .array_buffer => |array_buffer| {
+                    return Writable{
+                        .buffer = StaticPipeWriter.create(event_loop, subprocess, result, .{ .array_buffer = array_buffer }),
+                    };
+                },
+                .fd => |fd| {
+                    return Writable{ .fd = fd };
+                },
+                .dup2 => |dup2| {
+                    return Writable{ .fd = dup2.to.toFd() };
+                },
+                .inherit => {
+                    return Writable{ .inherit = {} };
+                },
+                .memfd, .path, .ignore => {
+                    return Writable{ .ignore = {} };
+                },
+                .ipc, .capture => {
+                    return Writable{ .ignore = {} };
+                },
+            }
+        }
+
+        if (comptime Environment.isPosix) {
+            if (stdio == .pipe) {
+                _ = bun.sys.setNonblocking(result.?);
+            }
+        }
+
+        switch (stdio) {
+            .dup2 => @panic("TODO dup2 stdio"),
+            .pipe => {
+                const pipe = JSC.WebCore.FileSink.create(event_loop, result.?);
+
+                switch (pipe.writer.start(pipe.fd, true)) {
+                    .result => {},
+                    .err => |err| {
+                        _ = err; // autofix
+                        pipe.deref();
+                        return error.UnexpectedCreatingStdin;
+                    },
+                }
+
+                subprocess.weak_file_sink_stdin_ptr = pipe;
+                subprocess.ref();
+                subprocess.flags.has_stdin_destructor_called = false;
+                subprocess.flags.deref_on_stdin_destroyed = true;
+
+                pipe.writer.handle.poll.flags.insert(.socket);
+
+                return Writable{
+                    .pipe = pipe,
+                };
+            },
+
+            .blob => |blob| {
+                return Writable{
+                    .buffer = StaticPipeWriter.create(event_loop, subprocess, result, .{ .blob = blob }),
+                };
+            },
+            .array_buffer => |array_buffer| {
+                return Writable{
+                    .buffer = StaticPipeWriter.create(event_loop, subprocess, result, .{ .array_buffer = array_buffer }),
+                };
+            },
+            .memfd => |memfd| {
+                bun.assert(memfd != bun.invalid_fd);
+                return Writable{ .memfd = memfd };
+            },
+            .fd => {
+                return Writable{ .fd = result.? };
+            },
+            .inherit => {
+                return Writable{ .inherit = {} };
+            },
+            .path, .ignore => {
+                return Writable{ .ignore = {} };
+            },
+            .ipc, .capture => {
+                return Writable{ .ignore = {} };
+            },
+        }
+    }
+
+    pub fn toJS(this: *Writable, globalThis: *JSC.JSGlobalObject, subprocess: *Subprocess) JSValue {
+        return switch (this.*) {
+            .fd => |fd| fd.toJS(globalThis),
+            .memfd, .ignore => JSValue.jsUndefined(),
+            .buffer, .inherit => JSValue.jsUndefined(),
+            .pipe => |pipe| {
+                this.* = .{ .ignore = {} };
+                if (subprocess.process.hasExited() and !subprocess.flags.has_stdin_destructor_called) {
+                    // onAttachedProcessExit() can call deref on the
+                    // subprocess. Since we never called ref(), it would be
+                    // unbalanced to do so, leading to a use-after-free.
+                    // So, let's not do that.
+                    // https://github.com/oven-sh/bun/pull/14092
+                    bun.debugAssert(!subprocess.flags.deref_on_stdin_destroyed);
+                    const debug_ref_count = if (Environment.isDebug) subprocess.ref_count else 0;
+                    pipe.onAttachedProcessExit();
+                    if (Environment.isDebug) {
+                        bun.debugAssert(subprocess.ref_count.active_counts == debug_ref_count.active_counts);
+                    }
+                    return pipe.toJS(globalThis);
+                } else {
+                    subprocess.flags.has_stdin_destructor_called = false;
+                    subprocess.weak_file_sink_stdin_ptr = pipe;
+                    subprocess.ref();
+                    subprocess.flags.deref_on_stdin_destroyed = true;
+                    if (@intFromPtr(pipe.signal.ptr) == @intFromPtr(subprocess)) {
+                        pipe.signal.clear();
+                    }
+                    return pipe.toJSWithDestructor(
+                        globalThis,
+                        JSC.WebCore.Sink.DestructorPtr.init(subprocess),
+                    );
+                }
+            },
+        };
+    }
+
+    pub fn finalize(this: *Writable) void {
+        const subprocess: *Subprocess = @fieldParentPtr("stdin", this);
+        if (subprocess.this_jsvalue != .zero) {
+            if (JSC.Codegen.JSSubprocess.stdinGetCached(subprocess.this_jsvalue)) |existing_value| {
+                JSC.WebCore.FileSink.JSSink.setDestroyCallback(existing_value, 0);
+            }
+        }
+
+        return switch (this.*) {
+            .pipe => |pipe| {
+                if (pipe.signal.ptr == @as(*anyopaque, @ptrCast(this))) {
+                    pipe.signal.clear();
+                }
+
+                pipe.deref();
+
+                this.* = .{ .ignore = {} };
+            },
+            .buffer => {
+                this.buffer.updateRef(false);
+                this.buffer.deref();
+            },
+            .memfd => |fd| {
+                fd.close();
+                this.* = .{ .ignore = {} };
+            },
+            .ignore => {},
+            .fd, .inherit => {},
+        };
+    }
+
+    pub fn close(this: *Writable) void {
+        switch (this.*) {
+            .pipe => |pipe| {
+                _ = pipe.end(null);
+            },
+            .memfd => |fd| {
+                fd.close();
+                this.* = .{ .ignore = {} };
+            },
+            .fd => {
+                this.* = .{ .ignore = {} };
+            },
+            .buffer => {
+                this.buffer.close();
+            },
+            .ignore => {},
+            .inherit => {},
+        }
+    }
+};
+
+const std = @import("std");
+const bun = @import("bun");
+const JSC = bun.JSC;
+const Subprocess = JSC.API.Subprocess;
+const Stdio = bun.spawn.Stdio;
+const StdioResult = Subprocess.StdioResult;
+const Environment = bun.Environment;
+const Output = bun.Output;
+const JSValue = JSC.JSValue;
+const StaticPipeWriter = Subprocess.StaticPipeWriter;
+const js = Subprocess.js;
+const assertStdioResult = Subprocess.assertStdioResult;