shell(windows): report a failed pipe reader start through on_reader_error instead of leaking the subprocess

src/runtime/shell/interpreter.rs

@@ -1562,8 +1562,8 @@ impl Interpreter {
                 // resources: deinit every live `Cmd` (kills the child, frees
                 // the `ShellSubprocess`, readers, redirection fd). Slots stay
                 // occupied so the env walk below still sees pipeline-duped
-                // Cmd envs. Windows: leak-over-UAF, see
-                // `ShellSubprocess::abort_after_failed_start`.
+                // Cmd envs. Windows: leaked, because
+                // `ShellSubprocess::deinit_in_flight_io` is POSIX-only.
                 #[cfg(not(windows))]
                 {
                     let node_count = this.nodes.get().len();

src/runtime/shell/states/Cmd.rs

@@ -1029,7 +1029,7 @@ impl Cmd {
         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);
+            self.exit_code = Some(e.get_errno() as ExitCode);
         }
         let redirect = self.ast_node().redirect;
         let Exec::Subproc(sub) = &mut self.exec else {
@@ -1065,7 +1065,7 @@ impl Cmd {
         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);
+            self.exit_code = Some(e.get_errno() as ExitCode);
         }
         let redirect = self.ast_node().redirect;
         let Exec::Subproc(sub) = &mut self.exec else {

src/runtime/shell/states/CondExpr.rs

@@ -167,8 +167,7 @@ impl CondExpr {
     ) -> Yield {
         let parent = interp.as_condexpr(this).base.parent;
         if let Some(e) = err {
-            // Recover the positive errno (`to_shell_system_error` negated it).
-            let exit_code: ExitCode = e.errno.unsigned_abs() as ExitCode;
+            let exit_code: ExitCode = e.get_errno() as ExitCode;
             return interp.child_done(parent, this, exit_code);
         }
         if matches!(

src/runtime/shell/subproc.rs

@@ -403,56 +403,44 @@ impl ShellSubprocess {
         }
     }
 
-    /// Tear down a subprocess whose stdio start() failed. Marks pending pipe readers as
-    /// errored so PipeReader.deinit's done-assert passes, drops the exit handler so a
-    /// later onProcessExit doesn't touch the freed Subprocess, then deinits.
-    ///
-    /// Windows: PipeReader.deinit asserts the libuv source is closed. Whether the source
-    /// is uv-initialized depends on how far startWithCurrentPipe got, so a blind close or
-    /// destroy is unsafe. Fall back to leaking the Subprocess (pre-existing behavior)
-    /// rather than risk closing an uninitialized handle.
+    /// Tear down a subprocess whose stdin writer `start()` failed. The
+    /// stdout/stderr readers have not started yet: mark them errored, drop
+    /// the exit handler, then free the subprocess.
     fn abort_after_failed_start(this: *mut Self) {
-        #[cfg(windows)]
-        {
-            // SAFETY: `this` is the live allocation; it is deliberately leaked below,
-            // so release the Ctrl+C accounting by hand.
-            unsafe { (*this).ctrl_c_child = None };
-            return;
-        }
-        #[cfg(not(windows))]
-        {
-            // SAFETY: `this` was created via `heap::alloc` in `spawn` and is
-            // uniquely owned here; reclaim and tear down.
-            let mut subproc = unsafe { bun_core::heap::take(this) };
-            for r in [&mut subproc.stdout, &mut subproc.stderr] {
-                if let Readable::Pipe(pipe) = r {
-                    // `start()` failed before any reader callback registered,
-                    // so the `Arc` is expected to be uniquely held. Write
-                    // unconditionally rather than via
-                    // `Arc::get_mut`, which would silently skip the state
-                    // transition if a future change bumped the strong count.
-                    debug_assert_eq!(Arc::strong_count(pipe), 1);
-                    let p = arc_as_mut_ptr(pipe);
-                    // SAFETY: see `arc_as_mut_ptr` — single-threaded shell; no
-                    // other borrow live. Accesses scoped to this statement.
-                    unsafe {
-                        if matches!((*p).state, PipeReaderState::Pending) {
-                            (*p).state = PipeReaderState::Err(None);
-                        }
+        // SAFETY: `this` was created via `heap::alloc` in `spawn` and is
+        // uniquely owned here; reclaim and tear down.
+        let mut subproc = unsafe { bun_core::heap::take(this) };
+        for r in [&mut subproc.stdout, &mut subproc.stderr] {
+            if let Readable::Pipe(pipe) = r {
+                // `start()` failed before any reader callback registered,
+                // so the `Arc` is expected to be uniquely held. Write
+                // unconditionally rather than via
+                // `Arc::get_mut`, which would silently skip the state
+                // transition if a future change bumped the strong count.
+                debug_assert_eq!(Arc::strong_count(pipe), 1);
+                let p = arc_as_mut_ptr(pipe);
+                // SAFETY: see `arc_as_mut_ptr` — single-threaded shell; no
+                // other borrow live. Accesses scoped to this statement.
+                unsafe {
+                    if matches!((*p).state, PipeReaderState::Pending) {
+                        // Close the never-started `uv::Pipe` without
+                        // `on_reader_done`; POSIX closes the fd in `drop`.
+                        #[cfg(windows)]
+                        (*p).reader.deinit();
+                        (*p).state = PipeReaderState::Err(None);
                     }
                 }
             }
-            subproc.proc().set_exit_handler_default();
-            // Dropping `subproc` runs `ShellSubprocess::drop` → `finalize_sync`.
         }
+        subproc.proc().set_exit_handler_default();
+        // Dropping `subproc` runs `ShellSubprocess::drop` → `finalize_sync`.
     }
 
     /// Stop stdio still active because the `Cmd` is deinited mid-flight (VM
     /// shutdown); a no-op after a normal close. Readers stop without firing
     /// `on_reader_done`, queued capture chunks are cancelled (the `IOWriter`
     /// queue holds a raw pointer into the freed `PipeReader`), a pending
-    /// buffer-stdin writer is closed. POSIX-only, same tradeoff as
-    /// [`Self::abort_after_failed_start`].
+    /// buffer-stdin writer is closed. POSIX-only.
     ///
     /// # Safety
     /// `this` must be the live `heap::alloc`'d subprocess with no outstanding
@@ -887,37 +875,26 @@ impl ShellSubprocess {
             return Err(ShellErr::Sys(sys_err));
         }
 
+        // A failed reader start is reported through `on_reader_error`.
         // SAFETY: `subprocess` is live; the slot is passed raw because the
         // reader can complete synchronously and overwrite it via `on_close_io`.
-        if let Err(err) = unsafe {
+        unsafe {
             Readable::start_pipe_reader(
                 &raw mut (*subprocess).stdout,
                 subprocess,
                 event_loop,
                 !spawn_args.lazy,
-            )
-        } {
-            let sys_err = err.to_shell_system_error();
-            // SAFETY: scoped `&mut` for the kill; see above.
-            let _ = unsafe { (*subprocess).try_kill(SignalCode::SIGTERM as i32) };
-            Self::abort_after_failed_start(subprocess);
-            return Err(ShellErr::Sys(sys_err));
+            );
         }
 
         // SAFETY: as for stdout above.
-        if let Err(err) = unsafe {
+        unsafe {
             Readable::start_pipe_reader(
                 &raw mut (*subprocess).stderr,
                 subprocess,
                 event_loop,
                 !spawn_args.lazy,
-            )
-        } {
-            let sys_err = err.to_shell_system_error();
-            // SAFETY: scoped `&mut` for the kill; see above.
-            let _ = unsafe { (*subprocess).try_kill(SignalCode::SIGTERM as i32) };
-            Self::abort_after_failed_start(subprocess);
-            return Err(ShellErr::Sys(sys_err));
+            );
         }
 
         log!("returning");
@@ -1194,24 +1171,24 @@ impl Readable {
         process: *mut ShellSubprocess,
         event_loop: EventLoopHandle,
         eager: bool,
-    ) -> bun_sys::Result<()> {
+    ) {
         // The reader must outlive the re-entrant calls below even if they
         // drop this slot's `Arc`; clone it as a keepalive.
         // SAFETY: caller contract; borrow scoped to the clone.
         let keepalive = match unsafe { &*slot } {
             Readable::Pipe(pipe) => Arc::clone(pipe),
-            _ => return Ok(()),
+            _ => return,
         };
         let p = arc_as_mut_ptr(&keepalive);
-        // SAFETY: see `arc_as_mut_ptr` — single-threaded shell; the
-        // re-entrant reader callbacks only hold raw `*mut PipeReader`, and
-        // each `&mut` below is scoped to its own call.
-        unsafe { (*p).start(process, event_loop) }?;
+        // SAFETY: see `arc_as_mut_ptr` — single-threaded shell; `keepalive`
+        // holds `p` live across both calls, the re-entrant reader callbacks
+        // only hold raw `*mut PipeReader`, and the `&mut` for `read_all` is
+        // scoped to its own call.
+        unsafe { PipeReader::start(p, process, event_loop) };
         if eager {
             // SAFETY: as above.
             unsafe { (*p).read_all() };
         }
-        Ok(())
     }
 
     pub(crate) fn r#ref(&mut self) {
@@ -1796,43 +1773,65 @@ impl PipeReader {
         }
     }
 
-    pub(crate) fn start(
-        &mut self,
+    /// Start reading. A failed start is reported through `on_reader_error`
+    /// (the Cmd records the errno as the exit code), not returned. Raw `this`
+    /// because that dispatch re-enters this allocation through the
+    /// `Readable::Pipe` `Arc`; no `&mut PipeReader` may be live across it.
+    ///
+    /// # Safety
+    /// `this` must point into a live `Arc<PipeReader>` that the caller keeps
+    /// alive across the call (see [`Readable::start_pipe_reader`]).
+    pub(crate) unsafe fn start(
+        this: *mut Self,
         process: *mut ShellSubprocess,
         event_loop: EventLoopHandle,
-    ) -> bun_sys::Result<()> {
-        // self.ref();
-        self.process = Some(process);
-        self.event_loop = event_loop;
+    ) {
+        // SAFETY: caller contract; plain field writes, no callback runs.
+        unsafe {
+            (*this).process = Some(process);
+            (*this).event_loop = event_loop;
+        }
         #[cfg(windows)]
         {
-            return self.reader.start_with_current_pipe();
+            // SAFETY: caller contract; the `&mut reader` ends when the call
+            // returns, before any dispatch.
+            if let bun_sys::Result::Err(err) = unsafe { (*this).reader.start_with_current_pipe() } {
+                // Close the pipe without `on_reader_done`, then report like
+                // a read error. Returning `Err` leaked the subprocess with
+                // its exit handler armed (#43604).
+                // SAFETY: as above.
+                unsafe { (*this).reader.deinit() };
+                // SAFETY: caller contract; nothing touches `this` afterwards.
+                unsafe { Self::on_reader_error(this, &err) };
+            }
         }
 
         // `reader` owns the fd from here; `Drop` closes an un-started one.
+        // `PosixBufferedReader::start` always returns `Ok` and reports a
+        // poll-registration failure through `on_reader_error`.
         #[cfg(not(windows))]
-        match self.reader.start(self.stdio_result.take().unwrap(), true) {
-            bun_sys::Result::Err(err) => bun_sys::Result::Err(err),
-            bun_sys::Result::Ok(()) => {
-                // `reader.start` reports a poll-registration failure through
-                // `on_reader_error` (not its return value), so the reader may
-                // already be errored/torn down here; same guard as
-                // `SubprocessPipeReader::start`.
-                if matches!(self.state, PipeReaderState::Err(_)) {
-                    return Ok(());
-                }
-                #[cfg(unix)]
-                {
-                    // TODO: are these flags correct
-                    if let Some(poll) = self.reader.handle.get_poll() {
-                        poll.set_flag(bun_io::FilePollFlag::Socket);
-                    }
-                    self.reader
-                        .flags
-                        .insert(bun_io::pipe_reader::PosixFlags::SOCKET);
+        {
+            // SAFETY: caller contract; the `&mut reader` ends when `start`
+            // returns, and the keepalive keeps `this` live past the error
+            // dispatch inside it.
+            let errored = unsafe {
+                let fd = (*this).stdio_result.take().unwrap();
+                let _ = (*this).reader.start(fd, true);
+                matches!((*this).state, PipeReaderState::Err(_))
+            };
+            if errored {
+                return;
+            }
+            // SAFETY: caller contract; no callback runs from these flag writes.
+            #[cfg(unix)]
+            unsafe {
+                if let Some(poll) = (*this).reader.handle.get_poll() {
+                    poll.set_flag(bun_io::FilePollFlag::Socket);
                 }
-
-                Ok(())
+                (*this)
+                    .reader
+                    .flags
+                    .insert(bun_io::pipe_reader::PosixFlags::SOCKET);
             }
         }
     }

test/js/bun/shell/shell-pipe-read-fault.test.ts

@@ -2,8 +2,10 @@
 // on the eager read, or epoll_ctl() registering the pipe) so the reader error
 // surfaces synchronously from inside the spawn call. The command must finish
 // with the syscall errno as its exit code, not tear state down under the spawn.
+// The Windows case at the end injects the same kind of failure into
+// uv_read_start through a debug-only env var.
 import { afterAll, beforeAll, expect, test } from "bun:test";
-import { bunEnv, bunExe, isASAN, isLinux, tempDir } from "harness";
+import { bunEnv, bunExe, isASAN, isDebug, isLinux, isWindows, tempDir } from "harness";
 import { join } from "node:path";
 
 const cc = Bun.which("cc") || Bun.which("gcc") || Bun.which("clang");
@@ -576,3 +578,52 @@ test.concurrent.skipIf(!isLinux || !cc)(
     });
   },
 );
+
+// Windows counterpart of the epoll_ctl failures above: uv_read_start on a
+// freshly spawned stdout/stderr pipe fails (libuv reports UV_EINVAL in the
+// wild). The shell returned that error from the spawn, killed the child and
+// leaked the ShellSubprocess with its exit handler still armed. The command
+// failed (exit 1, "Invalid argument"), its Cmd node was freed, and when the
+// killed child's exit callback fired it resolved the stale CmdHandle:
+//   panic: expected Node::Cmd at Node#2, got Free
+// The reader now reports the failure through on_reader_error like POSIX does,
+// so the command finishes with the errno as its exit code and the child's
+// exit is delivered to a live Cmd. A real uv_read_start failure cannot
+// be provoked from JS, so this uses the same debug-only fault injection as
+// test/js/bun/spawn/spawn-pipe-start-error.test.ts.
+const EINVAL = 22;
+
+test.concurrent.skipIf(!isWindows || !isDebug)(
+  "shell survives a failed uv_read_start on the stdout and stderr pipes during spawn (windows)",
+  async () => {
+    const fixture = /* js */ `
+import { $ } from "bun";
+const r = await $\`\${process.execPath} -e 1\`.quiet().nothrow();
+// Wait out one more child lifetime so the first child's exit callback has
+// fired before the process exits. Ignored stdio spawns no pipe reader, so
+// the fault injection does not apply to this one.
+await Bun.spawn({ cmd: [process.execPath, "-e", "1"], stdio: ["ignore", "ignore", "ignore"] }).exited;
+console.log(JSON.stringify({ exitCode: r.exitCode, stderr: r.stderr.toString() }));
+`;
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", fixture, "--debug-crash-handler-use-trace-string"],
+      env: { ...bunEnv, BUN_INTERNAL_FAIL_PIPE_READER_START: "1" },
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+    const line = stdout.trim().split("\n").pop() ?? "";
+    let parsed: unknown;
+    try {
+      parsed = JSON.parse(line);
+    } catch {
+      parsed = line;
+    }
+    // One combined assertion so a crash surfaces stderr and the exit code in the diff.
+    expect({ parsed, stderr, exitCode }).toEqual({
+      parsed: { exitCode: EINVAL, stderr: "" },
+      stderr: "",
+      exitCode: 0,
+    });
+  },
+);