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 6057a967e208..6fe337597398 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,11 +68,15 @@ impl Cmd { pub struct SubprocExec { pub(crate) child: *mut ShellSubprocess, pub(crate) buffered_closed: BufferedIoClosed, + /// 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 /// 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, @@ -564,6 +568,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, })); @@ -922,7 +927,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) @@ -952,15 +957,16 @@ 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 - // (`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; @@ -981,12 +987,27 @@ impl Cmd { Yield::suspended() } - fn buffered_output_close_stdout(&mut self, err: Option) { + /// 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(_))); - 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 +1038,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 +1071,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 5c4ed46bf625..94ab19300b1a 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]) { @@ -1653,6 +1671,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,43 +1682,57 @@ 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); + 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`, 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); - } else 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: `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). - unsafe { PipeReader::try_signal_done_to_cmd(self.parent_mut()) } } } @@ -1755,17 +1788,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 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`). + /// 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!( @@ -1897,31 +1922,33 @@ 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) - ); + /// `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'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 { + (*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. + // 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 } @@ -1929,7 +1956,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`. /// @@ -2044,7 +2071,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`. @@ -2053,18 +2080,47 @@ impl PipeReader { Yield::Suspended } - fn take_captured_error(&mut self) -> Option { + /// 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 + /// 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. + 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) + } + } + + /// 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)); } } } @@ -2072,7 +2128,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 } @@ -2224,14 +2280,20 @@ 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`: 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; - 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..1564063c72c9 --- /dev/null +++ b/test/js/bun/shell/epipe-relay-fixture.ts @@ -0,0 +1,88 @@ +// 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. 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 { existsSync, readFileSync, writeFileSync, 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. + const report = `stdout write failed: ${e.code}`; + console.error(`producer: ${report}`); + writeFileSync(process.env.PRODUCER_REPORT!, report); + process.exit(0); + } + Bun.sleepSync(1); + } +} + +if (mode === "produce-once") { + writeSync(1, "y\n"); + 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. +while (true) { + try { + writeSync(1, "still has a reader\n"); + } catch (e: any) { + if (e.code === "EPIPE" || e.code === "ECONNRESET") 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. 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": + 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}`); +} +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 27f6eabe8302..3d8ae1b3e95a 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, tempDir } from "harness"; +import { join } from "node:path"; import { createTestBuilder } from "./test_builder"; const TestBuilder = createTestBuilder(import.meta.path); @@ -20,3 +21,70 @@ 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 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, reportPath], + 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), + // 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); + 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 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"); + }); + + 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"); + }); +});