shell(cat): restart the shared stdin reader for a later cat, notify each listener once

src/runtime/shell/IOReader.rs

@@ -42,11 +42,8 @@ pub(crate) type ReaderImpl = bun_io::BufferedReader;
 struct State {
     fd: Fd,
     buf: Vec<u8>,
+    /// Listeners of the read cycle currently in flight; see `take_readers`.
     readers: Readers,
-    /// The raw `sys::Error`. `SystemError` is not `Clone`
-    /// in the Rust port yet, so we keep the source error to re-derive a fresh
-    /// `SystemError` per callee in `on_reader_done_cb`.
-    raw_err: Option<sys::Error>,
     evtloop: EventLoopHandle,
     #[cfg(windows)]
     is_reading: bool,
@@ -89,12 +86,17 @@ impl IOReader {
         // held by the bun_io read loop never overlaps a `&mut State` derived in a
         // vtable callback (see struct doc comment).
         //
-        // MUST NOT be invoked from within a `BufferedReaderParent` vtable
-        // callback (`on_read_chunk_cb`/`on_reader_done_cb`/`on_reader_error`):
-        // the read loop already holds a live `&mut ReaderImpl` on its stack
-        // while the callback runs (PipeReader.rs aliasing contract), so
-        // re-deriving here would create two simultaneous `&mut` to the same
-        // BufferedReader = Stacked-Borrows UB.
+        // The `BufferedReaderParent` callback bodies below (`on_read_chunk_cb`/
+        // `on_reader_done_cb`/`on_reader_error`) must not call this: a `&mut
+        // ReaderImpl` can be live on the stack while they run, always on
+        // Windows (`WindowsBufferedReader::on_read` dispatches them from
+        // `&mut self`) and on POSIX when `PosixBufferedReader::start()`
+        // reports a registration failure synchronously. The poll-driven POSIX
+        // dispatches (`on_poll` read loops, `done()`/`on_error()` in tail
+        // position) go through a copied vtable and a raw pointer with no
+        // borrow of the reader live, which is what lets a command that the
+        // trampoline starts from inside `on_reader_done_cb`/`on_reader_error`
+        // call `start()` and arm the next read.
         unsafe { &mut *self.reader.get() }
     }
 
@@ -127,7 +129,6 @@ impl IOReader {
                 fd,
                 buf: Vec::new(),
                 readers: Readers::new(),
-                raw_err: None,
                 evtloop,
                 #[cfg(windows)]
                 is_reading: false,
@@ -198,12 +199,12 @@ impl IOReader {
         #[cfg(not(windows))]
         {
             let r = self.reader();
-            let need_start = match &r.handle {
-                bun_io::pipes::PollOrFd::Closed => true,
-                bun_io::pipes::PollOrFd::Poll(p) => !p.is_registered(),
-                bun_io::pipes::PollOrFd::Fd(_) => true,
-            };
-            if need_start {
+            // A finished cycle (EOF or error) leaves the one-shot poll fired
+            // and not re-armed: still `is_registered()`, but it will never
+            // fire again. `has_pending_read()` is false then, so a listener
+            // added after that cycle gets a new one (which reads EOF again on
+            // a pipe, or whatever the fd has to offer now).
+            if !r.has_pending_read() {
                 let fd = self.state().fd;
                 if let Err(e) = r.start(fd, true) {
                     self.on_reader_error(&e);
@@ -272,14 +273,13 @@ impl IOReader {
         if should_continue && !self.state().readers.is_empty() {
             self.set_reading(true);
             // NOTE: no explicit re-arm (`registerPoll()` on posix /
-            // `startWithCurrentPipe()` on windows) here: that would re-derive
-            // a second `&mut ReaderImpl` while the bun_io read loop still
-            // holds one on its stack (PipeReader.rs aliasing contract) —
-            // Stacked-Borrows UB.
-            // On posix the re-arm is redundant: the read loop re-registers
-            // itself after the callback returns based on the `bool` we return
-            // (PipeReader.rs:731/755/846/920/986). On Windows the re-arm is
-            // also handled by the caller (`on_file_read`'s defer block /
+            // `startWithCurrentPipe()` on windows) here: on Windows this
+            // callback runs from under `WindowsBufferedReader::on_read`'s
+            // `&mut self` (see `reader()`), and it is not needed anyway.
+            // On posix the read loop re-registers itself after the callback
+            // returns based on the `bool` we return (the `register_poll` calls
+            // at the end of the PipeReader.rs read loops). On Windows the
+            // re-arm is also handled by the caller (`on_file_read`'s epilogue /
             // `uv_read_start` for streams) — but `startWithCurrentPipe()` had
             // a SECOND load-bearing side effect: `buffer().clearRetainingCapacity()`,
             // which keeps `WindowsBufferedReader._buffer` bounded between
@@ -295,12 +295,8 @@ impl IOReader {
         // alive across the loop.
         let _keepalive = self.keepalive();
         self.set_reading(false);
-        let s = self.state();
-        s.raw_err = Some(err.clone());
-        // NOTE: reshaped for borrowck — copy out before dispatching.
-        let readers: Vec<ChildPtr> = s.readers.clone();
-        let interp = s.interp;
-        for r in readers {
+        let interp = self.state().interp;
+        for r in self.take_readers() {
             // Re-derive a fresh SystemError per callee (see
             // IOWriter.on_error note).
             let ee = err.to_shell_system_error();
@@ -315,19 +311,22 @@ impl IOReader {
         // Hold a strong ref across the body.
         let _keepalive = self.keepalive();
         self.set_reading(false);
-        let s = self.state();
-        let readers: Vec<ChildPtr> = s.readers.clone();
-        let interp = s.interp;
-        // `SystemError` isn't `Clone` yet, so we keep the source `sys::Error`
-        // (which IS `Clone`) and re-derive a fresh `SystemError` per callee —
-        // same approach as `on_reader_error`.
-        let raw_err = s.raw_err.clone();
-        for r in readers {
-            let ee = raw_err.as_ref().map(|e| e.to_shell_system_error());
-            self.run_yield(dispatch_reader_done(r, ee, interp));
+        let interp = self.state().interp;
+        for r in self.take_readers() {
+            self.run_yield(dispatch_reader_done(r, None, interp));
         }
     }
 
+    /// Detaches the listeners of the cycle that just ended before notifying
+    /// them. Notifying one can synchronously run the rest of the script: a
+    /// `cat` started there registers into the emptied list and restarts the
+    /// reader, so it is notified by its own cycle rather than by this one, and
+    /// nothing stays behind to be notified again by a later cycle under a
+    /// `NodeId` that has been freed or recycled by then.
+    fn take_readers(&self) -> Readers {
+        core::mem::take(&mut self.state().readers)
+    }
+
     fn run_yield(&self, y: Yield) {
         let Some(interp) = self.state().interp else {
             debug_assert!(

test/js/bun/shell/commands/cat.test.ts

@@ -0,0 +1,144 @@
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe, isLinux, isWindows } from "harness";
+import { closeSync, openSync } from "node:fs";
+
+// On POSIX the builtin `cat` is only used with this flag set (see
+// `Kind::DISABLED_ON_POSIX`); without it `cat` is the system binary.
+const builtinEnv = { ...bunEnv, BUN_ENABLE_EXPERIMENTAL_SHELL_BUILTINS: "1" };
+
+// Code for a child bun that runs `script` and prints the script's stdout
+// followed by an `exit=<code>` trailer. A hang or a crash in the child shows up
+// as a missing trailer. (Both go through process.stdout: console.log takes a
+// separate path to fd 1 and can overtake a large pending process.stdout.write.)
+function childCode(script: string, quiet: boolean): string {
+  const run = `Bun.$\`\${{ raw: ${JSON.stringify(script)} }}\`.nothrow()`;
+  const trailer = `process.stdout.write("exit=" + r.exitCode + "\\n");`;
+  return quiet
+    ? `const r = await ${run}.quiet(); process.stdout.write(r.stdout); ${trailer}`
+    : `const r = await ${run}; ${trailer}`;
+}
+
+type Stdin =
+  // Written to a pipe that is closed right away, so stdin reaches EOF.
+  | { input: string }
+  // Reading a directory fails, so every `cat` reading stdin fails.
+  | { directory: string };
+
+function spawnChild(script: string, stdin: Stdin, quiet: boolean) {
+  const cmd = [bunExe(), "-e", childCode(script, quiet)];
+  if ("directory" in stdin) {
+    const fd = openSync(stdin.directory, "r");
+    try {
+      return Bun.spawn({ cmd, env: builtinEnv, stdin: fd, stdout: "pipe", stderr: "pipe" });
+    } finally {
+      closeSync(fd);
+    }
+  }
+  const proc = Bun.spawn({ cmd, env: builtinEnv, stdin: "pipe", stdout: "pipe", stderr: "pipe" });
+  proc.stdin.write(stdin.input);
+  proc.stdin.end();
+  return proc;
+}
+
+async function runScript(script: string, stdin: Stdin, { quiet = true } = {}) {
+  await using proc = spawnChild(script, stdin, quiet);
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  return { stdout, stderr, exitCode };
+}
+
+// Every `cat` reading the script's stdin (or a pipeline stage's stdin)
+// registers on the single IOReader owned by that fd. Once the first `cat` has
+// consumed a read cycle (EOF or error), a later `cat` on the same fd has to
+// start a new one and must be the only listener notified by it.
+//
+// Skipped on Windows, where the reader closes its libuv source at EOF; a second
+// `cat` on the same fd there is covered by the fix in #29986.
+describe.skipIf(isWindows)("cat (builtin) sharing one stdin reader", () => {
+  describe("after the first cat reached EOF", () => {
+    const scripts: [script: string, stdout: string][] = [
+      ["cat; echo ---; cat", "hi\n---\n"],
+      ["cat && echo --- && cat", "hi\n---\n"],
+      // The second restart starts from a reader that was already restarted once.
+      ["cat; cat; cat", "hi\n"],
+      // The subshell / if node is allocated before the second cat, so the second
+      // cat does not reuse the first cat's node id: a listener entry left over
+      // from the first cat would be dispatched to a node that is not a cat.
+      ["cat; (echo ---; cat)", "hi\n---\n"],
+      ["cat; if true; then echo ---; cat; fi", "hi\n---\n"],
+    ];
+
+    // With captured output, the first cat's completion runs the rest of the
+    // script synchronously, so the second cat registers from inside the
+    // reader's EOF callback.
+    describe("captured stdout", () => {
+      test.concurrent.each(scripts)("%s", async (script, expected) => {
+        const result = await runScript(script, { input: "hi\n" });
+        expect(result).toEqual({ stdout: `${expected}exit=0\n`, stderr: "", exitCode: 0 });
+      });
+    });
+
+    // With stdout going through an IOWriter, a command can also complete from a
+    // write callback, after the EOF callback has returned.
+    describe("inherited stdout", () => {
+      test.concurrent.each(scripts)("%s", async (script, expected) => {
+        const result = await runScript(script, { input: "hi\n" }, { quiet: false });
+        expect(result).toEqual({ stdout: `${expected}exit=0\n`, stderr: "", exitCode: 0 });
+      });
+    });
+
+    test.concurrent("input spanning several reads", async () => {
+      const input = Buffer.alloc(300_000, "abcdefghij\n").toString();
+      const result = await runScript("cat; cat", { input });
+      expect(result.stderr).toBe("");
+      expect(result.stdout.length).toBe(input.length + "exit=0\n".length);
+      expect(result.stdout).toBe(`${input}exit=0\n`);
+      expect(result.exitCode).toBe(0);
+    });
+
+    test.concurrent("stdin of a pipeline stage", async () => {
+      const result = await runScript("echo hi | (cat; echo ---; cat)", { input: "" });
+      expect(result).toEqual({ stdout: "hi\n---\nexit=0\n", stderr: "", exitCode: 0 });
+    });
+
+    // Bun.spawn's stdin pipe is a socketpair; this is the same thing over an
+    // actual pipe.
+    test.concurrent("stdin is a pipe", async () => {
+      await using proc = Bun.spawn({
+        cmd: ["sh", "-c", 'printf "hi\\n" | "$0" -e "$1"', bunExe(), childCode("cat; echo ---; cat", true)],
+        env: builtinEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+      expect({ stdout, stderr, exitCode }).toEqual({ stdout: "hi\n---\nexit=0\n", stderr: "", exitCode: 0 });
+    });
+
+    // The first cat fails on its stdout write while the read that delivered the
+    // chunk is still running and unregisters itself. With captured output the
+    // second cat registers right away, while that read is still in flight, and
+    // is served by it. With stdout going through an IOWriter, `echo` completes
+    // later, so the read reaches EOF with nobody listening and the second cat
+    // registers only after that.
+    describe.if(isLinux)("first cat unregistering mid-read", () => {
+      test.concurrent.each([true, false])("quiet: %p", async quiet => {
+        const result = await runScript(
+          "cat > /dev/full || echo first-failed; cat && echo second-ok",
+          { input: "hi\n" },
+          { quiet },
+        );
+        expect(result).toEqual({ stdout: "first-failed\nsecond-ok\nexit=0\n", stderr: "", exitCode: 0 });
+      });
+    });
+  });
+
+  describe("after the first cat failed to read", () => {
+    test.concurrent.each([
+      "cat || echo first-failed; echo ---; cat || echo second-failed",
+      // Second cat in a node id different from the first cat's (see above).
+      "cat || echo first-failed; (echo ---; cat) || echo second-failed",
+    ])("%s", async script => {
+      const result = await runScript(script, { directory: import.meta.dir });
+      expect(result).toEqual({ stdout: "first-failed\n---\nsecond-failed\nexit=0\n", stderr: "", exitCode: 0 });
+    });
+  });
+});