shell(cat): finish after a read error instead of cancelling the queued output and hanging

src/runtime/shell/builtin/cat.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use crate::shell::ExitCode;
-use crate::shell::builtin::{Builtin, BuiltinIO, BuiltinInput, BuiltinState, IoKind, Kind};
+use crate::shell::builtin::{Builtin, BuiltinInput, BuiltinState, IoKind, Kind};
 use crate::shell::interpreter::{
     FlagParser, Interpreter, NodeId, ParseFlagResult, parse_flags, shell_openat, unsupported_flag,
 };
@@ -14,6 +14,12 @@ pub struct Cat {
     pub(crate) state: CatState,
 }
 
+/// An input is finished once the reader has reported done (`in_done`, with the
+/// read error's errno in `errno`, 0 on EOF) and every stdout chunk queued from
+/// it has completed (`chunks_done >= chunks_queued`); whichever callback
+/// observes both acts on `errno`. Queued chunks are never cancelled on a read
+/// error: a cancelled chunk completes without calling back, so nothing would
+/// be left to finish the command.
 #[derive(Default)]
 pub enum CatState {
     #[default]
@@ -22,6 +28,7 @@ pub enum CatState {
         in_done: bool,
         chunks_queued: usize,
         chunks_done: usize,
+        /// Exit code once the queued chunks have drained.
         errno: ExitCode,
     },
     ExecFilepathArgs {
@@ -33,8 +40,11 @@ pub enum CatState {
         reader: Option<Arc<IOReader>>,
         chunks_queued: usize,
         chunks_done: usize,
-        out_done: bool,
         in_done: bool,
+        /// Non-zero once the current file failed to read: the command ends
+        /// with it as the exit code once the queued chunks have drained,
+        /// instead of moving on to the next file.
+        errno: ExitCode,
     },
     WaitingWriteErr,
 }
@@ -79,8 +89,8 @@ impl Cat {
                 reader: None,
                 chunks_queued: 0,
                 chunks_done: 0,
-                out_done: false,
                 in_done: false,
+                errno: 0,
             }
         };
 
