child_process: apply kernel backpressure to stdout/stderr pipes

src/io/PipeReader.rs

@@ -439,7 +439,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,25 @@ 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.readableFlowing === null
+      ) {
         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.readableFlowing === null
+      ) {
         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/SubprocessPipeReader.rs

@@ -147,18 +147,38 @@
         &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,
           });
         }
 

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,45 @@ 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();
+    }
+  });
+});