child_process: apply kernel backpressure to stdout/stderr pipes

src/io/PipeReader.rs

@@ -157,12 +157,13 @@ bitflags::bitflags! {
         const MEMFD                    = 1 << 7;
         const USE_PREAD                = 1 << 8;
         const IS_PAUSED                = 1 << 9;
+        const KEEP_ALIVE               = 1 << 10; // default true
     }
 }
 
 impl PosixFlags {
     pub const fn new() -> Self {
-        PosixFlags::CLOSE_HANDLE
+        Self::from_bits_truncate(PosixFlags::CLOSE_HANDLE.bits() | PosixFlags::KEEP_ALIVE.bits())
     }
 }
 
@@ -179,7 +180,10 @@ impl PosixBufferedReader {
         }
     }
 
-    pub fn update_ref(&self, value: bool) {
+    pub fn update_ref(&mut self, value: bool) {
+        // Remember the ref state so a poll created later (lazy start) honours
+        // an unref() that preceded the first registration.
+        self.flags.set(PosixFlags::KEEP_ALIVE, value);
         let Some(poll) = self.handle.get_poll() else {
             return;
         };
@@ -331,11 +335,11 @@ impl PosixBufferedReader {
         self.buffer()
     }
 
-    pub fn disable_keeping_process_alive<C>(&self, _event_loop_ctx: C) {
+    pub fn disable_keeping_process_alive<C>(&mut self, _event_loop_ctx: C) {
         self.update_ref(false);
     }
 
-    pub fn enable_keeping_process_alive<C>(&self, _event_loop_ctx: C) {
+    pub fn enable_keeping_process_alive<C>(&mut self, _event_loop_ctx: C) {
         self.update_ref(true);
     }
 
@@ -414,7 +418,9 @@ impl PosixBufferedReader {
         };
         poll.set_owner(Owner::new(PollTag::BufferedReader, owner_ptr.cast()));
 
-        if !poll.has_flag(FilePollFlag::WasEverRegistered) {
+        if !poll.has_flag(FilePollFlag::WasEverRegistered)
+            && self.flags.contains(PosixFlags::KEEP_ALIVE)
+        {
             poll.enable_keeping_process_alive(ev);
         }
 
@@ -439,7 +445,9 @@ impl PosixBufferedReader {
         if self.get_fd() != fd {
             self.handle = PollOrFd::Fd(fd);
         }
-        self.register_poll();
+        if !self.flags.contains(PosixFlags::IS_PAUSED) {
+            self.register_poll();
+        }
 
         sys::Result::Ok(())
     }

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

@@ -152,13 +152,15 @@ function read(this: NativeReadable, maxToRead: number) {
   var result = ptr.pull(chunk, this[kCloseState]);
   $assert(result !== undefined);
   $debug(
-    `[${this.debugId}] pull ${chunk?.byteLength} bytes, result: ${result instanceof Promise ? "<pending>" : result}, closeState: ${this[kCloseState][0]}`,
+    `[${this.debugId}] pull ${chunk?.byteLength} bytes, result: ${$isPromise(result) ? "<pending>" : $isTypedArrayView(result) ? `<${result.byteLength} bytes>` : result}, closeState: ${this[kCloseState][0]}`,
   );
   if ($isPromise(result)) {
     this[kPendingRead] = true;
     return result.then(
       result => {
-        $debug(`[${this.debugId}] pull, resolved: ${result}, closeState: ${this[kCloseState][0]}`);
+        $debug(
+          `[${this.debugId}] pull, resolved: ${$isTypedArrayView(result) ? `<${result.byteLength} bytes>` : result}, closeState: ${this[kCloseState][0]}`,
+        );
         this[kPendingRead] = false;
         this[kRemainingChunk] = handleResult(this, result, chunk, this[kCloseState][0]);
       },

src/js/node/child_process.ts

@@ -1135,13 +1135,13 @@ class ChildProcess extends EventEmitter {
 
       if (stdout === undefined) {
         this.#stdout = this.#getBunSpawnIo(1, this.#encoding, true);
-      } else if (stdout && this.#stdioOptions[1] === "pipe" && !stdout?.destroyed) {
+      } else if (stdout && this.#stdioOptions[1] === "pipe" && !stdout.destroyed && stdout.readable) {
         stdout.resume?.();
       }
 
       if (stderr === undefined) {
         this.#stderr = this.#getBunSpawnIo(2, this.#encoding, true);
-      } else if (stderr && this.#stdioOptions[2] === "pipe" && !stderr?.destroyed) {
+      } else if (stderr && this.#stdioOptions[2] === "pipe" && !stderr.destroyed && stderr.readable) {
         stderr.resume?.();
       }
     }

src/runtime/api/bun/js_bun_spawn_bindings.rs

@@ -1737,7 +1737,9 @@ pub(crate) fn spawn_maybe_sync<const IS_SYNC: bool>(
         // Note: pass `subprocess_nn` (the `NonNull<Subprocess<'static>>`
         // captured above) instead of the live `&mut subprocess`, which would
         // alias with the `&mut subprocess.stdout` borrow held by `pipe`.
-        if let Err(err) = Readable::pipe_reader_mut(pipe).start(subprocess_nn, event_loop_nn) {
+        if let Err(err) =
+            Readable::pipe_reader_mut(pipe).start(subprocess_nn, event_loop_nn, !IS_SYNC && lazy)
+        {
             let _ = subprocess.try_kill(subprocess.kill_signal);
             let _ = global_this.throw_value(err.to_js(global_this));
             return Err(JsError::Thrown);
@@ -1751,7 +1753,9 @@ pub(crate) fn spawn_maybe_sync<const IS_SYNC: bool>(
 
     if let Readable::Pipe(pipe) = subprocess.stderr.get() {
         // Note: see stdout arm above — avoid aliased &mut.
-        if let Err(err) = Readable::pipe_reader_mut(pipe).start(subprocess_nn, event_loop_nn) {
+        if let Err(err) =
+            Readable::pipe_reader_mut(pipe).start(subprocess_nn, event_loop_nn, !IS_SYNC && lazy)
+        {
             let _ = subprocess.try_kill(subprocess.kill_signal);
             let _ = global_this.throw_value(err.to_js(global_this));
             return Err(JsError::Thrown);

src/runtime/api/bun/subprocess.rs

@@ -1020,16 +1020,22 @@ impl Subprocess<'_> {
 
         // Node.js keeps reading stdout/stderr until EOF after the direct child
         // is reaped (a grandchild may still be writing). Sync and async both
-        // resume reads here; timeout/maxBuffer bound the sync wait.
+        // resume reads here; timeout/maxBuffer bound the sync wait. A lazy
+        // reader is paused until JS pulls, so unpause it first; backpressure
+        // is moot once the direct child has exited.
         if let Readable::Pipe(pipe) = self.stdout.get() {
             if !pipe.reader.is_done() {
-                Readable::pipe_reader_mut(pipe).reader.read();
+                let reader = &mut Readable::pipe_reader_mut(pipe).reader;
+                reader.unpause();
+                reader.read();
             }
         }
 
         if let Readable::Pipe(pipe) = self.stderr.get() {
             if !pipe.reader.is_done() {
-                Readable::pipe_reader_mut(pipe).reader.read();
+                let reader = &mut Readable::pipe_reader_mut(pipe).reader;
+                reader.unpause();
+                reader.read();
             }
         }
 

src/runtime/api/bun/subprocess/SubprocessPipeReader.rs

@@ -147,18 +147,37 @@ impl PipeReader {
         &mut self,
         process: NonNull<Subprocess<'static>>,
         event_loop: NonNull<EventLoop>,
+        lazy: bool,
     ) -> bun_sys::Result<()> {
         self.r#ref();
         self.process = Some(ParentRef::from(process));
         self.event_loop = event_loop.into();
         self.event_loop_handle = bun_jsc::EventLoopHandle::init(event_loop.as_ptr().cast::<()>());
         #[cfg(windows)]
         {
+            if lazy {
+                // Leave IS_PAUSED set (the init default) so uv_read_start is
+                // deferred until JS first pulls; the kernel pipe buffer then
+                // provides backpressure and the child blocks.
+                let reader_ptr = core::ptr::from_mut(&mut self.reader).cast::<core::ffi::c_void>();
+                if let Some(source) = self.reader.source.as_mut() {
+                    source.set_data(reader_ptr);
+                }
+                self.reader
+                    .flags
+                    .remove(bun_io::pipe_reader::WindowsFlags::IS_DONE);
+                return bun_sys::Result::Ok(());
+            }
             return self.reader.start_with_current_pipe();
         }
 
         #[cfg(not(windows))]
         {
+            if lazy {
+                // Defer poll registration until JS first pulls so the kernel
+                // pipe buffer provides backpressure and the child blocks.
+                self.reader.flags.insert(PosixFlags::IS_PAUSED);
+            }
             // PosixBufferedReader.start() always returns .result, but if poll
             // registration fails it synchronously invokes onReaderError() first,
             // which drops both the Readable.pipe ref (via onCloseIO) and the ref we

src/runtime/webcore/FileReader.rs

@@ -506,6 +506,9 @@ impl FileReader {
             {
                 use bun_io::pipe_reader::PosixFlags;
                 if !was_lazy && self.reader().flags.contains(PosixFlags::POLLABLE) {
+                    // A from_pipe() reader may arrive with IS_PAUSED set (lazy
+                    // subprocess stdio); clear it so read() does not no-op.
+                    self.reader().unpause();
                     self.reader().read();
                 }
             }

test/js/bun/spawn/spawn.test.ts

@@ -453,7 +453,6 @@ for (let [gcTick, label] of [
             stdout: "pipe",
             stdin: new Blob([hugeString + "\n"]),
             stderr: "inherit",
-            lazy: true,
           });
         }
 
@@ -530,6 +529,31 @@ for (let [gcTick, label] of [
           });
         }
 
+        it.skipIf(isWindows)("lazy: true releases an unread pipe after the child exits", async () => {
+          // With lazy the reader is paused until JS pulls. If a slot is never
+          // read, on_process_exit must still drain it so the fd and the
+          // Subprocess wrapper are released.
+          const refs: WeakRef<any>[] = [];
+          for (let i = 0; i < 50; i++) {
+            const p = spawn({
+              cmd: ["sh", "-c", "echo out; echo err >&2"],
+              stdout: "pipe",
+              stderr: "pipe",
+              lazy: true,
+            });
+            expect(await p.stdout.text()).toBe("out\n");
+            await p.exited;
+            refs.push(new WeakRef(p));
+          }
+          Bun.gc(true);
+          await Bun.sleep(0);
+          Bun.gc(true);
+          const alive = refs.filter(r => r.deref() !== undefined).length;
+          // Allow a couple of stragglers for GC timing; the regression kept
+          // all 50 Strong-rooted.
+          expect(alive).toBeLessThan(5);
+        });
+
         it("should allow reading stdout after a few milliseconds", async () => {
           for (let i = 0; i < 50; i++) {
             const proc = Bun.spawn({

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

@@ -1,7 +1,7 @@
 import { semver, write } from "bun";
 import { afterAll, beforeEach, describe, expect, it } from "bun:test";
 import fs from "fs";
-import { bunEnv, bunExe, isLinux, isWindows, nodeExe, runBunInstall, shellExe, tmpdirSync } from "harness";
+import { bunEnv, bunExe, isLinux, isPosix, isWindows, nodeExe, runBunInstall, shellExe, tmpdirSync } from "harness";
 import { ChildProcess, exec, execFile, execFileSync, execSync, fork, spawn, spawnSync } from "node:child_process";
 import { once } from "node:events";
 import { promisify } from "node:util";
@@ -920,3 +920,55 @@ console.log(JSON.stringify({ uid: process.getuid(), threwCode: thrown?.code, thr
     expect(r.error?.code).toBe("ENOTSUP");
   });
 });
+
+// Regression: Bun registered the stdout/stderr poll immediately, so the native
+// reader drained the child's output into an unbounded in-memory buffer before
+// any JS consumer attached. The child never blocked on a full pipe, and once
+// 'exit' fired the autoResume path discarded the entire buffered output, so a
+// late reader received 0 bytes. With kernel backpressure the child blocks at
+// the pipe buffer until JS starts reading, matching Node.
+describe.skipIf(!isPosix)("stdout pipe backpressure", () => {
+  it("blocks the child until a reader attaches and delivers every byte", async () => {
+    const SIZE = 1024 * 1024;
+    const c = spawn("sh", ["-c", `head -c ${SIZE} /dev/zero`], {
+      stdio: ["ignore", "pipe", "ignore"],
+      env: bunEnv,
+    });
+    try {
+      // Give the event loop time to do whatever eager draining it would do
+      // without backpressure. Deadline-polled: breaks early if the child
+      // manages to exit.
+      const deadline = Date.now() + 1000;
+      while (c.exitCode === null && Date.now() < deadline) {
+        await new Promise(r => setImmediate(r));
+      }
+
+      // SIZE is larger than the kernel socket buffer, so the child cannot
+      // have finished writing without the parent reading.
+      expect(c.exitCode).toBeNull();
+
+      // Attach late and count every byte. Previously this reported 0.
+      let got = 0;
+      c.stdout!.on("data", chunk => {
+        got += chunk.length;
+      });
+      await once(c.stdout!, "end");
+      expect(got).toBe(SIZE);
+
+      await once(c, "close");
+      expect(c.exitCode).toBe(0);
+    } finally {
+      c.kill();
+    }
+  });
+
+  it("still drains a paused stdout to 'close' after the child exits", async () => {
+    const c = spawn("sh", ["-c", "echo hello"], {
+      stdio: ["ignore", "pipe", "ignore"],
+      env: bunEnv,
+    });
+    c.stdout!.pause();
+    await once(c, "close");
+    expect(c.exitCode).toBe(0);
+  });
+});