Fix: after pausing stdin, a subprocess should be able to read from stdin (#23341)

Fixes #23333, Fixes #13978

### What does this PR do?

### How did you verify your code works?

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Claude Bot <[email protected]>
Co-authored-by: Claude <[email protected]>
Co-authored-by: pfg <[email protected]>
Co-authored-by: Zack Radisic <[email protected]>

Author: 6 people

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

@@ -397,7 +397,9 @@ const StreamTransfer = struct {
                     log("max_size reached, ending stream", .{});
                     if (this.route.server) |server| {
                         // dont need to ref because we are already holding a ref and will be derefed in onReaderDone
-                        this.reader.pause();
+                        if (!bun.Environment.isPosix) {
+                            this.reader.pause();
+                        }
                         // we cannot free inside onReadChunk this would be UAF so we schedule it to be done in the next event loop tick
                         this.eof_task = jsc.AnyTask.New(StreamTransfer, StreamTransfer.onReaderDone).init(this);
                         server.vm().enqueueTask(jsc.Task.init(&this.eof_task.?));
@@ -430,7 +432,9 @@ const StreamTransfer = struct {
                 // pause the reader so deref until onWritable
                 defer this.deref();
                 this.resp.onWritable(*StreamTransfer, onWritable, this);
-                this.reader.pause();
+                if (!bun.Environment.isPosix) {
+                    this.reader.pause();
+                }
                 return false;
             },
             .want_more => {

src/bun.js/api/streams.classes.ts

@@ -72,6 +72,10 @@ function source(name) {
               fn: "setRawModeFromJS",
               length: 1,
             },
+            setFlowing: {
+              fn: "setFlowingFromJS",
+              length: 1,
+            },
           }
         : {}),
     },

src/bun.js/webcore/FileReader.zig

@@ -18,6 +18,7 @@ lazy: Lazy = .{ .none = {} },
 buffered: std.ArrayListUnmanaged(u8) = .{},
 read_inside_on_pull: ReadDuringJSOnPullResult = .{ .none = {} },
 highwater_mark: usize = 16384,
+flowing: bool = true,
 
 pub const IOReader = bun.io.BufferedReader;
 pub const Poll = IOReader;
@@ -487,6 +488,14 @@ pub fn onPull(this: *FileReader, buffer: []u8, array: jsc.JSValue) streams.Resul
     }
 
     if (!this.reader.hasPendingRead()) {
+        // If not flowing (paused), don't initiate new reads
+        if (!this.flowing) {
+            log("onPull({d}) = pending (not flowing)", .{buffer.len});
+            this.pending_value.set(this.parent().globalThis, array);
+            this.pending_view = buffer;
+            return .{ .pending = &this.pending };
+        }
+
         this.read_inside_on_pull = .{ .js = buffer };
         this.reader.read();
 
@@ -581,32 +590,15 @@ pub fn onReaderDone(this: *FileReader) void {
             }
             this.buffered = .{};
             this.pending.run();
-        } else if (this.buffered.items.len > 0) {
-            const this_value = this.parent().this_jsvalue;
-            const globalThis = this.parent().globalThis;
-            if (this_value != .zero) {
-                if (Source.js.onDrainCallbackGetCached(this_value)) |cb| {
-                    const buffered = this.buffered;
-                    this.buffered = .{};
-                    this.parent().incrementCount();
-                    defer _ = this.parent().decrementCount();
-                    this.eventLoop().js.runCallback(
-                        cb,
-                        globalThis,
-                        .js_undefined,
-                        &.{
-                            jsc.ArrayBuffer.fromBytes(buffered.items, .Uint8Array).toJS(globalThis) catch |err| {
-                                this.pending.result = .{ .err = .{ .WeakJSValue = globalThis.takeException(err) } };
-                                return;
-                            },
-                        },
-                    );
-                }
-            }
         }
+        // Don't handle buffered data here - it will be returned on the next onPull
+        // This ensures proper ordering of chunks
     }
 
-    this.parent().onClose();
+    // Only close the stream if there's no buffered data left to deliver
+    if (this.buffered.items.len == 0) {
+        this.parent().onClose();
+    }
     if (this.waiting_for_onReaderDone) {
         this.waiting_for_onReaderDone = false;
         _ = this.parent().decrementCount();
@@ -631,6 +623,26 @@ pub fn setRawMode(this: *FileReader, flag: bool) bun.sys.Maybe(void) {
     return this.reader.setRawMode(flag);
 }
 
+pub fn setFlowing(this: *FileReader, flag: bool) void {
+    log("setFlowing({}) was={}", .{ flag, this.flowing });
+
+    if (this.flowing == flag) {
+        return;
+    }
+
+    this.flowing = flag;
+
+    if (flag) {
+        this.reader.unpause();
+        if (!this.reader.isDone() and !this.reader.hasPendingRead()) {
+            // Kick off a new read if needed
+            this.reader.read();
+        }
+    } else {
+        this.reader.pause();
+    }
+}
+
 pub fn memoryCost(this: *const FileReader) usize {
     // ReadableStreamSource covers @sizeOf(FileReader)
     return this.reader.memoryCost() + this.buffered.capacity;

src/bun.js/webcore/ReadableStream.zig

@@ -577,6 +577,19 @@ pub fn NewSource(
             @compileError("setRawMode is not implemented on " ++ @typeName(Context));
         }
 
+        pub fn setFlowingFromJS(this: *ReadableStreamSourceType, _: *jsc.JSGlobalObject, call_frame: *jsc.CallFrame) bun.JSError!JSValue {
+            if (@hasDecl(Context, "setFlowing")) {
+                const flag = call_frame.argument(0);
+                if (Environment.allow_assert) {
+                    bun.assert(flag.isBoolean());
+                }
+                this.context.setFlowing(flag == .true);
+                return .js_undefined;
+            }
+
+            return .js_undefined;
+        }
+
         const supports_ref = setRefUnrefFn != null;
 
         pub const js = @field(jsc.Codegen, "JS" ++ name_ ++ "InternalReadableStreamSource");