diff --git a/src/runtime/shell/states/Cmd.rs b/src/runtime/shell/states/Cmd.rs index 2b362fc02636..6ac403149be6 100644 --- a/src/runtime/shell/states/Cmd.rs +++ b/src/runtime/shell/states/Cmd.rs @@ -68,12 +68,7 @@ impl Cmd { pub struct SubprocExec { pub(crate) child: *mut ShellSubprocess, pub(crate) buffered_closed: BufferedIoClosed, - /// 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`) - /// to drive. + /// Null until the spawn has returned; gates the `Yield` built by `finish_if_done`. pub(crate) interp: *mut Interpreter, pub(crate) this_id: NodeId, } @@ -883,7 +878,7 @@ impl Cmd { core::mem::take(&mut me.exec) }; // `me`'s borrow ended above: the teardown below re-enters this Cmd - // via stdin `on_close_io` → `buffered_input_close` (a no-op once + // via stdin `on_stdin_writer_close` → `buffered_input_close` (a no-op once // `exec` is taken). match exec { Exec::None => {} @@ -919,11 +914,7 @@ impl Cmd { } // ── Subprocess callbacks (legacy `*Cmd` backref shape) ──────────────── - // `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 - // `&mut self`. + // Each returns its Yield: running it can free this Cmd and its subprocess. /// True once the command has both an exit code and (for subprocesses) /// all buffered stdio closed. @@ -939,11 +930,13 @@ impl Cmd { } } - /// Mark the subprocess's buffered stdin as closed. - pub(crate) fn buffered_input_close(&mut self) { - if let Exec::Subproc(sub) = &mut self.exec { - sub.buffered_closed.close_stdin(); - } + /// Mark the subprocess's buffered stdin as closed; this may be the close that finishes it. + pub(crate) fn buffered_input_close(&mut self) -> Yield { + let Exec::Subproc(sub) = &mut self.exec else { + return Yield::suspended(); + }; + sub.buffered_closed.close_stdin(); + self.finish_if_done() } /// Mark the subprocess's buffered stdout/stderr as closed (flushing the @@ -958,27 +951,32 @@ impl Cmd { OutKind::Stdout => self.buffered_output_close_stdout(err), OutKind::Stderr => self.buffered_output_close_stderr(err), } - if self.has_finished() { - // Set `state = Done` and hand the Yield back to the caller - // (`PipeReader::run_yield`), which drives the trampoline with the - // `*mut Interpreter` it already holds, landing in `Cmd::next` → - // `CmdState::Done` → `interp.child_done(...)`. - self.state = CmdState::Done; - let (interp, this_id) = match &self.exec { - Exec::Subproc(sub) => (sub.interp, sub.this_id), - // Only the subprocess path calls this; builtin output goes - // through `Builtin::done` → `on_exec_done`. - _ => return Yield::suspended(), - }; - // Same gate as `on_exit`: `exec.interp` stays null until the spawn - // returns, so a Yield run here would reach `Cmd::deinit` and free the - // `ShellSubprocess` still on the spawn frame. `transition_to_exec` resumes. - if interp.is_null() { - return Yield::suspended(); - } - return Yield::Next(this_id); + self.finish_if_done() + } + + /// Called by `ShellSubprocess::on_process_exit`. + pub(crate) fn on_exit(&mut self, exit_code: ExitCode) -> Yield { + log!("cmd exit code={}", exit_code); + self.exit_code = Some(exit_code); + self.finish_if_done() + } + + /// `Done` once the exit code and every piped stdio are in; the caller runs the Yield. + fn finish_if_done(&mut self) -> Yield { + if !self.has_finished() { + return Yield::suspended(); } - Yield::suspended() + self.state = CmdState::Done; + let (interp, this_id) = match &self.exec { + Exec::Subproc(sub) => (sub.interp, sub.this_id), + // Builtins finish through `Builtin::done` → `on_exec_done` instead. + _ => return Yield::suspended(), + }; + // Still inside the spawn (see `transition_to_exec`, which resumes from `state`). + if interp.is_null() { + return Yield::suspended(); + } + Yield::Next(this_id) } fn buffered_output_close_stdout(&mut self, err: Option) { @@ -1050,31 +1048,6 @@ impl Cmd { ); child.close_io(StdioKind::Stderr); } - - /// Called by `ShellSubprocess::on_process_exit`. - pub(crate) fn on_exit(&mut self, exit_code: ExitCode) { - self.exit_code = Some(exit_code); - let has_finished = self.has_finished(); - log!("cmd exit code={} has_finished={}", exit_code, has_finished); - if has_finished { - self.state = CmdState::Done; - // `self` lives inside `interp.nodes`, so resume via the stashed - // backrefs. - let (interp, this_id) = match &self.exec { - Exec::Subproc(sub) => (sub.interp, sub.this_id), - _ => return, - }; - if interp.is_null() { - return; - } - // SAFETY: `interp` outlives every spawned subprocess (it owns the - // arena slot containing `self`). `&mut self` is dead by NLL after - // this point so the `&Interpreter` borrow does not alias it. - // The caller (`ShellSubprocess::on_process_exit`) does not touch - // its `*mut Cmd` again after this returns. - Yield::Next(this_id).run(unsafe { &*interp }); - } - } } fn set_stdio_from_redirect(stdio: &mut [Stdio; 3], flags: ast::RedirectFlags, fd: bun_sys::Fd) { diff --git a/src/runtime/shell/subproc.rs b/src/runtime/shell/subproc.rs index a6ea81124a14..ef89368218c7 100644 --- a/src/runtime/shell/subproc.rs +++ b/src/runtime/shell/subproc.rs @@ -3,9 +3,7 @@ use std::sync::Arc; #[cfg(unix)] use crate::api::bun::process::SpawnResultExt as _; -use crate::api::bun::process::{ - self as bun_process, Process, Rusage, SignalCodeExt, SpawnOptions, Status, -}; +use crate::api::bun::process::{self as bun_process, Process, SignalCodeExt, SpawnOptions, Status}; #[cfg(windows)] use crate::api::bun::process::{WindowsOptions, WindowsStdioResult}; use crate::api::bun::subprocess as JscSubprocess; @@ -276,15 +274,19 @@ impl JscSubprocess::static_pipe_writer::StaticPipeWriterProcess for ShellSubproc const POLL_OWNER_TAG: bun_io::PollTag = bun_io::posix_event_loop::poll_tag::SHELL_STATIC_PIPE_WRITER; unsafe fn on_close_io(this: *mut Self, kind: StdioKind) { - // SAFETY: caller (StaticPipeWriter) guarantees `this` is live. - unsafe { (*this).on_close_io(kind) } + // `Writable::init` only ever creates the writer for stdin. + debug_assert!(matches!(kind, StdioKind::Stdin)); + // SAFETY: `StaticPipeWriter::on_close` passes its live process backref and does not + // touch it afterwards. Forwarded raw (not autoref'd): the callee may free `*this`. + unsafe { Self::on_stdin_writer_close(this) } } } bun_spawn::link_impl_ProcessExit! { Shell for ShellSubprocess => |this| { - on_process_exit(process, status, rusage) => - (*this).on_process_exit(&*process, &status, rusage), + // Forwarded raw, not autoref'd: the callee may free `*this`. + on_process_exit(_process, status, _rusage) => + ShellSubprocess::on_process_exit(this, &status), } } @@ -299,15 +301,42 @@ impl ShellSubprocess { unsafe { &mut *self.process } } - pub(crate) fn on_static_pipe_writer_done(&mut self) { + /// The `< ${buffer}` stdin writer closed: release it and let the `Cmd` finish. + /// + /// # Safety + /// `this` must be live and unborrowed. It may be freed (by `Cmd::deinit`, reached + /// through the Yield run here) by the time this returns, which is why it is raw. + unsafe fn on_stdin_writer_close(this: *mut Self) { + { + // SAFETY: caller contract; the borrow ends with this block, before the Yield runs. + let slot = unsafe { &mut (*this).stdin }; + match core::mem::replace(slot, Writable::Ignore) { + Writable::Buffer(buffer) => { + // Releases `create()`'s ref; `start()`'s keeps the writer alive until + // this callback returns. + // SAFETY: single-threaded; sole borrow of the payload. + unsafe { buffer_mut(&buffer) }.source.detach(); + buffer.deref(); + } + // The writer only exists while the slot holds it. + other => { + *slot = other; + return; + } + } + } + // SAFETY: caller contract; `CmdHandle` is `Copy`, no borrow is kept. + let handle = unsafe { (*this).cmd_parent }; log!( - "Subproc(0x{:x}) onStaticPipeWriterDone(cmd={})", - std::ptr::from_mut(self) as usize, - self.cmd_parent.id + "Subproc(0x{:x}) onStdinWriterClose(cmd={})", + this as usize, + handle.id ); - // SAFETY: cmd_parent backref resolves to the owning Cmd which outlives - // the subprocess (freed only in `Cmd::deinit` after all stdio closes). - unsafe { self.cmd_parent.cmd_mut() }.buffered_input_close(); + // SAFETY: the owning Cmd outlives its subprocess (its `deinit` is what frees it); + // the `&mut Cmd` ends before the Yield runs. + let y = unsafe { handle.cmd_mut() }.buffered_input_close(); + // May free `*this`. + y.run(&handle.interp); } pub(crate) fn has_exited(&self) -> bool { @@ -392,62 +421,37 @@ impl ShellSubprocess { self.close_io(StdioKind::Stderr); } + /// A stdout/stderr `PipeReader` finished: swap it out of its slot for its buffered bytes. pub(crate) fn on_close_io(&mut self, kind: StdioKind) { - match kind { - StdioKind::Stdin => match &mut self.stdin { - Writable::Pipe(pipe) => { - // DerefMut on the owning `&mut FileSinkPtr` encapsulates - // the access. - pipe.source.with_mut(|s| s.clear()); - // FileSinkPtr::drop derefs. - self.stdin = Writable::Ignore; - } - Writable::Buffer(_) => { - self.on_static_pipe_writer_done(); - // RefPtr has no Drop — move it out before reassigning so the - // create ref is actually released. - if let Writable::Buffer(buffer) = - core::mem::replace(&mut self.stdin, Writable::Ignore) - { - // SAFETY: single-threaded; sole borrow of the payload. - unsafe { buffer_mut(&buffer) }.source.detach(); - buffer.deref(); - } - } - _ => {} - }, - StdioKind::Stdout | StdioKind::Stderr => { - let out: &mut Readable = match kind { - StdioKind::Stdout => &mut self.stdout, - StdioKind::Stderr => &mut self.stderr, - StdioKind::Stdin => unreachable!(), - }; - if let Readable::Pipe(pipe) = core::mem::replace(out, Readable::Ignore) { - // The only callers reach here from inside - // `PipeReader::on_reader_done`/`on_reader_error`, which still - // hold a raw `*mut PipeReader` to this same allocation. - // Route every read/write through `Arc::as_ptr` (no `Deref`) - // so we never materialise a `&PipeReader` that would alias - // those callers' access; see `PipeReader::take_done_buffer`. - let pp = Arc::as_ptr(&pipe).cast_mut(); - // SAFETY: `pp` projects from the Arc allocation's NonNull; - // raw place read of the discriminant + raw-ptr write - // through `take_done_buffer` (see its doc). - let buf = unsafe { - if matches!(&(*pp).state, PipeReaderState::Done(_)) { - Some(PipeReader::take_done_buffer(pp)) - } else { - None - } - }; - if let Some(buf) = buf { - *out = Readable::Buffer(buf); - } else { - *out = Readable::Ignore; - } - drop(pipe); // deref + let out: &mut Readable = match kind { + StdioKind::Stdout => &mut self.stdout, + StdioKind::Stderr => &mut self.stderr, + StdioKind::Stdin => unreachable!("stdin closes through on_stdin_writer_close"), + }; + if let Readable::Pipe(pipe) = core::mem::replace(out, Readable::Ignore) { + // The only callers reach here from inside + // `PipeReader::on_reader_done`/`on_reader_error`, which still + // hold a raw `*mut PipeReader` to this same allocation. + // Route every read/write through `Arc::as_ptr` (no `Deref`) + // so we never materialise a `&PipeReader` that would alias + // those callers' access; see `PipeReader::take_done_buffer`. + let pp = Arc::as_ptr(&pipe).cast_mut(); + // SAFETY: `pp` projects from the Arc allocation's NonNull; + // raw place read of the discriminant + raw-ptr write + // through `take_done_buffer` (see its doc). + let buf = unsafe { + if matches!(&(*pp).state, PipeReaderState::Done(_)) { + Some(PipeReader::take_done_buffer(pp)) + } else { + None } + }; + if let Some(buf) = buf { + *out = Readable::Buffer(buf); + } else { + *out = Readable::Ignore; } + drop(pipe); // deref } } @@ -503,11 +507,11 @@ impl ShellSubprocess { /// # Safety /// `this` must be the live `heap::alloc`'d subprocess with no outstanding /// borrows; single-threaded shell. Raw (not `&mut self`) because the - /// stdin close re-enters `on_close_io(&mut Self)` through the writer's + /// stdin close re-enters `on_stdin_writer_close` through the writer's /// process backref. #[cfg(not(windows))] pub(crate) unsafe fn deinit_in_flight_io(this: *mut Self) { - // Claim `start()`'s +1, `close()` (fires `on_close` → `on_close_io`: + // Claim `start()`'s +1, `close()` (fires `on_close` → `on_stdin_writer_close`: // slot → `Ignore`, `create()`'s ref released), release the claimed // ref — the JS `Subprocess::close_io` stdin shape. // SAFETY: caller contract; the `stdin` borrow ends before `close()`. @@ -958,8 +962,13 @@ impl ShellSubprocess { Ok(()) } - pub(crate) fn on_process_exit(&mut self, _: &Process, status: &Status, _: &Rusage) { - log!("onProcessExit({:x})", std::ptr::from_mut(self) as usize); + /// Exit handler (`link_impl_ProcessExit!` above). + /// + /// # Safety + /// Same contract as [`Self::on_stdin_writer_close`]: `this` must be live and unborrowed, + /// and may be freed by the time this returns. + unsafe fn on_process_exit(this: *mut Self, status: &Status) { + log!("onProcessExit({:x})", this as usize); let exit_code: Option = 'brk: { if let Status::Exited(exited) = &status { break 'brk Some(exited.code); @@ -978,16 +987,18 @@ impl ShellSubprocess { break 'brk None; }; - if let Some(code) = exit_code { - let handle = self.cmd_parent; - // SAFETY: cmd_parent backref outlives subprocess; resolved - // through the node arena so it survives `Vec` reallocation. - // `&mut self` is dead by NLL before `on_exit` re-enters interp. - let cmd = unsafe { handle.cmd_mut() }; - if cmd.exit_code.is_none() { - cmd.on_exit(code.into()); - } + let Some(code) = exit_code else { return }; + // SAFETY: caller contract; `CmdHandle` is `Copy`, no borrow is kept. + let handle = unsafe { (*this).cmd_parent }; + // SAFETY: the owning Cmd outlives its subprocess (its `deinit` is what frees it); + // the `&mut Cmd` ends before the Yield runs. + let cmd = unsafe { handle.cmd_mut() }; + if cmd.exit_code.is_some() { + return; } + let y = cmd.on_exit(code.into()); + // May free `*this`. + y.run(&handle.interp); } } diff --git a/src/spawn/static_pipe_writer.rs b/src/spawn/static_pipe_writer.rs index 2d6d006f1054..0a307503f87a 100644 --- a/src/spawn/static_pipe_writer.rs +++ b/src/spawn/static_pipe_writer.rs @@ -23,12 +23,11 @@ bun_output::declare_scope!(StaticPipeWriter, hidden); pub trait StaticPipeWriterProcess { const POLL_OWNER_TAG: bun_io::PollTag; /// # Safety - /// `this` must point to a live `Self`. + /// `this` must be a live `Self`. Called once, from the writer's close; the impl may free it. unsafe fn on_close_io(this: *mut Self, kind: StdioKind); } /// Generic over the owning process type (e.g. `Subprocess`, `ShellSubprocess`). -/// `P` must expose `fn on_close_io(&mut self, kind: StdioKind)`. // Cleanup lives in `impl Drop` below; the final Box free is // the derive's default destructor (`drop(heap::take(this))`). #[derive(bun_ptr::RefCounted)] @@ -38,7 +37,7 @@ pub struct StaticPipeWriter { pub(crate) writer: IOWriter

, pub(crate) stdio_result: StdioResult, pub source: Source, - /// BACKREF: parent process is notified on close; never owned/destroyed here. + /// BACKREF: parent process, notified (and nulled) on close; never owned/destroyed here. pub(crate) process: *mut P, pub(crate) event_loop: EventLoopHandle, /// True while `start()`'s `+1` ref is outstanding. @@ -270,9 +269,11 @@ impl StaticPipeWriter

{ // frees that storage so no dangling slice survives the close. self.buffer = RawSlice::EMPTY; self.source.detach(); - // SAFETY: `process` is a backref to the owning process, guaranteed alive - // for the lifetime of this writer (the process owns/outlives its stdio writers). - unsafe { P::on_close_io(self.process, StdioKind::Stdin) }; + let process = core::mem::replace(&mut self.process, core::ptr::null_mut()); + // SAFETY: the owning process keeps this backref alive until it has been told + // the writer closed, which is this (single) call; it may be freed by it, and + // the field is already nulled so nothing here can use it afterwards. + unsafe { P::on_close_io(process, StdioKind::Stdin) }; #[cfg(windows)] if release_start_ref { // SAFETY: `started` was the token for start()'s outstanding +1; diff --git a/test/js/bun/shell/bunshell.test.ts b/test/js/bun/shell/bunshell.test.ts index 95a990ba68e7..73d07aa3aedd 100644 --- a/test/js/bun/shell/bunshell.test.ts +++ b/test/js/bun/shell/bunshell.test.ts @@ -3109,6 +3109,137 @@ describe("stdin redirect from a zero-length buffer delivers EOF to the spawned c }); }); +describe("stdin redirect still held open by a helper after the command's process has exited", () => { + // `< ${buf}` is pumped into the child over a pipe. The child hands its stdin + // to a detached helper and exits without reading any of it, and the helper + // only acts once the test has seen the child disappear, so the exit code and + // the stdout/stderr EOFs have been processed by the time the stdin side + // closes: that close is what has to complete the command. The helper either + // just exits (the pending write fails) or drains the redirect, which it must + // receive in full: the shell keeps pumping for whoever still holds the pipe. + const SIZE = 4 << 20; // far more than a pipe or socketpair buffers, so the write is still pending when the child exits + // Each helper records what it did in RESULT_FILE once released, so a helper + // that died early (which would also fail the pending write) cannot pass as one + // that exited on cue. + const helper = (resultExpression: string) => ` + const deadline = Date.now() + 60_000; + while (!(await Bun.file(process.env.RELEASE_FILE).exists()) && Date.now() < deadline) await Bun.sleep(5); + await Bun.write(process.env.RESULT_FILE, ${resultExpression}); + `; + const exitingHelper = helper(`"released"`); + const drainingHelper = helper(`"read " + (await Bun.stdin.bytes()).length + " bytes"`); + const childCode = ` + Bun.spawn({ + cmd: [process.execPath, "-e", process.env.HELPER_CODE], + stdin: "inherit", + stdout: "ignore", + stderr: Bun.file(process.env.HELPER_STDERR), + detached: true, + }).unref(); + await Bun.write(process.env.PID_FILE, String(process.pid)); + console.log("child stdout"); + console.error("child stderr"); + process.exitCode = 3; + `; + + async function until(what: string, poll: () => T | Promise): Promise { + const deadline = Date.now() + 30_000; + while (true) { + const value = await poll(); + if (value) return value; + if (Date.now() > deadline) throw new Error(`${what} did not happen within 30s`); + await Bun.sleep(5); + } + } + async function fileContents(path: string) { + return (await Bun.file(path).exists()) ? Bun.file(path).text() : ""; + } + function hasExited(pid: number) { + try { + process.kill(pid, 0); + return false; + } catch { + return true; + } + } + + async function run(helperCode: string, command: (env: Record) => $.ShellPromise) { + using dir = tempDir("shell-stdin-held-open", {}); + const env: Record = { + ...bunEnv, + HELPER_CODE: helperCode, + HELPER_STDERR: join(String(dir), "helper.stderr"), + PID_FILE: join(String(dir), "pid"), + RELEASE_FILE: join(String(dir), "release"), + RESULT_FILE: join(String(dir), "result"), + }; + // The ASAN CI lanes run with this set, which makes the child SIGKILL the helper when it exits. + delete env.BUN_FEATURE_FLAG_NO_ORPHANS; + const running = command(env).then(r => r); // `$` is lazy; start it now. + let pid = 0; + try { + pid = Number(await until("child pid file", () => fileContents(env.PID_FILE!))); + await until("child exit", () => hasExited(pid)); + } finally { + // Also on the failure path, so the helper and the command wind down now rather + // than lingering for the rest of the file. + await Bun.write(env.RELEASE_FILE!, ""); + if (pid !== 0 && !hasExited(pid)) process.kill(pid, "SIGKILL"); + await running; + } + const result = await running; + const helperResult = await until("helper result", () => fileContents(env.RESULT_FILE!)).catch(async error => { + throw new Error(`${error.message}; helper stderr: ${JSON.stringify(await fileContents(env.HELPER_STDERR!))}`); + }); + return { + stdout: result.stdout.toString(), + stderr: result.stderr.toString(), + exitCode: result.exitCode, + helper: helperResult, + }; + } + + const redirects: Array<[string, () => Buffer | Blob]> = [ + ["Buffer", () => Buffer.alloc(SIZE, "a")], + ["Blob", () => new Blob([Buffer.alloc(SIZE, "a")])], + ]; + + test.concurrent.each(redirects)("helper exits without reading: %s", async (_name, input) => { + const out = await run(exitingHelper, env => $`${BUN} -e ${childCode} < ${input()}`.env(env).quiet().nothrow()); + expect(out).toEqual({ stdout: "child stdout\n", stderr: "child stderr\n", exitCode: 3, helper: "released" }); + }); + + test.concurrent.each(redirects)("helper drains the redirect: %s", async (_name, input) => { + const out = await run(drainingHelper, env => $`${BUN} -e ${childCode} < ${input()}`.env(env).quiet().nothrow()); + expect(out).toEqual({ + stdout: "child stdout\n", + stderr: "child stderr\n", + exitCode: 3, + helper: `read ${SIZE} bytes`, + }); + }); + + test.concurrent("inside a pipeline", async () => { + const upper = "process.stdout.write((await Bun.stdin.text()).toUpperCase())"; + const out = await run(exitingHelper, env => + $`${BUN} -e ${childCode} < ${Buffer.alloc(SIZE, "a")} | ${BUN} -e ${upper}`.env(env).quiet().nothrow(), + ); + expect(out).toEqual({ stdout: "CHILD STDOUT\n", stderr: "child stderr\n", exitCode: 0, helper: "released" }); + }); + + test.concurrent("as the left side of ||", async () => { + const out = await run(exitingHelper, env => + $`${BUN} -e ${childCode} < ${Buffer.alloc(SIZE, "a")} || echo failed`.env(env).quiet().nothrow(), + ); + expect(out).toEqual({ + stdout: "child stdout\nfailed\n", + stderr: "child stderr\n", + exitCode: 0, + helper: "released", + }); + }); +}); + test("output redirect buffer for an external command stays attached until the command finishes", async () => { // `> ${buf}` for an external (non-builtin) command stores the buffer and // copies the child's stdout into it as chunks arrive across event-loop