FileReader: release the event loop while a pipe reader is stopped at its highwater mark

src/io/PipeReader.rs

@@ -4,7 +4,7 @@ use core::ptr::NonNull;
 
 use bun_sys::{self as sys, Fd};
 
-use crate::{EventLoopHandle, FilePollFlag, FilePollKind, FilePollRef, Owner, PollTag};
+use crate::{EventLoopHandle, FilePollKind, FilePollRef, Owner, PollTag};
 // `bun.Async.Loop` — on POSIX the uws `us_loop_t`, on Windows the embedded
 // `uv_loop_t` (`bun_io::Loop` is the cfg-aliased nominal that picks the
 // right one). `BufferedReaderParent::loop_` returns this so callers in T3+
@@ -239,6 +239,10 @@ impl PosixBufferedReader {
         let Some(poll) = self.handle.get_poll() else {
             return;
         };
+        // An unarmed poll delivers nothing; `try_register_poll` applies KEEP_ALIVE when it arms.
+        if value && !poll.is_watching() {
+            return;
+        }
         poll.set_keeping_process_alive(self.vtable.event_loop(), value);
     }
 
@@ -529,9 +533,8 @@ impl PosixBufferedReader {
         };
         poll.set_owner(Owner::new(PollTag::BufferedReader, owner_ptr.cast()));
 
