From 89f30ec5f1c9307b2d8f1137b4aff51570eadfef Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 22 Aug 2026 02:46:55 +0000 Subject: [PATCH 1/7] shell: stop reading a subprocess's output once relaying it to stdout/stderr fails When the shell's stdout or stderr writer fails (EPIPE, the reader of the process's stdout is gone), CapturedWriter recorded the error but the PipeReader kept reading the child's pipe into buffered_output for as long as the child kept writing. The command never finished, so the awaited `$` never settled, and the child was never told that its output goes nowhere. PipeReader::on_capture_failed closes the shell's end of the pipe while the child still has its end open and settles the reader's state, so the existing try_signal_done_to_cmd path reports the relay error to the Cmd right away and the child's next write fails, as it would under a real shell. The rejection can also arrive synchronously, from inside the reader's own on_read_chunk (the writer is already dead when the chunk is enqueued). The Cmd then drops its Arc while the bun_io read loop is still on the stack, so the reader now gives the loop a ref_/deref keepalive, like IOReader does, and on_read_chunk / do_write no longer hold a borrow of the PipeReader across the trampoline. --- src/runtime/shell/states/Cmd.rs | 6 +- src/runtime/shell/subproc.rs | 140 +++++++++++++++++------ test/js/bun/shell/epipe-relay-fixture.ts | 74 ++++++++++++ test/js/bun/shell/epipe.test.ts | 58 +++++++++- 4 files changed, 237 insertions(+), 41 deletions(-) create mode 100644 test/js/bun/shell/epipe-relay-fixture.ts diff --git a/src/runtime/shell/states/Cmd.rs b/src/runtime/shell/states/Cmd.rs index 6057a967e208..334f6da5f0ec 100644 --- a/src/runtime/shell/states/Cmd.rs +++ b/src/runtime/shell/states/Cmd.rs @@ -72,7 +72,7 @@ pub struct SubprocExec { /// (`buffered_output_close` / `on_exit`) can hand a [`Yield`] back to the /// trampoline. The `Cmd` lives inside `interp.nodes`, so we stash the /// indices and - /// return `Yield::Next(this_id)` for the caller (`PipeReader::run_yield`) + /// return `Yield::Next(this_id)` for the caller (`PipeReader::run_yield_with`) /// to drive. pub(crate) interp: *mut Interpreter, pub(crate) this_id: NodeId, @@ -922,7 +922,7 @@ impl Cmd { // `ShellSubprocess` / `PipeReader` hold a `*mut Cmd` backref and call // these via `&mut self`. The NodeId-arena port stashes `(interp, this_id)` // on `SubprocExec` so the resulting `Yield` can be driven by the caller's - // `PipeReader::run_yield` without aliasing `&Interpreter` against + // `PipeReader::run_yield_with` without aliasing `&Interpreter` against // `&mut self`. /// True once the command has both an exit code and (for subprocesses) @@ -960,7 +960,7 @@ impl Cmd { } if self.has_finished() { // Set `state = Done` and hand the Yield back to the caller - // (`PipeReader::run_yield`), which drives the trampoline with the + // (`PipeReader::run_yield_with`), which drives the trampoline with the // `*mut Interpreter` it already holds, landing in `Cmd::next` → // `CmdState::Done` → `interp.child_done(...)`. self.state = CmdState::Done; diff --git a/src/runtime/shell/subproc.rs b/src/runtime/shell/subproc.rs index 5c4ed46bf625..36f862175ef4 100644 --- a/src/runtime/shell/subproc.rs +++ b/src/runtime/shell/subproc.rs @@ -1653,6 +1653,7 @@ impl CapturedWriter { chunk.len(), parent.buffered_output.len() ); + let interp = parent.interp; // `dead == false` ⇒ writer.is_some() (set in PipeReader::create). let writer = self .writer @@ -1663,7 +1664,12 @@ impl CapturedWriter { // `io_writer::ChildPtr::subproc_capture` / `WriterTag::Subproc`. let child = io_writer::ChildPtr::subproc_capture(std::ptr::from_mut(self).cast::()); let y = writer.enqueue(child, None, chunk); - self.parent().run_yield(y); + // A dead writer hands the chunk's error completion back as `y` + // (`IOWriter::handle_dead_writer` / `on_sync_error`); running it lands + // in `on_iowriter_chunk` → `on_capture_failed`, which re-enters the Cmd + // and derefs the embedding `PipeReader` through `Readable::Pipe`, so + // no `&PipeReader` may be live here (see `run_yield_with`). + PipeReader::run_yield_with(interp, y); } pub(crate) fn on_iowriter_chunk(&mut self, amount: usize, err: Option) -> Yield { @@ -1693,12 +1699,17 @@ impl CapturedWriter { e.syscall ); self.err = Some(e); - } else if !all_written { + // SAFETY: `parent_mut` recovers the embedding `PipeReader`; raw-ptr + // form per `on_capture_failed` / `try_signal_done_to_cmd` contract + // (no `&mut PipeReader` held across the Cmd re-entry). The call may + // free the PipeReader, and `self` with it, so nothing touches + // `self` after it. + return unsafe { PipeReader::on_capture_failed(self.parent_mut()) }; + } + if !all_written { return Yield::Suspended; } - // SAFETY: `parent_mut` recovers the embedding `PipeReader`; raw-ptr - // form per `try_signal_done_to_cmd` contract (no `&mut PipeReader` - // held across the Cmd re-entry). + // SAFETY: as above. unsafe { PipeReader::try_signal_done_to_cmd(self.parent_mut()) } } } @@ -1756,16 +1767,13 @@ impl PipeReader { } /// Drive a `Yield` from inside an async I/O callback. Mirrors - /// `IOWriter::run_yield` / `IOReader::run_yield`. `interp` is wired at - /// `create` time from the spawning `Cmd`; the null guard is a defensive - /// debug-assert for tests that construct a PipeReader without a Cmd. - fn run_yield(&self, y: Yield) { - Self::run_yield_with(self.interp, y); - } - - /// Free-function form of [`run_yield`] for callers that must not hold any - /// `&PipeReader` borrow across the interpreter trampoline (which can - /// re-derive `&PipeReader` via the `Readable::Pipe` `Arc`). + /// `IOWriter::run_yield` / `IOReader::run_yield`. `interp` is the reader's + /// own `interp` field, wired at `create` time from the spawning `Cmd`; the + /// null guard is a defensive debug-assert for tests that construct a + /// PipeReader without a Cmd. Takes the pointer rather than `&self` because + /// callers must not hold any `&PipeReader` borrow across the interpreter + /// trampoline, which can re-derive `&PipeReader` via the `Readable::Pipe` + /// `Arc`. fn run_yield_with(interp: *mut crate::shell::interpreter::Interpreter, y: Yield) { if interp.is_null() { debug_assert!( @@ -1898,30 +1906,42 @@ impl PipeReader { /// `BufferedReaderParent::on_read_chunk` adapter — invoked with the /// `PipeReader` registered via `reader.set_parent(self)`. - pub(crate) fn on_read_chunk(&mut self, chunk: &[u8], has_more: ReadState) -> bool { - self.buffered_output.append(chunk); - log!( - "PipeReader(0x{:x}, {}) onReadChunk(chunk_len={}, has_more={})", - std::ptr::from_mut(self) as usize, - out_kind_str(self.out_type), - chunk.len(), - read_state_str(has_more) - ); + /// + /// Takes `*mut Self` like [`Self::on_reader_done`]: `do_write` can reject + /// the chunk synchronously (the shell's writer is already dead) and then + /// finish this pipe's capture (`on_capture_failed`), which re-enters the + /// `Cmd` and reaches this same allocation again through the + /// `Readable::Pipe` `Arc`. Only the `captured_writer` field is borrowed + /// across that call. + /// + /// # Safety + /// `this` must point into the live `Arc` allocation; the + /// read loop that dispatches this holds the `ref_`/`deref` keepalive, so + /// the allocation outlives the call even if the `Cmd` drops its `Arc`. + pub(crate) unsafe fn on_read_chunk(this: *mut Self, chunk: &[u8], has_more: ReadState) -> bool { + // SAFETY: caller contract; each borrow is scoped to its own statement. + unsafe { + (*this).buffered_output.append(chunk); + log!( + "PipeReader(0x{:x}, {}) onReadChunk(chunk_len={}, has_more={})", + this as usize, + out_kind_str((*this).out_type), + chunk.len(), + read_state_str(has_more) + ); - self.captured_writer.do_write(chunk); + (*this).captured_writer.do_write(chunk); + } // No explicit re-arm here (`register_poll()` on POSIX / // `start_with_current_pipe()` on Windows). This callback runs from - // inside the bun_io read loop, which still holds `&mut self.reader` - // on its stack and re-registers the poll itself based on the bool we - // return (`IOReader::on_read_chunk_cb` and - // `WindowsBufferedReader::on_read` document the same contract). - // - // Re-arming from here also violates `BufferedReaderParent`'s - // requirement that `on_read_chunk` never frees the reader: - // `register_poll()`'s failure path dispatches `on_reader_error`, - // which drops the last `Arc` and frees the - // `PosixBufferedReader` the loop is still reading through. + // inside the bun_io read loop, which first checks whether the + // callback closed the reader (`on_capture_failed` does) and otherwise + // re-registers the poll itself based on the bool we return + // (`IOReader::on_read_chunk_cb` and `WindowsBufferedReader::on_read` + // document the same contract). A re-arm from here would also run + // `register_poll()`'s failure path, `on_reader_error`, a terminal + // callback, from inside a non-terminal one. has_more != ReadState::Eof } @@ -1929,7 +1949,7 @@ impl PipeReader { /// `BufferedReader` stored at `set_parent` time. Keepalive for /// `on_reader_done` / `on_reader_error`: /// the returned guard keeps the allocation alive across - /// `run_yield` (which may free the owning `Cmd`) and `on_close_io` (which + /// `run_yield_with` (which may free the owning `Cmd`) and `on_close_io` (which /// drops the `Readable::Pipe` strong ref). Dropping the guard is the /// matching deref and may free `self`. /// @@ -2053,6 +2073,43 @@ impl PipeReader { Yield::Suspended } + /// The `CapturedWriter` failed to relay this pipe's output to the shell's + /// stdout/stderr and has recorded the error (typically EPIPE: nothing + /// reads the process's stdout any more). If the child still has its end + /// of the pipe open (`Pending`), stop reading and close ours, so that its + /// next write fails (EPIPE/SIGPIPE, as under a real shell) instead of + /// being appended to `buffered_output` for as long as it keeps writing + /// while the command never finishes. The state transition is the one + /// `on_reader_done` makes at EOF (`deinit` has just emptied the reader's + /// buffer, hence the empty `Done`); `try_signal_done_to_cmd` then hands + /// the relay error to the `Cmd` via `take_captured_error`, exactly as it + /// does when the relay fails after EOF. + /// + /// # Safety + /// Same contract as [`Self::try_signal_done_to_cmd`]. This is also + /// reached from inside this reader's own `on_read_chunk` (a chunk enqueued + /// on an already dead writer is rejected synchronously): `deinit` only + /// closes the handle, which the read loop checks for after the callback, + /// and the loop's `ref_`/`deref` keepalive (see the + /// `impl_buffered_reader_parent!` block below) keeps the allocation alive + /// until it returns even though the `Cmd` drops its `Arc` in here. + pub(crate) unsafe fn on_capture_failed(this: *mut Self) -> Yield { + // SAFETY: caller contract; the borrow ends before the Cmd re-entry in + // `try_signal_done_to_cmd`. `reader.deinit()` fires no callback. + unsafe { + if matches!((*this).state, PipeReaderState::Pending) { + log!( + "PipeReader(0x{:x}, {}) capture failed while pending, closing", + this as usize, + out_kind_str((*this).out_type) + ); + (*this).reader.deinit(); + (*this).state = PipeReaderState::Done(Box::default()); + } + Self::try_signal_done_to_cmd(this) + } + } + fn take_captured_error(&mut self) -> Option { if let Some(e) = self.captured_writer.err.take() { match core::mem::replace(&mut self.state, PipeReaderState::Pending) { @@ -2224,14 +2281,23 @@ impl Drop for PipeReader { // `on_reader_done`/`on_reader_error` forward the raw `*mut Self` (NOT // autoref) — see their doc-comments: the body builds an `Arc` keepalive that // may free `this` on drop, so a `&mut self` protector would be UB. +// +// `ref_`/`deref` bracket every read-loop entry (`bun_io`'s `ref_parent`) with +// a strong count on the `Arc` this reader lives in, the same way `IOReader` +// guards itself. `on_read_chunk` can finish the pipe's capture from inside the +// loop (`CapturedWriter::do_write` → `on_capture_failed`), which makes the Cmd +// drop the `Readable::Pipe` `Arc`; the loop still reads the reader's flags +// after the callback returns, so the last ref must not go away before then. bun_io::impl_buffered_reader_parent! { ShellPipeReader for PipeReader; has_on_read_chunk = true; - on_read_chunk = |this, chunk, has_more| (*this).on_read_chunk(&chunk, has_more); + on_read_chunk = |this, chunk, has_more| PipeReader::on_read_chunk(this, &chunk, has_more); on_reader_done = |this| PipeReader::on_reader_done(this); on_reader_error = |this, err| PipeReader::on_reader_error(this, &err); loop_ = |this| (*this).r#loop(); event_loop = |this| (*this).event_loop.as_event_loop_ctx(); + ref_ = |this| Arc::increment_strong_count(this.cast_const()); + deref = |this| Arc::decrement_strong_count(this.cast_const()); } // ─────────────────────────────────────────────────────────────────────────── diff --git a/test/js/bun/shell/epipe-relay-fixture.ts b/test/js/bun/shell/epipe-relay-fixture.ts new file mode 100644 index 000000000000..022f9c5abd43 --- /dev/null +++ b/test/js/bun/shell/epipe-relay-fixture.ts @@ -0,0 +1,74 @@ +// Spawned by epipe.test.ts. The test closes the read end of this process's +// stdout, then this process runs one shell command whose output the shell has +// to relay to that stdout. The relay fails (EPIPE), and the command has to +// finish anyway. The command's exit code is reported as the last stderr line. +// +// The command runs this same file in "produce" mode as an external command: a +// child that writes to its stdout for as long as that works and exits 0 once a +// write fails, which is what a real shell's SIGPIPE does to such a child. +import { $ } from "bun"; +import { writeSync } from "node:fs"; + +const mode = process.argv[2]; + +if (mode === "produce") { + const chunk = Buffer.alloc(64, "y\n"); + while (true) { + try { + writeSync(1, chunk); + } catch (e: any) { + // EPIPE, or ECONNRESET when the shell closed its end of the socket pair + // while a chunk it had not read yet was still queued in it. + console.error(`producer: stdout write failed: ${e.code}`); + process.exit(0); + } + Bun.sleepSync(1); + } +} + +if (mode === "produce-once") { + writeSync(1, "y\n"); + process.exit(0); +} + +// Wait until nothing reads our stdout anymore, so the relay really fails. +while (true) { + try { + writeSync(1, "still has a reader\n"); + } catch (e: any) { + if (e.code === "EPIPE") break; + if (e.code !== "EAGAIN") throw e; + } + await Bun.sleep(1); +} + +const producer = [process.execPath, import.meta.path, "produce"]; +let result; +switch (mode) { + // The relay fails asynchronously (the shell's writer learns about EPIPE from + // its poll) while the child keeps the pipe open. stderr is still relayed, so + // the command finishes once the child notices its stdout is gone and exits. + case "relay": + result = await $`${producer}`.nothrow(); + break; + // Same failure, but stdout is the only piped stream, so the failure finishes + // the command on the spot and the shell kills the child. + case "relay-only-stdout": + result = await $`${producer} 2> /dev/null`.nothrow(); + break; + // `echo` already broke the shell's writer, so the child's first chunk is + // rejected synchronously, from inside the read of that chunk. + case "dead-writer": + result = await $`echo hello; ${producer}`.nothrow(); + break; + case "dead-writer-only-stdout": + result = await $`echo hello; ${producer} 2> /dev/null`.nothrow(); + break; + // A child that exits 0 by itself: the failed relay still fails the command. + case "exits-by-itself": + result = await $`${[process.execPath, import.meta.path, "produce-once"]}`.nothrow(); + break; + default: + throw new Error(`unknown mode ${mode}`); +} +console.error(JSON.stringify({ settled: true, exitCode: result.exitCode })); diff --git a/test/js/bun/shell/epipe.test.ts b/test/js/bun/shell/epipe.test.ts index 27f6eabe8302..8e09265101dd 100644 --- a/test/js/bun/shell/epipe.test.ts +++ b/test/js/bun/shell/epipe.test.ts @@ -1,5 +1,6 @@ import { describe, expect, test } from "bun:test"; -import { isPosix } from "harness"; +import { bunEnv, bunExe, isPosix } from "harness"; +import { join } from "node:path"; import { createTestBuilder } from "./test_builder"; const TestBuilder = createTestBuilder(import.meta.path); @@ -20,3 +21,58 @@ describe.if(isPosix)("IOWriter epipe", () => { } }); }); + +// The shell relays an external command's output to the process's stdout while +// it captures it. Once nothing reads that stdout anymore, the relay fails with +// EPIPE. The command must still finish: the shell closes its end of the +// command's stdout, so the command's next write fails (the fixture's producer +// reports that and exits), instead of the shell reading the output into its +// capture buffer for as long as the command keeps writing. `$` used to never +// settle for such a command (the modes are described in the fixture). +describe.if(isPosix)("relayed output after the stdout reader went away", () => { + const fixture = join(import.meta.dir, "epipe-relay-fixture.ts"); + + async function runFixture(mode: string) { + await using proc = Bun.spawn({ + cmd: [bunExe(), fixture, mode], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + // Drop the read end of the fixture's stdout. The fixture waits until its + // own writes fail with EPIPE before it runs the command. + await proc.stdout.cancel(); + const [stderr, exitCode] = await Promise.all([proc.stderr.text(), proc.exited]); + const lastLine = stderr.trim().split("\n").at(-1)!; + // On a crash this shows the whole stderr instead of a JSON parse error. + const report = lastLine.startsWith("{") ? JSON.parse(lastLine) : stderr; + expect(report).toEqual({ settled: true, exitCode: expect.any(Number) }); + // The producer itself exits 0: a failed relay fails the command. The exact + // status (the errno today) is not pinned here. + expect(report.exitCode).not.toBe(0); + expect(exitCode).toBe(0); + return stderr; + } + + test.concurrent("the child keeps writing, its stderr is still relayed", async () => { + const stderr = await runFixture("relay"); + expect(stderr).toContain("producer: stdout write failed"); + }); + + test.concurrent("the child keeps writing, stdout is the only relayed stream", async () => { + await runFixture("relay-only-stdout"); + }); + + test.concurrent("the shell's writer is already dead when the child writes", async () => { + const stderr = await runFixture("dead-writer"); + expect(stderr).toContain("producer: stdout write failed"); + }); + + test.concurrent("the shell's writer is already dead, stdout is the only relayed stream", async () => { + await runFixture("dead-writer-only-stdout"); + }); + + test.concurrent("the child exits by itself", async () => { + await runFixture("exits-by-itself"); + }); +}); From e2aec7b3d598528a21837f86a61db9fde173c073 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 22 Aug 2026 03:30:05 +0000 Subject: [PATCH 2/7] test(shell): accept ECONNRESET in the epipe fixture's wait loop The fixture's stdout is a socket pair. If the test closes its end while a line is still unread, the next write fails with ECONNRESET, not EPIPE. The producer already treats both as the reader being gone. --- test/js/bun/shell/epipe-relay-fixture.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/test/js/bun/shell/epipe-relay-fixture.ts b/test/js/bun/shell/epipe-relay-fixture.ts index 022f9c5abd43..38f245c435af 100644 --- a/test/js/bun/shell/epipe-relay-fixture.ts +++ b/test/js/bun/shell/epipe-relay-fixture.ts @@ -32,11 +32,13 @@ if (mode === "produce-once") { } // Wait until nothing reads our stdout anymore, so the relay really fails. +// Our stdout is a socket pair too (Bun.spawn), so the failure is EPIPE, or +// ECONNRESET when the test closed its end while a line was still unread. while (true) { try { writeSync(1, "still has a reader\n"); } catch (e: any) { - if (e.code === "EPIPE") break; + if (e.code === "EPIPE" || e.code === "ECONNRESET") break; if (e.code !== "EAGAIN") throw e; } await Bun.sleep(1); From 5fa8142201fbdffb834fde3f938b429ffab1d8cc Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 22 Aug 2026 03:34:35 +0000 Subject: [PATCH 3/7] shell: shorten the comments around the capture-failure path --- src/runtime/shell/subproc.rs | 95 ++++++++++++------------------------ 1 file changed, 31 insertions(+), 64 deletions(-) diff --git a/src/runtime/shell/subproc.rs b/src/runtime/shell/subproc.rs index 36f862175ef4..837e47032696 100644 --- a/src/runtime/shell/subproc.rs +++ b/src/runtime/shell/subproc.rs @@ -1664,11 +1664,8 @@ impl CapturedWriter { // `io_writer::ChildPtr::subproc_capture` / `WriterTag::Subproc`. let child = io_writer::ChildPtr::subproc_capture(std::ptr::from_mut(self).cast::()); let y = writer.enqueue(child, None, chunk); - // A dead writer hands the chunk's error completion back as `y` - // (`IOWriter::handle_dead_writer` / `on_sync_error`); running it lands - // in `on_iowriter_chunk` → `on_capture_failed`, which re-enters the Cmd - // and derefs the embedding `PipeReader` through `Readable::Pipe`, so - // no `&PipeReader` may be live here (see `run_yield_with`). + // On a dead writer `y` is the chunk's error completion, which can free + // the embedding `PipeReader` (`on_capture_failed`). PipeReader::run_yield_with(interp, y); } @@ -1700,16 +1697,14 @@ impl CapturedWriter { ); self.err = Some(e); // SAFETY: `parent_mut` recovers the embedding `PipeReader`; raw-ptr - // form per `on_capture_failed` / `try_signal_done_to_cmd` contract - // (no `&mut PipeReader` held across the Cmd re-entry). The call may - // free the PipeReader, and `self` with it, so nothing touches - // `self` after it. + // form per `on_capture_failed` contract (no `&mut PipeReader` held + // across the Cmd re-entry). May free `self`; last use of it. return unsafe { PipeReader::on_capture_failed(self.parent_mut()) }; } if !all_written { return Yield::Suspended; } - // SAFETY: as above. + // SAFETY: as above, per `try_signal_done_to_cmd` contract. unsafe { PipeReader::try_signal_done_to_cmd(self.parent_mut()) } } } @@ -1766,14 +1761,9 @@ impl PipeReader { self.captured_writer_done(0) } - /// Drive a `Yield` from inside an async I/O callback. Mirrors - /// `IOWriter::run_yield` / `IOReader::run_yield`. `interp` is the reader's - /// own `interp` field, wired at `create` time from the spawning `Cmd`; the - /// null guard is a defensive debug-assert for tests that construct a - /// PipeReader without a Cmd. Takes the pointer rather than `&self` because - /// callers must not hold any `&PipeReader` borrow across the interpreter - /// trampoline, which can re-derive `&PipeReader` via the `Readable::Pipe` - /// `Arc`. + /// Drive a `Yield` from inside an async I/O callback with the reader's + /// `interp` (wired at `create`). Not `&self`: the trampoline can re-derive + /// `&PipeReader` via the `Readable::Pipe` `Arc`, or free it. fn run_yield_with(interp: *mut crate::shell::interpreter::Interpreter, y: Yield) { if interp.is_null() { debug_assert!( @@ -1905,19 +1895,14 @@ impl PipeReader { } /// `BufferedReaderParent::on_read_chunk` adapter — invoked with the - /// `PipeReader` registered via `reader.set_parent(self)`. - /// - /// Takes `*mut Self` like [`Self::on_reader_done`]: `do_write` can reject - /// the chunk synchronously (the shell's writer is already dead) and then - /// finish this pipe's capture (`on_capture_failed`), which re-enters the - /// `Cmd` and reaches this same allocation again through the - /// `Readable::Pipe` `Arc`. Only the `captured_writer` field is borrowed - /// across that call. + /// `PipeReader` registered via `reader.set_parent(self)`. Raw `*mut Self` + /// like [`Self::on_reader_done`]: `do_write` can re-enter the `Cmd` + /// (`on_capture_failed`), which reaches this allocation through the + /// `Readable::Pipe` `Arc`. /// /// # Safety - /// `this` must point into the live `Arc` allocation; the - /// read loop that dispatches this holds the `ref_`/`deref` keepalive, so - /// the allocation outlives the call even if the `Cmd` drops its `Arc`. + /// `this` must point into the live `Arc` allocation. The read + /// loop's `ref_`/`deref` keepalive covers the `Arc` the `Cmd` may drop. pub(crate) unsafe fn on_read_chunk(this: *mut Self, chunk: &[u8], has_more: ReadState) -> bool { // SAFETY: caller contract; each borrow is scoped to its own statement. unsafe { @@ -1933,15 +1918,10 @@ impl PipeReader { (*this).captured_writer.do_write(chunk); } - // No explicit re-arm here (`register_poll()` on POSIX / - // `start_with_current_pipe()` on Windows). This callback runs from - // inside the bun_io read loop, which first checks whether the - // callback closed the reader (`on_capture_failed` does) and otherwise - // re-registers the poll itself based on the bool we return - // (`IOReader::on_read_chunk_cb` and `WindowsBufferedReader::on_read` - // document the same contract). A re-arm from here would also run - // `register_poll()`'s failure path, `on_reader_error`, a terminal - // callback, from inside a non-terminal one. + // No explicit re-arm here: the bun_io read loop checks whether the + // callback closed the reader, then re-registers the poll itself from + // the bool we return (`IOReader::on_read_chunk_cb` documents the same + // contract). has_more != ReadState::Eof } @@ -2073,29 +2053,19 @@ impl PipeReader { Yield::Suspended } - /// The `CapturedWriter` failed to relay this pipe's output to the shell's - /// stdout/stderr and has recorded the error (typically EPIPE: nothing - /// reads the process's stdout any more). If the child still has its end - /// of the pipe open (`Pending`), stop reading and close ours, so that its - /// next write fails (EPIPE/SIGPIPE, as under a real shell) instead of - /// being appended to `buffered_output` for as long as it keeps writing - /// while the command never finishes. The state transition is the one - /// `on_reader_done` makes at EOF (`deinit` has just emptied the reader's - /// buffer, hence the empty `Done`); `try_signal_done_to_cmd` then hands - /// the relay error to the `Cmd` via `take_captured_error`, exactly as it - /// does when the relay fails after EOF. + /// The `CapturedWriter` could not relay this pipe's output (EPIPE: nothing + /// reads the process's stdout any more) and has recorded the error. While + /// the child still holds its end (`Pending`), close ours so its next write + /// fails, as under a real shell, instead of reading it for as long as it + /// writes. The state becomes what EOF leaves, so `try_signal_done_to_cmd` + /// reports the relay error to the `Cmd` now. /// /// # Safety - /// Same contract as [`Self::try_signal_done_to_cmd`]. This is also - /// reached from inside this reader's own `on_read_chunk` (a chunk enqueued - /// on an already dead writer is rejected synchronously): `deinit` only - /// closes the handle, which the read loop checks for after the callback, - /// and the loop's `ref_`/`deref` keepalive (see the - /// `impl_buffered_reader_parent!` block below) keeps the allocation alive - /// until it returns even though the `Cmd` drops its `Arc` in here. + /// Same contract as [`Self::try_signal_done_to_cmd`]. May run inside this + /// reader's own `on_read_chunk`: `deinit` fires no callback, and the read + /// loop's `ref_`/`deref` keepalive covers the `Arc` the `Cmd` drops here. pub(crate) unsafe fn on_capture_failed(this: *mut Self) -> Yield { - // SAFETY: caller contract; the borrow ends before the Cmd re-entry in - // `try_signal_done_to_cmd`. `reader.deinit()` fires no callback. + // SAFETY: caller contract; the borrow ends before the Cmd re-entry. unsafe { if matches!((*this).state, PipeReaderState::Pending) { log!( @@ -2282,12 +2252,9 @@ impl Drop for PipeReader { // autoref) — see their doc-comments: the body builds an `Arc` keepalive that // may free `this` on drop, so a `&mut self` protector would be UB. // -// `ref_`/`deref` bracket every read-loop entry (`bun_io`'s `ref_parent`) with -// a strong count on the `Arc` this reader lives in, the same way `IOReader` -// guards itself. `on_read_chunk` can finish the pipe's capture from inside the -// loop (`CapturedWriter::do_write` → `on_capture_failed`), which makes the Cmd -// drop the `Readable::Pipe` `Arc`; the loop still reads the reader's flags -// after the callback returns, so the last ref must not go away before then. +// `ref_`/`deref`: the read loop's keepalive on our `Arc` (as in `IOReader`). +// `on_read_chunk` can make the `Cmd` drop the `Readable::Pipe` `Arc` while the +// loop still has to read the reader's flags. bun_io::impl_buffered_reader_parent! { ShellPipeReader for PipeReader; has_on_read_chunk = true; From 669e7ed61127f1f97f57f558340576340db44f88 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 22 Aug 2026 03:37:35 +0000 Subject: [PATCH 4/7] shell: drop a redundant comment in CapturedWriter::do_write --- src/runtime/shell/subproc.rs | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/src/runtime/shell/subproc.rs b/src/runtime/shell/subproc.rs index 837e47032696..7352d47762bd 100644 --- a/src/runtime/shell/subproc.rs +++ b/src/runtime/shell/subproc.rs @@ -1664,8 +1664,6 @@ impl CapturedWriter { // `io_writer::ChildPtr::subproc_capture` / `WriterTag::Subproc`. let child = io_writer::ChildPtr::subproc_capture(std::ptr::from_mut(self).cast::()); let y = writer.enqueue(child, None, chunk); - // On a dead writer `y` is the chunk's error completion, which can free - // the embedding `PipeReader` (`on_capture_failed`). PipeReader::run_yield_with(interp, y); } @@ -2053,12 +2051,10 @@ impl PipeReader { Yield::Suspended } - /// The `CapturedWriter` could not relay this pipe's output (EPIPE: nothing - /// reads the process's stdout any more) and has recorded the error. While - /// the child still holds its end (`Pending`), close ours so its next write - /// fails, as under a real shell, instead of reading it for as long as it - /// writes. The state becomes what EOF leaves, so `try_signal_done_to_cmd` - /// reports the relay error to the `Cmd` now. + /// Relaying this pipe's output failed (the `CapturedWriter` holds the + /// error). While the child still holds its end (`Pending`), close ours so + /// its next write fails, as under a real shell, and settle the state as + /// EOF would so `try_signal_done_to_cmd` reports the error to the `Cmd`. /// /// # Safety /// Same contract as [`Self::try_signal_done_to_cmd`]. May run inside this From 843136145ffb8e0fe7711d3fb7eb3150a0c22671 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 22 Aug 2026 04:42:57 +0000 Subject: [PATCH 5/7] test(shell): run the relay-failure cases on Windows too The fixture and all five of its modes pass on a Windows debug build, and the unfixed canary hangs there the same way (the relay error maps to 4047). --- test/js/bun/shell/epipe.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/js/bun/shell/epipe.test.ts b/test/js/bun/shell/epipe.test.ts index 8e09265101dd..bdae673302ab 100644 --- a/test/js/bun/shell/epipe.test.ts +++ b/test/js/bun/shell/epipe.test.ts @@ -29,7 +29,7 @@ describe.if(isPosix)("IOWriter epipe", () => { // reports that and exits), instead of the shell reading the output into its // capture buffer for as long as the command keeps writing. `$` used to never // settle for such a command (the modes are described in the fixture). -describe.if(isPosix)("relayed output after the stdout reader went away", () => { +describe("relayed output after the stdout reader went away", () => { const fixture = join(import.meta.dir, "epipe-relay-fixture.ts"); async function runFixture(mode: string) { From 95c7d5ee60f8f3dfb565dc8b21cca8eaa888a2fa Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 22 Aug 2026 08:51:41 +0000 Subject: [PATCH 6/7] shell: let a child outlive a failed relay, and finish the raw-pointer dispatch A relay failure used to set the command's exit status at once. With stdout as the only piped stream (`2>&1`, `2>/dev/null`) that finished the command while the child was still running, and Cmd::deinit killed it with SIGKILL, while the default shape let the same child see EPIPE and exit by itself. The relay error is now a CaptureError::Relay: the stream is closed, the status is kept on SubprocExec::relay_status, and Cmd::on_exit applies it once the child has exited, so the command always waits for the child like a real shell does. Read errors (CaptureError::Read) keep finishing the command at once. CapturedWriter::on_iowriter_chunk takes *mut Self and the IOWriter dispatch forms no &mut: the call can drop the last Arc, which frees the CapturedWriter, the same way on_read_chunk already documents. WindowsBufferedReader::deinit released `_buffer` before close_impl could move it to a File with a read in flight (the `2>&1` dup shape reads through a uv_fs_read). The buffer is now released after close_impl. The fixture's producer records its failed write in a file, so every mode can check that the child saw the closed pipe and exited on its own, and `2>&1` is covered. --- src/io/PipeReader.rs | 35 +++---- src/runtime/shell/IOWriter.rs | 20 ++-- src/runtime/shell/states/Cmd.rs | 52 ++++++++--- src/runtime/shell/subproc.rs | 114 +++++++++++++++-------- test/js/bun/shell/epipe-relay-fixture.ts | 24 +++-- test/js/bun/shell/epipe.test.ts | 28 ++++-- 6 files changed, 182 insertions(+), 91 deletions(-) diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index a425cd64358f..8ae832e6e602 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -1783,24 +1783,25 @@ impl WindowsBufferedReader { /// before Drop; both paths are idempotent over an already-taken source. pub fn deinit(&mut self) { MaxBuf::remove_from_pipereader(&mut self.maxbuf); - self._buffer = Vec::new(); - let Some(source) = self.source.take() else { - return; - }; - if !source.is_closed() { - // closeImpl will take care of freeing the source. - // Dropping the `Box` here would free a uv_pipe_t still - // linked into the loop's handle queue → UAF. Restore the source so - // close_impl can do the proper take + hand-off to libuv - // (into_raw + uv_close). - self.source = Some(source); - self.close_impl::(); - } else { - // Already closing/closed: a uv close callback may still be pending - // on this allocation; dropping the Box would free memory libuv - // still owns, so leak it instead. - core::mem::forget(source); + if let Some(source) = self.source.take() { + if !source.is_closed() { + // closeImpl will take care of freeing the source. + // Dropping the `Box` here would free a uv_pipe_t still + // linked into the loop's handle queue → UAF. Restore the source so + // close_impl can do the proper take + hand-off to libuv + // (into_raw + uv_close). It also moves `_buffer` to the File + // when a read is in flight (libuv still writes into it), so the + // buffer is released only after it. + self.source = Some(source); + self.close_impl::(); + } else { + // Already closing/closed: a uv close callback may still be pending + // on this allocation; dropping the Box would free memory libuv + // still owns, so leak it instead. + core::mem::forget(source); + } } + self._buffer = Vec::new(); } #[cfg(windows)] diff --git a/src/runtime/shell/IOWriter.rs b/src/runtime/shell/IOWriter.rs index 4f8d012cab07..8665a2ae1e68 100644 --- a/src/runtime/shell/IOWriter.rs +++ b/src/runtime/shell/IOWriter.rs @@ -1238,13 +1238,19 @@ pub(crate) fn on_io_writer_chunk( WriterTag::Subproc => { let _ = interp; debug_assert!(!child.raw.is_null()); - // SAFETY: `raw` was set from `&mut CapturedWriter` in - // `CapturedWriter::do_write`; the PipeReader (and the embedded - // CapturedWriter) is kept alive by the `Readable::Pipe` Arc on - // the owning ShellSubprocess until `on_close_io` runs, which only - // happens after the writer has finished draining. Single-threaded. - let cw = unsafe { &mut *child.raw.cast::() }; - cw.on_iowriter_chunk(written, err) + // SAFETY: `raw` is the `CapturedWriter` inside the `Arc` + // that enqueued this chunk in `do_write`; a PipeReader cancels its + // queued chunks before it can be freed, so it is live here. The + // callee may drop that last `Arc` (`Cmd::buffered_output_close` → + // `close_io` → `detach`), so it is dispatched by raw pointer with + // no `&mut` formed here. Single-threaded. + unsafe { + crate::shell::subproc::CapturedWriter::on_iowriter_chunk( + child.raw.cast::(), + written, + err, + ) + } } } } diff --git a/src/runtime/shell/states/Cmd.rs b/src/runtime/shell/states/Cmd.rs index 334f6da5f0ec..c7981e915d54 100644 --- a/src/runtime/shell/states/Cmd.rs +++ b/src/runtime/shell/states/Cmd.rs @@ -8,7 +8,7 @@ use crate::shell::ast; use crate::shell::builtin::{Builtin, Kind as BuiltinKind}; use crate::shell::interpreter::{CowFd, Interpreter, Node, NodeId, ShellExecEnv, StateKind, log}; use crate::shell::io::{IO, OutKind as IoOutKind}; -use crate::shell::shell_body::subproc::{Readable, ShellSubprocess, StdioKind}; +use crate::shell::shell_body::subproc::{CaptureError, Readable, ShellSubprocess, StdioKind}; use crate::shell::states::assigns::{AssignCtx, Assigns}; use crate::shell::states::base::Base; use crate::shell::states::expansion::Expansion; @@ -68,6 +68,12 @@ impl Cmd { pub struct SubprocExec { pub(crate) child: *mut ShellSubprocess, pub(crate) buffered_closed: BufferedIoClosed, + /// Status the command gets once the child exits, set when relaying its + /// output to the shell's stdout/stderr failed while it was still running. + /// The shell closed its end of that pipe; the child goes on until its next + /// write fails or it finishes, like under a real shell, so the command + /// waits for its exit instead of killing it. + pub(crate) relay_status: Option, /// NodeId-arena backrefs so the legacy `&mut self` subprocess callbacks /// (`buffered_output_close` / `on_exit`) can hand a [`Yield`] back to the /// trampoline. The `Cmd` lives inside `interp.nodes`, so we stash the @@ -564,6 +570,7 @@ impl Cmd { interp.as_cmd_mut(this).exec = Exec::Subproc(Box::new(SubprocExec { child: core::ptr::null_mut(), buffered_closed, + relay_status: None, interp: core::ptr::null_mut(), this_id: this, })); @@ -952,11 +959,12 @@ impl Cmd { pub(crate) fn buffered_output_close( &mut self, kind: OutKind, - err: Option, + err: Option, ) -> Yield { + self.record_capture_error(err); match kind { - OutKind::Stdout => self.buffered_output_close_stdout(err), - OutKind::Stderr => self.buffered_output_close_stderr(err), + OutKind::Stdout => self.buffered_output_close_stdout(), + OutKind::Stderr => self.buffered_output_close_stderr(), } if self.has_finished() { // Set `state = Done` and hand the Yield back to the caller @@ -981,12 +989,29 @@ impl Cmd { Yield::suspended() } - fn buffered_output_close_stdout(&mut self, err: Option) { + /// A read error ends the command now, whether or not the child has exited + /// (`deinit` then kills it). A relay error ends it with that status once + /// the child has exited: the shell closed its end of the pipe, the child + /// runs on until its next write fails or it finishes. + fn record_capture_error(&mut self, err: Option) { debug_assert!(matches!(self.exec, Exec::Subproc(_))); - log!("cmd close buffered stdout"); - if let Some(e) = err { - self.exit_code = Some(e.errno.unsigned_abs() as ExitCode); + let Some(err) = err else { return }; + let status = err.errno().unsigned_abs() as ExitCode; + let Exec::Subproc(sub) = &mut self.exec else { + return; + }; + // SAFETY: `child` is the live subprocess owned by this Cmd. + let exited = unsafe { (*sub.child).has_exited() }; + match err { + CaptureError::Read(_) => self.exit_code = Some(status), + CaptureError::Relay(_) if exited => self.exit_code = Some(status), + CaptureError::Relay(_) => sub.relay_status = Some(status), } + } + + fn buffered_output_close_stdout(&mut self) { + debug_assert!(matches!(self.exec, Exec::Subproc(_))); + log!("cmd close buffered stdout"); let redirect = self.ast_node().redirect; let Exec::Subproc(sub) = &mut self.exec else { return; @@ -1017,12 +1042,9 @@ impl Cmd { child.close_io(StdioKind::Stdout); } - fn buffered_output_close_stderr(&mut self, err: Option) { + fn buffered_output_close_stderr(&mut self) { debug_assert!(matches!(self.exec, Exec::Subproc(_))); log!("cmd close buffered stderr"); - if let Some(e) = err { - self.exit_code = Some(e.errno.unsigned_abs() as ExitCode); - } let redirect = self.ast_node().redirect; let Exec::Subproc(sub) = &mut self.exec else { return; @@ -1053,7 +1075,11 @@ impl Cmd { /// Called by `ShellSubprocess::on_process_exit`. pub(crate) fn on_exit(&mut self, exit_code: ExitCode) { - self.exit_code = Some(exit_code); + let relay_status = match &mut self.exec { + Exec::Subproc(sub) => sub.relay_status.take(), + _ => None, + }; + self.exit_code = Some(relay_status.unwrap_or(exit_code)); let has_finished = self.has_finished(); log!("cmd exit code={} has_finished={}", exit_code, has_finished); if has_finished { diff --git a/src/runtime/shell/subproc.rs b/src/runtime/shell/subproc.rs index 7352d47762bd..82b98d9606db 100644 --- a/src/runtime/shell/subproc.rs +++ b/src/runtime/shell/subproc.rs @@ -1534,6 +1534,24 @@ pub enum PipeReaderState { Err(Option>), } +/// Why a `PipeReader` handed its `Cmd` an error. +pub enum CaptureError { + /// Reading the child's pipe failed. + Read(SystemError), + /// Relaying the output to the shell's stdout/stderr failed. The child may + /// still be running with the shell's end of the pipe closed + /// (`PipeReader::on_capture_failed`). + Relay(SystemError), +} + +impl CaptureError { + pub(crate) fn errno(&self) -> i32 { + match self { + CaptureError::Read(e) | CaptureError::Relay(e) => e.errno, + } + } +} + pub struct PipeReader { pub(crate) reader: IOReader, pub(crate) process: Option<*mut ShellSubprocess>, @@ -1636,7 +1654,7 @@ impl Default for CapturedWriter { } } -bun_core::impl_field_parent! { CapturedWriter => PipeReader.captured_writer; fn parent; fn mut parent_mut; } +bun_core::impl_field_parent! { CapturedWriter => PipeReader.captured_writer; fn parent; } impl CapturedWriter { pub(crate) fn do_write(&mut self, chunk: &[u8]) { @@ -1667,43 +1685,55 @@ impl CapturedWriter { PipeReader::run_yield_with(interp, y); } - pub(crate) fn on_iowriter_chunk(&mut self, amount: usize, err: Option) -> Yield { - let addr = std::ptr::from_mut(self) as usize; - let written = self.written + amount; - let parent = self.parent(); - log!( - "CapturedWriter({:x}, {}) onWrite({}, has_err={}) total_written={} total_to_write={}", - addr, - out_kind_str(parent.out_type), - amount, - err.is_some(), - written, - parent.buffered_output.len() - ); - let all_written = written >= parent.buffered_output.len() - && !matches!(parent.state, PipeReaderState::Pending); - self.written = written; - if let Some(e) = err { + /// The chunk completion from the shell's `IOWriter` (`WriterTag::Subproc`). + /// Raw `*mut Self` like the reader callbacks: the tail call can drop the + /// last `Arc` of the embedding `PipeReader` (`Cmd::buffered_output_close` + /// → `close_io` → `detach`), which frees `this`. + /// + /// # Safety + /// `this` must be the `CapturedWriter` of a live `PipeReader` (the pointer + /// `do_write` enqueued); no `&`/`&mut` to that `PipeReader` may be live. + pub(crate) unsafe fn on_iowriter_chunk( + this: *mut Self, + amount: usize, + err: Option, + ) -> Yield { + // SAFETY: caller contract; `this` is `PipeReader.captured_writer`, and + // every borrow below ends before the (maybe freeing) tail call. + unsafe { + let parent: *mut PipeReader = + bun_core::from_field_ptr!(PipeReader, captured_writer, this); + let written = (*this).written + amount; log!( - "CapturedWriter(0x{:x}, {}) onWrite errno={} errmsg={} errfd={:?} syscall={}", - addr, - out_kind_str(self.parent().out_type), - e.errno, - e.message, - e.fd, - e.syscall + "CapturedWriter({:x}, {}) onWrite({}, has_err={}) total_written={} total_to_write={}", + this as usize, + out_kind_str((*parent).out_type), + amount, + err.is_some(), + written, + (*parent).buffered_output.len() ); - self.err = Some(e); - // SAFETY: `parent_mut` recovers the embedding `PipeReader`; raw-ptr - // form per `on_capture_failed` contract (no `&mut PipeReader` held - // across the Cmd re-entry). May free `self`; last use of it. - return unsafe { PipeReader::on_capture_failed(self.parent_mut()) }; - } - if !all_written { - return Yield::Suspended; + let all_written = written >= (*parent).buffered_output.len() + && !matches!((*parent).state, PipeReaderState::Pending); + (*this).written = written; + if let Some(e) = err { + log!( + "CapturedWriter(0x{:x}, {}) onWrite errno={} errmsg={} errfd={:?} syscall={}", + this as usize, + out_kind_str((*parent).out_type), + e.errno, + e.message, + e.fd, + e.syscall + ); + (*this).err = Some(e); + return PipeReader::on_capture_failed(parent); + } + if !all_written { + return Yield::Suspended; + } + PipeReader::try_signal_done_to_cmd(parent) } - // SAFETY: as above, per `try_signal_done_to_cmd` contract. - unsafe { PipeReader::try_signal_done_to_cmd(self.parent_mut()) } } } @@ -2042,7 +2072,7 @@ impl PipeReader { let cmd = unsafe { (*proc).cmd_parent.cmd_mut() }; // SAFETY: caller contract — the `&mut` is scoped to this call and // ends *before* the `cmd` call below. - let e: Option = unsafe { (*this).take_captured_error() }; + let e: Option = unsafe { (*this).take_capture_error() }; // No `&`/`&mut PipeReader` is live here; `buffered_output_close` // is free to deref the sibling `Arc` in // `Readable::Pipe` for `pipe.slice()` / `close_io`. @@ -2076,18 +2106,22 @@ impl PipeReader { } } - fn take_captured_error(&mut self) -> Option { + /// The error this reader ends with, if any. A read error (`state`) wins + /// over a relay error (`captured_writer.err`); both leave the state `Err`. + fn take_capture_error(&mut self) -> Option { if let Some(e) = self.captured_writer.err.take() { match core::mem::replace(&mut self.state, PipeReaderState::Pending) { PipeReaderState::Done(buf) => { drop(buf); - self.state = PipeReaderState::Err(Some(Box::new(e))); + self.state = PipeReaderState::Err(None); + return Some(CaptureError::Relay(e)); } old @ PipeReaderState::Err(_) => { self.state = old; } PipeReaderState::Pending => { - self.state = PipeReaderState::Err(Some(Box::new(e))); + self.state = PipeReaderState::Err(None); + return Some(CaptureError::Relay(e)); } } } @@ -2095,7 +2129,7 @@ impl PipeReader { // Move it out (the only reader of // `state.Err` after this point is `Drop`, which tolerates `None`). if let PipeReaderState::Err(slot) = &mut self.state { - slot.take().map(|b| *b) + slot.take().map(|b| CaptureError::Read(*b)) } else { None } diff --git a/test/js/bun/shell/epipe-relay-fixture.ts b/test/js/bun/shell/epipe-relay-fixture.ts index 38f245c435af..1564063c72c9 100644 --- a/test/js/bun/shell/epipe-relay-fixture.ts +++ b/test/js/bun/shell/epipe-relay-fixture.ts @@ -5,9 +5,11 @@ // // The command runs this same file in "produce" mode as an external command: a // child that writes to its stdout for as long as that works and exits 0 once a -// write fails, which is what a real shell's SIGPIPE does to such a child. +// write fails, which is what a real shell's SIGPIPE does to such a child. The +// producer records the failed write in the file named by PRODUCER_REPORT (its +// stderr is not relayed in every mode); the report ends up in the JSON line. import { $ } from "bun"; -import { writeSync } from "node:fs"; +import { existsSync, readFileSync, writeFileSync, writeSync } from "node:fs"; const mode = process.argv[2]; @@ -19,7 +21,9 @@ if (mode === "produce") { } catch (e: any) { // EPIPE, or ECONNRESET when the shell closed its end of the socket pair // while a chunk it had not read yet was still queued in it. - console.error(`producer: stdout write failed: ${e.code}`); + const report = `stdout write failed: ${e.code}`; + console.error(`producer: ${report}`); + writeFileSync(process.env.PRODUCER_REPORT!, report); process.exit(0); } Bun.sleepSync(1); @@ -31,6 +35,9 @@ if (mode === "produce-once") { process.exit(0); } +const reportPath = process.argv[3]; +process.env.PRODUCER_REPORT = reportPath; + // Wait until nothing reads our stdout anymore, so the relay really fails. // Our stdout is a socket pair too (Bun.spawn), so the failure is EPIPE, or // ECONNRESET when the test closed its end while a line was still unread. @@ -53,11 +60,15 @@ switch (mode) { case "relay": result = await $`${producer}`.nothrow(); break; - // Same failure, but stdout is the only piped stream, so the failure finishes - // the command on the spot and the shell kills the child. + // Same failure, but stdout is the only piped stream. The command still waits + // for the child to notice and exit instead of killing it. case "relay-only-stdout": result = await $`${producer} 2> /dev/null`.nothrow(); break; + // The child's stderr shares the failed stdout pipe. + case "relay-stderr-to-stdout": + result = await $`${producer} 2>&1`.nothrow(); + break; // `echo` already broke the shell's writer, so the child's first chunk is // rejected synchronously, from inside the read of that chunk. case "dead-writer": @@ -73,4 +84,5 @@ switch (mode) { default: throw new Error(`unknown mode ${mode}`); } -console.error(JSON.stringify({ settled: true, exitCode: result.exitCode })); +const producerReport = existsSync(reportPath) ? readFileSync(reportPath, "utf8") : null; +console.error(JSON.stringify({ settled: true, exitCode: result.exitCode, producer: producerReport })); diff --git a/test/js/bun/shell/epipe.test.ts b/test/js/bun/shell/epipe.test.ts index bdae673302ab..3d8ae1b3e95a 100644 --- a/test/js/bun/shell/epipe.test.ts +++ b/test/js/bun/shell/epipe.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, isPosix } from "harness"; +import { bunEnv, bunExe, isPosix, tempDir } from "harness"; import { join } from "node:path"; import { createTestBuilder } from "./test_builder"; const TestBuilder = createTestBuilder(import.meta.path); @@ -24,17 +24,19 @@ describe.if(isPosix)("IOWriter epipe", () => { // The shell relays an external command's output to the process's stdout while // it captures it. Once nothing reads that stdout anymore, the relay fails with -// EPIPE. The command must still finish: the shell closes its end of the -// command's stdout, so the command's next write fails (the fixture's producer -// reports that and exits), instead of the shell reading the output into its -// capture buffer for as long as the command keeps writing. `$` used to never -// settle for such a command (the modes are described in the fixture). +// EPIPE. The command must still finish, the way it does under a real shell: +// the shell closes its end of the command's stdout, the command's next write +// fails (the fixture's producer reports that and exits 0), and the command +// finishes with the child's exit, not by the shell killing it. `$` used to +// never settle for such a command (the modes are described in the fixture). describe("relayed output after the stdout reader went away", () => { const fixture = join(import.meta.dir, "epipe-relay-fixture.ts"); async function runFixture(mode: string) { + using dir = tempDir("shell-epipe-relay", {}); + const reportPath = join(String(dir), "producer-report.txt"); await using proc = Bun.spawn({ - cmd: [bunExe(), fixture, mode], + cmd: [bunExe(), fixture, mode, reportPath], env: bunEnv, stdout: "pipe", stderr: "pipe", @@ -46,7 +48,13 @@ describe("relayed output after the stdout reader went away", () => { const lastLine = stderr.trim().split("\n").at(-1)!; // On a crash this shows the whole stderr instead of a JSON parse error. const report = lastLine.startsWith("{") ? JSON.parse(lastLine) : stderr; - expect(report).toEqual({ settled: true, exitCode: expect.any(Number) }); + expect(report).toEqual({ + settled: true, + exitCode: expect.any(Number), + // Written by the producer when its write failed (EPIPE, or ECONNRESET on + // Linux): the shell closed the pipe and let the child exit by itself. + producer: mode === "exits-by-itself" ? null : expect.stringContaining("stdout write failed: E"), + }); // The producer itself exits 0: a failed relay fails the command. The exact // status (the errno today) is not pinned here. expect(report.exitCode).not.toBe(0); @@ -63,6 +71,10 @@ describe("relayed output after the stdout reader went away", () => { await runFixture("relay-only-stdout"); }); + test.concurrent("the child keeps writing, its stderr shares the stdout pipe", async () => { + await runFixture("relay-stderr-to-stdout"); + }); + test.concurrent("the shell's writer is already dead when the child writes", async () => { const stderr = await runFixture("dead-writer"); expect(stderr).toContain("producer: stdout write failed"); From 1a640be9f3f7b90c8447c3195f166a225d536093 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 22 Aug 2026 08:55:45 +0000 Subject: [PATCH 7/7] shell: shorten the comments on the relay-status fields --- src/io/PipeReader.rs | 5 ++--- src/runtime/shell/states/Cmd.rs | 14 +++++--------- src/runtime/shell/subproc.rs | 3 +-- 3 files changed, 8 insertions(+), 14 deletions(-) diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index 8ae832e6e602..4c2dd85e0e09 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -1789,9 +1789,8 @@ impl WindowsBufferedReader { // Dropping the `Box` here would free a uv_pipe_t still // linked into the loop's handle queue → UAF. Restore the source so // close_impl can do the proper take + hand-off to libuv - // (into_raw + uv_close). It also moves `_buffer` to the File - // when a read is in flight (libuv still writes into it), so the - // buffer is released only after it. + // (into_raw + uv_close). It also takes `_buffer` for a read + // still in flight, so it runs before the buffer is released. self.source = Some(source); self.close_impl::(); } else { diff --git a/src/runtime/shell/states/Cmd.rs b/src/runtime/shell/states/Cmd.rs index c7981e915d54..6fe337597398 100644 --- a/src/runtime/shell/states/Cmd.rs +++ b/src/runtime/shell/states/Cmd.rs @@ -68,11 +68,9 @@ impl Cmd { pub struct SubprocExec { pub(crate) child: *mut ShellSubprocess, pub(crate) buffered_closed: BufferedIoClosed, - /// Status the command gets once the child exits, set when relaying its - /// output to the shell's stdout/stderr failed while it was still running. - /// The shell closed its end of that pipe; the child goes on until its next - /// write fails or it finishes, like under a real shell, so the command - /// waits for its exit instead of killing it. + /// Status a failed relay imposes once the child exits (`on_exit`). The + /// shell only closed its end of the pipe, so the child runs on until its + /// next write fails or it finishes, and the command waits for it. pub(crate) relay_status: Option, /// NodeId-arena backrefs so the legacy `&mut self` subprocess callbacks /// (`buffered_output_close` / `on_exit`) can hand a [`Yield`] back to the @@ -989,10 +987,8 @@ impl Cmd { Yield::suspended() } - /// A read error ends the command now, whether or not the child has exited - /// (`deinit` then kills it). A relay error ends it with that status once - /// the child has exited: the shell closed its end of the pipe, the child - /// runs on until its next write fails or it finishes. + /// A read error ends the command now (`deinit` kills a child still + /// running). A relay error becomes the status once the child has exited. fn record_capture_error(&mut self, err: Option) { debug_assert!(matches!(self.exec, Exec::Subproc(_))); let Some(err) = err else { return }; diff --git a/src/runtime/shell/subproc.rs b/src/runtime/shell/subproc.rs index 82b98d9606db..94ab19300b1a 100644 --- a/src/runtime/shell/subproc.rs +++ b/src/runtime/shell/subproc.rs @@ -1687,8 +1687,7 @@ impl CapturedWriter { /// The chunk completion from the shell's `IOWriter` (`WriterTag::Subproc`). /// Raw `*mut Self` like the reader callbacks: the tail call can drop the - /// last `Arc` of the embedding `PipeReader` (`Cmd::buffered_output_close` - /// → `close_io` → `detach`), which frees `this`. + /// last `Arc` of the embedding `PipeReader`, which frees `this`. /// /// # Safety /// `this` must be the `CapturedWriter` of a live `PipeReader` (the pointer