@@ -206,14 +216,14 @@ impl Cat {
                     chunks_done,
                     chunks_queued,
                     in_done,
-                    out_done,
+                    errno,
                     ..
                 } = &mut Self::state_mut(interp, cmd).state
                 {
                     *chunks_done = 0;
                     *chunks_queued = 0;
                     *in_done = false;
-                    *out_done = false;
+                    *errno = 0;
                     *slot = Some(Arc::clone(&reader));
                 }
                 reader.add_reader(ReaderChildPtr {
@@ -271,11 +281,11 @@ impl Cat {
                 chunks_queued,
                 chunks_done,
                 in_done,
-                ..
+                errno,
             } => {
                 *chunks_done += 1;
                 if *in_done && *chunks_done >= *chunks_queued {
-                    Step::Done(0)
+                    Step::Done(*errno)
                 } else {
                     Step::Suspend
                 }
@@ -284,17 +294,18 @@ impl Cat {
                 chunks_queued,
                 chunks_done,
                 in_done,
-                out_done,
+                errno,
+                reader,
                 ..
             } => {
                 *chunks_done += 1;
-                if *chunks_done >= *chunks_queued {
-                    *out_done = true;
-                }
-                if *in_done && *out_done {
-                    Step::Next
-                } else {
+                if !*in_done || *chunks_done < *chunks_queued {
                     Step::Suspend
+                } else if *errno != 0 {
+                    *reader = None;
+                    Step::Done(*errno)
+                } else {
+                    Step::Next
                 }
             }
             CatState::WaitingWriteErr => Step::Done(1),
@@ -338,8 +349,6 @@ impl Cat {
         err: Option<bun_sys::SystemError>,
     ) -> Yield {
         let errno: ExitCode = err.map(|e| e.get_errno() as ExitCode).unwrap_or(0);
-        let stdout_needs_io = Builtin::of(interp, cmd).stdout.needs_io().is_some();
-        let mut cancel = false;
         let step = match &mut Self::state_mut(interp, cmd).state {
             CatState::ExecStdin {
                 chunks_queued,
@@ -349,15 +358,8 @@ impl Cat {
             } => {
                 *st_errno = errno;
                 *in_done = true;
-                if errno != 0 {
-                    if *chunks_done >= *chunks_queued || !stdout_needs_io {
-                        Step::Done(errno)
-                    } else {
-                        cancel = true;
-                        Step::Suspend
-                    }
-                } else if *chunks_done >= *chunks_queued || !stdout_needs_io {
-                    Step::Done(0)
+                if *chunks_done >= *chunks_queued {
+                    Step::Done(errno)
                 } else {
                     Step::Suspend
                 }
@@ -366,34 +368,23 @@ impl Cat {
                 chunks_queued,
                 chunks_done,
                 in_done,
-                out_done,
+                errno: st_errno,
                 reader,
                 ..
             } => {
+                *st_errno = errno;
                 *in_done = true;
-                if errno != 0 {
-                    if *out_done || !stdout_needs_io {
-                        // Drop the reader ref.
-                        *reader = None;
-                        Step::Done(errno)
-                    } else {
-                        cancel = true;
-                        Step::Suspend
-                    }
-                } else if *out_done || *chunks_done >= *chunks_queued || !stdout_needs_io {
-                    Step::Next
-                } else {
+                if *chunks_done < *chunks_queued {
                     Step::Suspend
+                } else if errno != 0 {
+                    *reader = None;
+                    Step::Done(errno)
+                } else {
+                    Step::Next
                 }
             }
             CatState::WaitingWriteErr | CatState::Idle => Step::Suspend,
         };
-        if cancel {
-            let wchild = ChildPtr::new(cmd, WriterTag::Builtin);
-            if let BuiltinIO::Fd(fd) = &Builtin::of(interp, cmd).stdout {
-                fd.writer.cancel_chunks(wchild);
-            }
-        }
         match step {
             Step::Suspend => Yield::suspended(),
             Step::Done(code) => Builtin::done(interp, cmd, code),

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

@@ -0,0 +1,116 @@
+import type { FileSink } from "bun";
+import { dlopen, FFIType } from "bun:ffi";
+import { describe, expect, test } from "bun:test";
+import { bunEnv, bunExe, isLinux, isPosix, libcPathForDlopen, tempDir } from "harness";
+import { closeSync, writeSync } from "node:fs";
+
+// On POSIX the shell only runs its own `cat` when this flag is set (see
+// `Kind::DISABLED_ON_POSIX`); otherwise it spawns the system binary. The
+// scripts run in a child so the flag applies.
+const builtinEnv = { ...bunEnv, BUN_ENABLE_EXPERIMENTAL_SHELL_BUILTINS: "1" };
+
+// Unless `quiet` is set, cat's stdout is the shell's IOWriter on the child's
+// stdout pipe: a chunk cat reads stays queued there until the event loop
+// reaches the writable poll, which is the window the read-error paths under
+// test fire in. `r.stdout` is a tee of the same bytes, so the child's stdout
+// ends up as cat's own output followed by the report line. With `quiet`
+// nothing is queued (output goes straight into the capture buffer).
+function spawnReport(command: string, options: { quiet?: boolean; stdin?: number | "pipe"; cwd?: string } = {}) {
+  return Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      /* js */ `
+        import { $ } from "bun";
+        const r = await $\`${command}\`${options.quiet ? ".quiet()" : ""}.nothrow();
+        console.log(JSON.stringify({ exitCode: r.exitCode, stdout: r.stdout.toString(), stderr: r.stderr.toString() }));
+      `,
+    ],
+    env: builtinEnv,
+    cwd: options.cwd,
+    stdin: options.stdin ?? "ignore",
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+}
+
+function report(exitCode: number, stdout: string): string {
+  return JSON.stringify({ exitCode, stdout, stderr: "" }) + "\n";
+}
+
+const EIO = 5;
+
+// On Linux, once the slave side of a pty is closed, read() on the master
+// returns whatever the slave wrote and then fails with EIO. Handing the master
+// to the child as its stdin is a way to feed the builtin a genuine read error
+// preceded by data: cat gets both in the same poll wake.
+function openptyMasterWithClosedSlave(payload: string): number {
+  const { openpty } = dlopen(libcPathForDlopen(), {
+    openpty: {
+      args: [FFIType.ptr, FFIType.ptr, FFIType.ptr, FFIType.ptr, FFIType.ptr],
+      returns: FFIType.i32,
+    },
+  }).symbols;
+  const master = new Int32Array(1);
+  const slave = new Int32Array(1);
+  expect(openpty(master, slave, null, null, null)).toBe(0);
+  // No newline: the pty's output processing would turn it into "\r\n".
+  writeSync(slave[0], payload);
+  closeSync(slave[0]);
+  return master[0];
+}
+
+describe.concurrent("cat (builtin)", () => {
+  test.skipIf(!isPosix)("copies stdin to stdout until EOF", async () => {
+    await using proc = spawnReport("cat", { stdin: "pipe" });
+    const stdin = proc.stdin as FileSink;
+    stdin.write("piped in\n");
+    await stdin.end();
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout, stderr }).toEqual({ stdout: "piped in\n" + report(0, "piped in\n"), stderr: "" });
+    expect(exitCode).toBe(0);
+  });
+
+  test.skipIf(!isLinux)("read error with nothing queued: exits with the errno", async () => {
+    const master = openptyMasterWithClosedSlave("read before the error");
+    await using proc = spawnReport("cat", { quiet: true, stdin: master });
+    // The child holds its own copy of the master.
+    closeSync(master);
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout, stderr }).toEqual({ stdout: report(EIO, "read before the error"), stderr: "" });
+    expect(exitCode).toBe(0);
+  });
+
+  // Same error, but with the data still queued on stdout. This used to cancel
+  // the queued chunk and suspend; a cancelled chunk completes without calling
+  // back into cat, so the command never finished, the `$` promise never
+  // settled, and the data was dropped.
+  test.skipIf(!isLinux)("read error with data still queued: flushes it, then exits with the errno", async () => {
+    const master = openptyMasterWithClosedSlave("read before the error");
+    await using proc = spawnReport("cat", { stdin: master });
+    closeSync(master);
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    expect({ stdout, stderr }).toEqual({
+      stdout: "read before the error" + report(EIO, "read before the error"),
+      stderr: "",
+    });
+    expect(exitCode).toBe(0);
+  });
+
+  // File-argument state. A directory opens but cannot be read (Linux already
+  // refuses to register it with epoll), so the reader fails before anything is
+  // queued. With stdout on an fd this used to suspend forever too: finishing
+  // was gated on a flag that only a completed chunk could set.
+  test.skipIf(!isPosix)("unreadable file argument exits with the errno instead of hanging", async () => {
+    using dir = tempDir("shell-cat-dir-arg", { "sub/.keep": "" });
+    await using proc = spawnReport("cat sub", { cwd: String(dir) });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    const parsed = JSON.parse(stdout);
+    expect({ parsed, stderr }).toEqual({
+      parsed: { exitCode: expect.any(Number), stdout: "", stderr: "" },
+      stderr: "",
+    });
+    expect(parsed.exitCode).toBeGreaterThan(0);
+    expect(exitCode).toBe(0);
+  });
+});