-        if !poll.has_flag(FilePollFlag::WasEverRegistered)
-            && self.flags.contains(PosixFlags::KEEP_ALIVE)
-        {
+        // Re-applied on every arm: `pause()` unregisters, which drops it.
+        if self.flags.contains(PosixFlags::KEEP_ALIVE) {
             poll.enable_keeping_process_alive(ev);
         }
 

src/io/posix_event_loop.rs

@@ -903,22 +903,23 @@ impl FilePoll {
     ) -> sys::Result<()> {
         debug_assert!(fd.native() >= 0 && fd != INVALID_FD);
 
-        if !(self.flags.contains(Flags::PollReadable)
+        let registered = self.flags.contains(Flags::PollReadable)
             || self.flags.contains(Flags::PollWritable)
             || self.flags.contains(Flags::PollProcess)
             || self.flags.contains(Flags::PollMachport)
-            || self.flags.contains(Flags::PollMemoryPressure))
-        {
-            // no-op
+            || self.flags.contains(Flags::PollMemoryPressure);
+        // The `needs_rearm` skip below keeps the disarmed kernel registration, so teardown must still delete it.
+        let disarmed_only = !registered && self.flags.contains(Flags::NeedsRearm);
+        if !registered && !(disarmed_only && force_unregister) {
             return sys::Result::Ok(());
         }
 
-        debug_assert!(fd != INVALID_FD);
         let watcher_fd = loop_.fd;
-        let both_directions =
-            self.flags.contains(Flags::PollReadable) && self.flags.contains(Flags::PollWritable);
+        let both_directions = disarmed_only
+            || (self.flags.contains(Flags::PollReadable)
+                && self.flags.contains(Flags::PollWritable));
         let flag: Flags = 'brk: {
-            if self.flags.contains(Flags::PollReadable) {
+            if disarmed_only || self.flags.contains(Flags::PollReadable) {
                 break 'brk Flags::Readable;
             }
             if self.flags.contains(Flags::PollWritable) {

src/runtime/webcore/FileReader.rs

@@ -655,8 +655,7 @@ impl FileReader {
             // No JS read is waiting; stop at the highwater mark and let onPull restart. `started` gates it: a non-lazy `Bun.spawn` pipe is already reading before any consumer attaches, and throttling then deadlocks a child alternating stdout/stderr writes.
             let keep_going = !self.started.get()
                 || (self.flowing.get() && self.buffered.get().len() < self.highwater_mark);
-            // A completion-driven reader keeps issuing reads unless stopped; `on_pull` restarts it.
-            #[cfg(windows)]
+            // `false` only ends this read loop; `pause()` stops the reader and releases its hold on the event loop until `on_pull`.
             if !keep_going {
                 self.reader().pause();
             }
@@ -809,6 +808,8 @@ impl FileReader {
         }
 
         if !self.reader().has_pending_read() && self.flowing.get() {
+            // A consumer is pulling again: undo the highwater pause from `on_read_chunk`.
+            self.reader().unpause();
             // SAFETY: the reader cell is live for `self`'s lifetime; `read_into` is the raw re-entrancy-safe entry (EOF/error dispatch runs user JS).
             let (amount_read, state) = unsafe { IOReader::read_into(self.reader.get(), buffer) };
             bun_core::scoped_log!(FileReader, "onPull({}) = {}", buffer.len(), amount_read);

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

@@ -807,6 +807,134 @@ describe("should not hang", () => {
   }
 });
 
+// What keeps a process alive whose only handle is a child's stdout reader (the
+// child itself is unref'd): a read that can still deliver something does, a
+// reader stopped at its highwater mark does not. Stopped, the pipe reader's
+// poll is unregistered until the next pull, so it could not even observe the
+// child going away. Node: readStop() at the highWaterMark leaves the handle
+// inactive, and a pending read keeps it active.
+describe.skipIf(isWindows)("stdout reader of an unref'd child and process lifetime", () => {
+  async function run(script: string) {
+    await using proc = spawn({
+      cmd: [bunExe(), "-e", script],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect(stderr).toBe("");
+    expect(exitCode).toBe(0);
+    return stdout;
+  }
+
+  // "saturating" fills the pipe faster than the parent reads it; "trickling"
+  // lets most parent reads end in EAGAIN, the path that used to re-arm the
+  // poll regardless of the highwater stop.
+  const producers = {
+    saturating: `const chunk = Buffer.alloc(8192, 120); for (;;) require("fs").writeSync(1, chunk);`,
+    trickling: `const chunk = Buffer.alloc(2048, 120); setInterval(() => require("fs").writeSync(1, chunk), 1);`,
+  };
+  for (const [kind, producer] of Object.entries(producers)) {
+    it.concurrent(
+      `an idle reader stopped at the highwater mark does not keep the process alive (${kind} writer)`,
+      async () => {
+        const stdout = await run(`
+        const producer = Bun.spawn({
+          cmd: [process.execPath, "-e", ${JSON.stringify(producer)}],
+          stdin: "ignore",
+          stdout: "pipe",
+          stderr: "inherit",
+        });
+        const reader = producer.stdout.getReader();
+        // The running child no longer counts; only its stdout reader can keep this process alive now.
+        producer.unref();
+        let firstLength = -1;
+        process.on("exit", () => {
+          console.log(JSON.stringify({ gotFirstChunk: firstLength > 0, producerExitCode: producer.exitCode }));
+          producer.kill("SIGKILL");
+        });
+        firstLength = (await reader.read()).value.length;
+        // The reader stays locked and idle while the child keeps writing: the pipe
+        // reader fills to its highwater mark and stops, and then nothing is pending.
+      `);
+        expect(JSON.parse(stdout)).toEqual({ gotFirstChunk: true, producerExitCode: null });
+      },
+    );
+  }
+
+  it.concurrent("a pending read keeps the process alive until the child writes", async () => {
+    const child = `const fs = require("fs"); fs.readSync(0, Buffer.alloc(4)); fs.writeSync(1, "pong");`;
+    // Not top-level await: an unsettled entry-module promise keeps the process
+    // running on its own and would hide a read that does not.
+    const stdout = await run(`
+      const child = Bun.spawn({
+        cmd: [process.execPath, "-e", ${JSON.stringify(child)}],
+        stdin: "pipe",
+        stdout: "pipe",
+        stderr: "inherit",
+      });
+      const reader = child.stdout.getReader();
+      child.unref();
+      (async () => {
+        const pending = reader.read();
+        child.stdin.write("ping");
+        child.stdin.end();
+        // Only this read is left to wait for. It must hold the process until "pong" arrives.
+        const { value } = await pending;
+        console.log(new TextDecoder().decode(value));
+      })();
+    `);
+    expect(stdout).toBe("pong\n");
+  });
+
+  // node:child_process pause() stops the pipe reader (the poll is unregistered)
+  // and resume() arms it again; the re-armed poll must hold the loop like the
+  // first one did. The writer starts only once the parent is reading ("s"),
+  // overfills the paused Readable, and sends "END" after a pause the parent has
+  // to sit through idle on the pipe with nothing else pending.
+  it.concurrent("a child_process stdout resumed after pause() keeps the process alive again", async () => {
+    const writer = `
+      const fs = require("fs");
+      const chunk = Buffer.alloc(65536, "x");
+      const waitForParent = () => fs.readSync(0, Buffer.alloc(1), 0, 1, null);
+      waitForParent();
+      for (let i = 0; i < 16; i++) fs.writeSync(1, chunk);
+      waitForParent();
+      setTimeout(() => fs.writeSync(1, "END"), 100);
+    `;
+    const stdout = await run(`
+      const { spawn } = require("node:child_process");
+      const child = spawn(process.execPath, ["-e", ${JSON.stringify(writer)}], { stdio: ["pipe", "pipe", "inherit"] });
+      child.unref();
+      let received = 0;
+      let paused = false;
+      let drained = false;
+      child.stdout.on("data", chunk => {
+        received += chunk.length;
+        if (!paused) {
+          paused = true;
+          child.stdout.pause();
+          const resumeWhenFull = () => {
+            if (child.stdout.readableLength >= child.stdout.readableHighWaterMark) {
+              console.log("resume");
+              child.stdout.resume();
+            } else {
+              setImmediate(resumeWhenFull);
+            }
+          };
+          setImmediate(resumeWhenFull);
+        } else if (!drained && received >= 16 * 65536) {
+          drained = true;
+          child.stdin.end("g");
+        }
+      });
+      child.stdout.on("end", () => console.log("end " + received));
+      setImmediate(() => child.stdin.write("s"));
+    `);
+    expect(stdout).toBe("resume\nend " + (16 * 65536 + 3) + "\n");
+  });
+});
+
 describe("unref() + .exited with nothing else ref'd (Windows)", () => {
   // Windows: with only an unref'd uv_process_t left, uv_run() used to skip its
   // body and never dequeue the IOCP exit packet, so these children busy-spun

test/js/node/process/process-stdin.test.ts

@@ -305,6 +305,58 @@ test.concurrent("stdin should not allow process to exit when not paused", async
   expect(await proc.stderr.text()).toMatchInlineSnapshot(`""`);
 });
 
+// unref() drops stdin's hold on the event loop and ref() takes it back: after
+// the pair the child has nothing else pending and must still receive every
+// later byte up to EOF.
+test.concurrent("stdin.ref() after unref() keeps the process alive until EOF", async () => {
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+      let total = 0;
+      process.stdin.on("data", chunk => {
+        if (total === 0) {
+          process.stdin.unref();
+          process.stdin.ref();
+          process.stdout.write("ready\\n");
+        }
+        total += chunk.length;
+      });
+      process.stdin.on("end", () => {
+        process.stdout.write("TOTAL " + total + "\\n");
+      });
+      `,
+    ],
+    stdin: "pipe",
+    stdout: "pipe",
+    stderr: "pipe",
+    env: bunEnv,
+  });
+  const reader = proc.stdout.getReader();
+  const decoder = new TextDecoder();
+  let stdout = "";
+  proc.stdin.write("x");
+  while (!stdout.includes("ready\n")) {
+    const { value, done } = await reader.read();
+    if (done) break;
+    stdout += decoder.decode(value);
+  }
+  // The child is idle again with only stdin to wait for; these bytes must still reach it.
+  const rest = Buffer.alloc(64 * 1024, "y");
+  proc.stdin.write(rest);
+  await proc.stdin.end();
+  while (true) {
+    const { value, done } = await reader.read();
+    if (done) break;
+    stdout += decoder.decode(value);
+  }
+  const [stderr, exitCode] = await Promise.all([proc.stderr.text(), proc.exited]);
+  expect(stderr).toBe("");
+  expect(stdout).toBe(`ready\nTOTAL ${1 + rest.length}\n`);
+  expect(exitCode).toBe(0);
+});
+
 test.concurrent("a throw from a 'data' listener is an uncaughtException, and stdin keeps reading", async () => {
   await using proc = Bun.spawn({
     cmd: [
@@ -498,6 +550,70 @@ describe.skipIf(isWindows)("pipe backpressure", () => {
     expect(deltaMB).toBeLessThan(maxDeltaMB);
   });
 
+  // Stopped at the backstop, the reader's one-shot poll is left unarmed until
+  // the next read, so it can observe nothing (not even the writer going away)
+  // and must not keep the event loop alive on its own. Node behaves the same:
+  // readStop() at the highWaterMark leaves the handle inactive.
+  test.concurrent("a reader stopped at the highwater backstop does not keep the process alive", async () => {
+    const { first } = await run(`
+      const rd = Bun.stdin.stream().getReader();
+      const c = await rd.read();
+      process.stdout.write(JSON.stringify({ first: (c.value?.length ?? 0) > 0 }));
+      // No further read and no exit(): once the backstop engages nothing is pending.
+    `);
+    expect(first).toBe(true);
+  });
+
+  // The same over an anonymous pipe (the blocking-pipe read path; Bun.spawn
+  // stdio above is a socketpair). The writer keeps the pipe full until the
+  // reader is gone, so EOF can never be what lets the reader exit.
+  test.concurrent("over a shell pipe, a reader stopped at the backstop does not keep the process alive", async () => {
+    using dir = tempDir("stdin-backstop-pipe", {
+      "writer.js": `
+        const chunk = Buffer.alloc(65536, 0x78);
+        process.stdout.on("error", () => process.exit(0));
+        (function pump() {
+          while (process.stdout.write(chunk)) {}
+          process.stdout.once("drain", pump);
+        })();
+      `,
+      "reader.js": `
+        const reader = Bun.stdin.stream().getReader();
+        const { value } = await reader.read();
+        process.on("exit", () => console.log("EXIT"));
+        console.log("FIRST " + (value.byteLength > 0));
+      `,
+    });
+    const { promise, resolve } = Promise.withResolvers<{ err: Error | null; stdout: string; stderr: string }>();
+    exec(
+      `"${bunExe()}" writer.js | "${bunExe()}" reader.js`,
+      { cwd: String(dir), env: bunEnv },
+      (err, stdout, stderr) => resolve({ err, stdout, stderr }),
+    );
+    expect(await promise).toEqual({ err: null, stdout: "FIRST true\nEXIT\n", stderr: "" });
+  });
+
+  // Stopping unregisters a fired one-shot poll without a syscall, which leaves
+  // its disarmed registration in the kernel. Cancelling the stream must still
+  // take it out: fd 0 stays open, and the next poll on it would hit EEXIST.
+  test.concurrent("a reader cancelled at the highwater backstop frees fd 0 for the next reader", async () => {
+    const { second } = await run(`
+      const { getEventLoopStats } = require("bun:internal-for-testing");
+      // Not top-level await: an unsettled entry-module promise would keep the process alive by itself.
+      (async () => {
+        const reader = Bun.stdin.stream().getReader();
+        await reader.read();
+        // The parent keeps the pipe full, so the reader soon stops and lets go of the loop.
+        while (getEventLoopStats().loopActive) await new Promise(resolve => setImmediate(resolve));
+        await reader.cancel();
+        const next = await Bun.file(0).stream().getReader().read();
+        process.stdout.write(JSON.stringify({ second: (next.value?.length ?? 0) > 0 }));
+        process.exit(0);
+      })();
+    `);
+    expect(second).toBe(true);
+  });
+
   test.concurrent("reading resumes after the highwater backstop", async () => {
     // Stop reading long enough for the backstop to engage, then drain to EOF
     // and make sure every byte written by the parent is delivered.