child_process: honor child.stdout.pause() once the stream has flowed

src/io/PipeReader.rs

@@ -458,7 +458,11 @@ impl PosixBufferedReader {
 
     // Exists for consistently with Windows.
     pub fn has_pending_read(&self) -> bool {
-        matches!(&self.handle, PollOrFd::Poll(poll) if poll.is_registered())
+        // `is_watching()` (registered && !needs-rearm) rather than
+        // `is_registered()`: a one-shot poll that has fired but not been
+        // re-armed will not deliver another callback, so callers that skip
+        // `read()` on "pending" must not be told one is in flight.
+        matches!(&self.handle, PollOrFd::Poll(poll) if poll.is_watching())
     }
 
     pub fn watch(&mut self) {
@@ -672,16 +676,21 @@ impl PosixBufferedReader {
 
                         if streaming {
                             // Stream this chunk and register for next cycle
-                            if !parent.vtable.on_read_chunk(
+                            let keep_going = parent.vtable.on_read_chunk(
                                 &stack_buffer[..bytes_read],
                                 if received_hup && bytes_read < stack_buffer.len() {
                                     ReadState::Eof
                                 } else {
                                     ReadState::Progress
                                 },
-                            ) && !received_hup
-                                && !over_budget
-                            {
+                            );
+                            // Re-entrant JS inside on_read_chunk can close the
+                            // reader (nested on_pull -> read -> EOF); the
+                            // captured `fd` is then stale regardless of HUP.
+                            if parent.is_done() {
+                                return;
+                            }
+                            if !keep_going && !received_hup && !over_budget {
                                 return;
                             }
                         } else {
@@ -739,6 +748,9 @@ impl PosixBufferedReader {
                                         ReadState::Progress
                                     },
                                 );
+                                if parent.is_done() {
+                                    return;
+                                }
                                 // Closing for `over_budget` outranks the
                                 // consumer asking us to stop: it must still
                                 // happen, or nothing ever caps the pipe.
@@ -900,15 +912,21 @@ impl PosixBufferedReader {
                                 // Once HUP is set the kernel
                                 // returns the remaining bytes then 0, so
                                 // draining to `bytes_read == 0` is bounded.
-                                if !parent.vtable.on_read_chunk(
+                                let keep_going = parent.vtable.on_read_chunk(
                                     &event_loop.pipe_read_buffer_mut()[..head_start],
                                     if received_hup {
                                         ReadState::Eof
                                     } else {
                                         ReadState::Progress
                                     },
-                                ) && !received_hup
-                                {
+                                );
+                                // Re-entrant close (nested on_pull -> read ->
+                                // EOF) invalidates the captured `fd`; stop
+                                // before the next recv regardless of HUP.
+                                if parent.is_done() {
+                                    return;
+                                }
+                                if !keep_going && !received_hup {
                                     return;
                                 }
                                 head_start = 0;
@@ -949,15 +967,18 @@ impl PosixBufferedReader {
                 }
 
                 if head_start > 0 {
-                    if !parent.vtable.on_read_chunk(
+                    let keep_going = parent.vtable.on_read_chunk(
                         &event_loop.pipe_read_buffer_mut()[..head_start],
                         if received_hup {
                             ReadState::Eof
                         } else {
                             ReadState::Progress
                         },
-                    ) && !received_hup
-                    {
+                    );
+                    if parent.is_done() {
+                        return;
+                    }
+                    if !keep_going && !received_hup {
                         return;
                     }
                 }
@@ -1048,7 +1069,7 @@ impl PosixBufferedReader {
                                 .vtable
                                 .on_read_chunk(&parent._buffer, ReadState::Progress);
                             parent._buffer.clear();
-                            if !keep_going {
+                            if parent.is_done() || !keep_going {
                                 return;
                             }
                             continue;
@@ -1912,7 +1933,14 @@ impl WindowsBufferedReader {
         // grows by `amount_result` every chunk and never resets, so a 1 GB
         // `cat` holds 1 GB resident instead of ~64 KB. Clear it here, after
         // the streaming consumer has finished with `slice`.
-        if should_continue && has_more != ReadState::Eof && self.vtable.is_streaming_enabled() {
+        // `should_continue` no longer gates the clear: FileReader may say
+        // stop at its highwater mark while uv keeps delivering, and leaving
+        // `_buffer` uncleared would double-buffer (here + FileReader.buffered).
+        // Parents that want the reader paused call `reader().pause()`
+        // themselves; stopping here could free a parent whose caller still
+        // holds `this` (FileResponseStream on abort).
+        let _ = should_continue;
+        if has_more != ReadState::Eof && self.vtable.is_streaming_enabled() {
             self._buffer.clear();
         }
 

src/js/internal/streams/native-readable.ts

@@ -29,7 +29,7 @@ let dynamicallyAdjustChunkSize = (_?) => (
 
 type NativeReadable = typeof import("node:stream").Readable &
   typeof import("node:stream").Stream & {
-    push: (chunk: any) => void;
+    push: (chunk: any) => boolean;
     $bunNativePtr?: NativePtr;
     [kRefCount]: number;
     [kCloseState]: [boolean];
@@ -48,6 +48,7 @@ interface NativePtr {
   pull: (view: any, closer: any) => any;
   updateRef: (ref: boolean) => void;
   cancel: (error: any) => void;
+  setFlowing?: (flowing: boolean) => void;
 }
 
 let debugId = 0;
@@ -122,10 +123,14 @@ function getRemainingChunk(stream: NativeReadable, maxToRead?: number) {
 
 function read(this: NativeReadable, maxToRead: number) {
   $debug(`[${this.debugId}] read${this[kPendingRead] ? ", is already pending" : ""}`);
+  var ptr = this.$bunNativePtr;
+  // Readable called `_read`, so it wants data: make sure the native reader is
+  // not paused from a previous `push()===false` (readStart, like net.Socket).
+  // Runs even when a pull promise is outstanding so that promise can resolve.
+  if (ptr) ptr.setFlowing?.(true);
   if (this[kPendingRead]) {
     return;
   }
-  var ptr = this.$bunNativePtr;
   if (!ptr) {
     $debug(`[${this.debugId}] read, no ptr`);
     this.push(null);
@@ -139,13 +144,13 @@ function read(this: NativeReadable, maxToRead: number) {
       this[kHighWaterMark] = Math.min(this[kHighWaterMark], result);
     }
     if ($isTypedArrayView(result) && result.byteLength > 0) {
-      this.push(result);
+      pushAndCheck(this, result);
     }
     const drainResult = ptr.drain();
     this[kConstructed] = true;
     $debug(`[${this.debugId}] drain result: ${drainResult?.byteLength ?? "null"}`);
     if ((drainResult?.byteLength ?? 0) > 0) {
-      this.push(drainResult);
+      pushAndCheck(this, drainResult);
     }
   }
   const chunk = getRemainingChunk(this, maxToRead);
@@ -196,12 +201,22 @@ function handleResult(stream: NativeReadable, result: any, chunk: Buffer, isClos
   }
 }
 
+// `push()` returning false means the Readable's buffer is at/above hwm (or
+// the consumer paused); stop the native reader so kernel backpressure reaches
+// the writer (readStop, like net.Socket). The next `_read()` re-enables it.
+function pushAndCheck(stream: NativeReadable, chunk: any) {
+  if (!stream.push(chunk)) {
+    const ptr = stream.$bunNativePtr;
+    if (ptr) ptr.setFlowing?.(false);
+  }
+}
+
 function handleNumberResult(stream: NativeReadable, result: number, chunk: any, isClosed: boolean) {
   if (result > 0) {
     const slice = chunk.subarray(0, result);
     chunk = slice.byteLength < chunk.byteLength ? chunk.subarray(result) : undefined;
     if (slice.byteLength > 0) {
-      stream.push(slice);
+      pushAndCheck(stream, slice);
     }
   }
 
@@ -216,7 +231,7 @@ function handleNumberResult(stream: NativeReadable, result: number, chunk: any,
 
 function handleArrayBufferViewResult(stream: NativeReadable, result: any, chunk: any, isClosed: boolean) {
   if (result.byteLength > 0) {
-    stream.push(result);
+    pushAndCheck(stream, result);
   }
 
   if (isClosed) {

src/runtime/webcore/FileReader.rs

@@ -778,8 +778,15 @@ impl FileReader {
             unsafe { (*parent).increment_count() };
             self.pending.with_mut(|p| p.run());
             close_if_needed!();
-            // Re-entrant cancel closed the reader; tell the io caller to stop.
-            let ret = if self.done.get() { false } else { ret };
+            // Re-entrant cancel (sets `done`) or a nested on_pull that read to
+            // EOF (sets IS_DONE via on_reader_done but not `self.done`) closed
+            // the reader; tell the io caller to stop so it does not re-read the
+            // captured fd.
+            let ret = if self.done.get() || self.reader().is_done() {
+                false
+            } else {
+                ret
+            };
             // SAFETY: see `parent()`; the pin keeps the count >= 1, so this
             // never frees. `self` is not accessed after.
             let _ = unsafe { Source::decrement_count(parent) };
@@ -794,14 +801,19 @@ impl FileReader {
         }
 
         // No JS read is waiting; stop at the highwater mark. onPull restarts.
+        //
+        // `started` gates the backstop: a `from_pipe()` reader for non-lazy
+        // `Bun.spawn` is already reading when it arrives here, and throttling
+        // before any consumer has attached deadlocks a child that alternates
+        // stdout/stderr writes while the caller only awaits one of them.
         // SAFETY: see `reader_buffer` decl.
         let reader_buffer_len = unsafe { (*reader_buffer).len() };
-        let ret = self.flowing.get()
-            && !matches!(
-                self.read_inside_on_pull.get(),
-                ReadDuringJSOnPullResult::Temporary(_)
-            )
-            && self.buffered.get().len() + reader_buffer_len < self.highwater_mark;
+        let ret = !matches!(
+            self.read_inside_on_pull.get(),
+            ReadDuringJSOnPullResult::Temporary(_)
+        ) && (!self.started.get()
+            || (self.flowing.get()
+                && self.buffered.get().len() + reader_buffer_len < self.highwater_mark));
         close_if_needed!();
         ret
     }
@@ -950,12 +962,6 @@ impl FileReader {
         self.pending_value.with_mut(|p| p.set(&global, array));
         self.pending_view.set(buffer);
 
-        // `has_pending_read()` tracks the registration flag, not the kernel
-        // arm state: the highwater backstop leaves the one-shot poll disarmed.
-        if self.flowing.get() {
-            self.reader().watch();
-        }
-
         bun_core::scoped_log!(FileReader, "onPull({}) = pending", buffer_len);
 
         streams::Result::Pending(self.pending.as_ptr())

test/js/node/child_process/child_process.test.ts

@@ -992,6 +992,68 @@ describe.skipIf(!isPosix)("stdout pipe backpressure", () => {
   });
 });
 
+// child.stdout.pause() must stop the native reader so the kernel pipe fills
+// and the child blocks on write. Previously, once the stream had flowed even
+// once the native FileReader kept the poll armed (or uv_read_start active on
+// Windows) regardless of JS state, so the child wrote its entire output into
+// the parent's heap and 'data' kept firing after #handleOnExit resumed it.
+it("child.stdout.pause() after flowing stops native reads and blocks the child", async () => {
+  // 20 MB: well above any kernel socket buffer, small enough to drain fast
+  // once resumed on ASAN.
+  const SIZE = 20 * 1024 * 1024;
+  const writer = `const c=Buffer.alloc(1<<20,97);let w=0;(function f(){while(w<${SIZE}){const n=Math.min(c.length,${SIZE}-w);w+=n;if(!process.stdout.write(c.subarray(0,n))){process.stdout.once('drain',f);return}}})()`;
+  const c = spawn(bunExe(), ["-e", writer], {
+    stdio: ["ignore", "pipe", "ignore"],
+    env: bunEnv,
+  });
+  try {
+    let events = 0;
+    let bytes = 0;
+    let eventsAfterPause = 0;
+    const { promise: firstData, resolve: gotFirst, reject: failFirst } = Promise.withResolvers<void>();
+    c.on("error", failFirst);
+    c.on("close", () => failFirst(new Error("child closed before first 'data'")));
+    c.stdout!.on("data", (d: Buffer) => {
+      events++;
+      bytes += d.length;
+      if (events === 1) {
+        c.stdout!.pause();
+        gotFirst();
+      } else {
+        eventsAfterPause++;
+      }
+    });
+    await firstData;
+
+    expect(c.stdout!.isPaused()).toBe(true);
+
+    // Give the eager-read path every chance to over-buffer: without the fix
+    // the native reader has already drained most of SIZE and the child is
+    // racing to exit. Deadline-polled; breaks early if the bug is present.
+    const deadline = Date.now() + 1000;
+    while (c.exitCode === null && Date.now() < deadline) {
+      await new Promise(r => setImmediate(r));
+    }
+
+    // Core assertion: pause() applied backpressure, so the child cannot have
+    // finished writing SIZE bytes and is blocked on a full pipe.
+    expect(c.exitCode).toBeNull();
+    expect(eventsAfterPause).toBe(0);
+    expect(c.stdout!.isPaused()).toBe(true);
+    // At most a few pipe-buffer-sized chunks were delivered before pause took
+    // effect, never the whole output.
+    expect(bytes).toBeLessThan(SIZE);
+
+    // Resuming delivers every byte and the child exits cleanly.
+    c.stdout!.resume();
+    await once(c, "close");
+    expect(bytes).toBe(SIZE);
+    expect(c.exitCode).toBe(0);
+  } finally {
+    c.kill();
+  }
+});
+
 // When spawn fails (ENOENT, bad cwd, etc.) the ChildProcess emits 'error' and
 // 'close' but never 'exit'. The abort listener on options.signal was only
 // removed on 'exit', so every failed spawn against a shared AbortSignal leaked