diff --git a/src/runtime/shell/IOReader.rs b/src/runtime/shell/IOReader.rs index cf9919704e8d..aacf36adafb5 100644 --- a/src/runtime/shell/IOReader.rs +++ b/src/runtime/shell/IOReader.rs @@ -42,11 +42,8 @@ pub(crate) type ReaderImpl = bun_io::BufferedReader; struct State { fd: Fd, buf: Vec, + /// 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, evtloop: EventLoopHandle, #[cfg(windows)] is_reading: bool, @@ -89,12 +86,11 @@ 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. + // Not called from the callback bodies below: `WindowsBufferedReader` + // (and `PosixBufferedReader::start()` on a synchronous registration + // failure) invokes them from under a `&mut ReaderImpl`. The POSIX poll + // dispatches hold no borrow (raw pointer, copied vtable), so a command + // started from a done/error notification may `start()` a new read. unsafe { &mut *self.reader.get() } } @@ -127,7 +123,6 @@ impl IOReader { fd, buf: Vec::new(), readers: Readers::new(), - raw_err: None, evtloop, #[cfg(windows)] is_reading: false, @@ -198,12 +193,10 @@ 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 { + // Not `is_registered()`: a finished read (EOF or error) leaves the + // one-shot poll registered but fired, so a listener added after it + // needs a new read, which reads the fd again (EOF again on a pipe). + if !r.has_pending_read() { let fd = self.state().fd; if let Err(e) = r.start(fd, true) { self.on_reader_error(&e); @@ -271,21 +264,9 @@ impl IOReader { let should_continue = has_more != bun_io::ReadState::Eof; 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 / - // `uv_read_start` for streams) — but `startWithCurrentPipe()` had - // a SECOND load-bearing side effect: `buffer().clearRetainingCapacity()`, - // which keeps `WindowsBufferedReader._buffer` bounded between - // chunks. That clear is now performed by - // `WindowsBufferedReader::on_read` after the streaming chunk is - // consumed, so we still do nothing here. + // No re-arm here (none is allowed on Windows, see `reader()`): the + // caller re-arms once we return (on posix from the `bool` below), + // and `WindowsBufferedReader::on_read` clears the chunk buffer. } should_continue } @@ -295,12 +276,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 = 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 +292,20 @@ impl IOReader { // Hold a strong ref across the body. let _keepalive = self.keepalive(); self.set_reading(false); - let s = self.state(); - let readers: Vec = 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)); } } + /// The listeners of the read that just finished. Taken out before they are + /// notified: a notification can synchronously start the next `cat`, which + /// registers for (and starts) a new read, and a notified entry left behind + /// would be notified again later, by then under a recycled `NodeId`. + 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!( diff --git a/test/js/bun/shell/commands/cat.test.ts b/test/js/bun/shell/commands/cat.test.ts new file mode 100644 index 000000000000..785dc46b59bb --- /dev/null +++ b/test/js/bun/shell/commands/cat.test.ts @@ -0,0 +1,194 @@ +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. On Windows +// the builtin is the default and the flag does nothing. +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=` 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 (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. +describe("cat (builtin) sharing one stdin reader", () => { + // On Windows the reader closes its libuv source at EOF, so starting a new + // read there needs the separate fix in #29986. + describe.skipIf(isWindows)("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 }); + }); + + // On a pipe the new read only reports EOF again, which looks the same as + // completing the second cat on the spot. A tty's EOF (^D) is used up by the + // read that sees it, so here the second cat only finishes if it really reads + // the fd again and gets the input typed after the first cat is done. + test.concurrent("stdin is a tty: the second cat reads the input typed for it", async () => { + let output = ""; + const separator = Promise.withResolvers(); + const trailer = Promise.withResolvers(); + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", childCode("cat; echo ---; cat", false)], + env: builtinEnv, + terminal: { + data(_, chunk) { + output += Buffer.from(chunk).toString(); + if (output.includes("---\n")) separator.resolve(); + if (/exit=\d+\n/.test(output)) trailer.resolve(); + }, + // Fires once the exited child's output has all been delivered, so a + // child that dies early fails the await below instead of timing out. + // (A no-op for whichever promise the output above already resolved.) + exit() { + const error = new Error(`child exited early, output so far: ${JSON.stringify(output)}`); + (output.includes("---\n") ? trailer : separator).reject(error); + }, + }, + }); + await using terminal = proc.terminal!; + // Same values on Linux and macOS. Without these the typed input would be + // echoed into `output` and the child's "\n" would come back as "\r\n". + // The child has nothing to read yet, so nothing has been output either. + const ECHO = 0x8; + const OPOST = 0x1; + terminal.localFlags &= ~ECHO; + terminal.outputFlags &= ~OPOST; + terminal.write("hi\n\x04"); + await separator.promise; + terminal.write("more\n\x04"); + await trailer.promise; + expect(output).toBe("hi\n---\nmore\nexit=0\n"); + expect(await proc.exited).toBe(0); + }); + + // The first cat fails its stdout write from inside the read that delivered + // the chunk and unregisters itself. With captured output the rest of the + // script runs before that read continues: the second cat attaches to it + // (re-registering the poll that is being serviced) and is served by its + // EOF; a third cat, started from the second one's EOF notification, is + // served by the wakeup that re-registration produces; with a subprocess + // after the second cat instead, that wakeup finds nobody to notify. With + // stdout going through an IOWriter, `echo` completes later, so the read + // reaches EOF with nobody listening and the second cat starts a new read. + describe.if(isLinux)("first cat unregistering mid-read", () => { + const first = "cat > /dev/full || echo first-failed"; + test.concurrent.each([ + [`${first}; cat && echo second-ok`, true, "first-failed\nsecond-ok\n"], + [`${first}; cat && echo second-ok; cat && echo third-ok`, true, "first-failed\nsecond-ok\nthird-ok\n"], + [`${first}; cat && echo second-ok; /bin/true`, true, "first-failed\nsecond-ok\n"], + [`${first}; cat && echo second-ok`, false, "first-failed\nsecond-ok\n"], + ])("%s (quiet: %p)", async (script, quiet, expected) => { + const result = await runScript(script, { input: "hi\n" }, { quiet }); + expect(result).toEqual({ stdout: `${expected}exit=0\n`, stderr: "", exitCode: 0 }); + }); + }); + }); + + // Starting a new read after a failed one already worked everywhere; what + // these pin down is that its failure is reported to the second cat only. This + // block runs on Windows too. + 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 }); + }); + }); +});