spawn: release the stdio PipeReader through its pointer, not a &mut receiver

src/runtime/api/bun/js_bun_spawn_bindings.rs

@@ -28,6 +28,7 @@ use crate::api::bun_process::SpawnResultExt as _;
 use crate::api::bun_process::{self as spawn, CStrPtr, Process, Rusage, SpawnOptions};
 // User-facing JS `Stdio` enum (extract/as_spawn_option/is_piped).
 use crate::api::bun_spawn::stdio::{self, Stdio};
+use crate::api::bun_subprocess::subprocess_pipe_reader::PipeReader;
 use crate::api::bun_subprocess::{
     self as Subprocess, Readable, Subprocess as SubprocessT, Writable,
 };
@@ -1742,25 +1743,44 @@ fn spawn_maybe_sync<const IS_SYNC: bool>(
     // Start the readers before the Writable::Buffer stdin writer so that if
     // the writer's start() throws below, both PipeReaders have taken their
     // start() ref and on_process_exit's later drain is refcount-balanced.
+    // Both calls go through the reader's own pointer because either may end
+    // the reader's life before returning (a failed start, or EOF inside
+    // read_all), at which point on_close_io has already replaced the slot, so
+    // the slot is re-read in between instead of reusing `pipe`.
     if let Readable::Pipe(pipe) = subprocess.stdout.get() {
-        // Note: pass `subprocess_nn` (the `NonNull<Subprocess<'static>>`
-        // captured above) instead of the live `&mut subprocess`, which would
-        // alias with the `&mut subprocess.stdout` borrow held by `pipe`.
-        Readable::pipe_reader_mut(pipe).start(subprocess_nn, event_loop_nn, !IS_SYNC && lazy);
-        if (IS_SYNC || !lazy) && matches!(subprocess.stdout.get(), Readable::Pipe(_)) {
+        // SAFETY: the slot holds a ref on a live reader and no borrow of it is
+        // held here; `subprocess_nn` is the heap-pinned Subprocess.
+        unsafe {
+            PipeReader::start(
+                pipe.as_ptr(),
+                subprocess_nn,
+                event_loop_nn,
+                !IS_SYNC && lazy,
+            )
+        };
+        if IS_SYNC || !lazy {
             if let Readable::Pipe(pipe) = subprocess.stdout.get() {
-                Readable::pipe_reader_mut(pipe).read_all();
+                // SAFETY: as above; the slot still holding `Pipe` means start()
+                // succeeded and the reader is live.
+                unsafe { PipeReader::read_all(pipe.as_ptr()) };
             }
         }
     }
 
     if let Readable::Pipe(pipe) = subprocess.stderr.get() {
-        // Note: see stdout arm above — avoid aliased &mut.
-        Readable::pipe_reader_mut(pipe).start(subprocess_nn, event_loop_nn, !IS_SYNC && lazy);
-
-        if (IS_SYNC || !lazy) && matches!(subprocess.stderr.get(), Readable::Pipe(_)) {
+        // SAFETY: see the stdout arm.
+        unsafe {
+            PipeReader::start(
+                pipe.as_ptr(),
+                subprocess_nn,
+                event_loop_nn,
+                !IS_SYNC && lazy,
+            )
+        };
+        if IS_SYNC || !lazy {
             if let Readable::Pipe(pipe) = subprocess.stderr.get() {
-                Readable::pipe_reader_mut(pipe).read_all();
+                // SAFETY: see the stdout arm.
+                unsafe { PipeReader::read_all(pipe.as_ptr()) };
             }
         }
     }

src/runtime/api/bun/subprocess/SubprocessPipeReader.rs

@@ -32,7 +32,7 @@ pub enum State {
 #[ref_count(destroy = PipeReader::deinit, debug_name = "PipeReader")]
 pub struct PipeReader {
     pub(crate) reader: IOReader,
-    // Backref to owning Subprocess; cleared in detach()/onReaderDone()/onReaderError().
+    // Backref to owning Subprocess; cleared in detach()/finish().
     // `ParentRef` encapsulates the single unsafe deref behind a safe `Deref`/`get()`;
     // the Subprocess owns this PipeReader (via `Readable::Pipe`) and is guaranteed
     // live whenever `process.is_some()` — see `on_close_io`/`finalize` ordering.
@@ -136,55 +136,87 @@ impl PipeReader {
         }
     }
 
-    pub(crate) fn read_all(&mut self) {
-        if matches!(self.state, State::Pending) {
-            // SAFETY: `self.reader` is live; `read` is the raw
-            // re-entrancy-safe entry (its dispatch runs user JS).
-            unsafe { IOReader::read(&raw mut self.reader) };
+    /// Drives the reader synchronously. EOF or an error inside the read
+    /// reaches `on_reader_done`/`on_reader_error`, which may release the last
+    /// ref, so `*this` may be gone on return.
+    ///
+    /// # Safety
+    /// `this` must point to a live, started `PipeReader`; no `&`/`&mut` to
+    /// `*this` may be live across the call.
+    pub(crate) unsafe fn read_all(this: *mut Self) {
+        // SAFETY: caller contract; nothing borrowed from `*this` is held when
+        // `read` (the raw re-entrancy-safe entry, whose done/error dispatch may
+        // free `*this`) runs.
+        unsafe {
+            if matches!((*this).state, State::Pending) {
+                IOReader::read(&raw mut (*this).reader);
+            }
         }
     }
 
-    pub(crate) fn start(
-        &mut self,
+    /// Takes the reader's own ref for the read in flight; `finish` releases it
+    /// once the read ends. If registering the pipe fails, `finish` runs before
+    /// this returns and `*this` is freed on return, so callers must re-read the
+    /// `Readable` slot instead of reusing `this`.
+    ///
+    /// # Safety
+    /// `this` must point to a live `PipeReader` from `create()`; no `&`/`&mut`
+    /// to `*this` may be live across the call.
+    pub(crate) unsafe fn start(
+        this: *mut Self,
         process: NonNull<Subprocess<'static>>,
         event_loop: NonNull<EventLoop>,
         lazy: bool,
     ) {
-        self.r#ref();
-        self.process = Some(ParentRef::from(process));
-        self.event_loop = event_loop.into();
-        self.event_loop_handle = bun_jsc::EventLoopHandle::init(event_loop.as_ptr().cast::<()>());
+        // SAFETY: caller contract; each borrow ends at its `;`.
+        unsafe {
+            (*this).r#ref();
+            (*this).process = Some(ParentRef::from(process));
+            (*this).event_loop = event_loop.into();
+            (*this).event_loop_handle =
+                bun_jsc::EventLoopHandle::init(event_loop.as_ptr().cast::<()>());
+        }
         #[cfg(windows)]
         {
             if lazy {
                 // Leave IS_PAUSED set (the init default) so uv_read_start is
                 // deferred until JS first pulls; the kernel pipe buffer then
                 // provides backpressure and the child blocks.
-                let reader_ptr = core::ptr::from_mut(&mut self.reader).cast::<core::ffi::c_void>();
-                if let Some(source) = self.reader.source.as_mut() {
-                    source.set_data(reader_ptr);
+                // SAFETY: caller contract; only `reader` is borrowed, and the
+                // borrow ends with the block.
+                unsafe {
+                    let reader = &raw mut (*this).reader;
+                    if let Some(source) = (*reader).source.as_mut() {
+                        source.set_data(reader.cast::<core::ffi::c_void>());
+                    }
+                    (*reader)
+                        .flags
+                        .remove(bun_io::pipe_reader::WindowsFlags::IS_DONE);
                 }
-                self.reader
-                    .flags
-                    .remove(bun_io::pipe_reader::WindowsFlags::IS_DONE);
                 return;
             }
-            // Hold one more ref so `self` survives the on_reader_error() teardown
-            // below long enough to return; matches the POSIX keepalive.
+            // The failure path below releases both the Readable's ref (via
+            // on_close_io) and the ref taken above; the guard keeps `*this`
+            // allocated until this returns. Its drop is then the final
+            // release, made through `this` with no borrow of `*this` live.
             //
-            // SAFETY: `self` is live; ScopedRef bumps the intrusive refcount and
-            // derefs on Drop. The deref may free `*self`, but no borrow of `self`
-            // outlives the guard's drop on return.
-            let _keepalive = unsafe { ScopedRef::new(std::ptr::from_mut::<PipeReader>(self)) };
-            if let bun_sys::Result::Err(err) = self.reader.start_with_current_pipe() {
+            // SAFETY: caller contract.
+            let _keepalive = unsafe { ScopedRef::new(this) };
+            // SAFETY: caller contract; the `reader` borrow ends when the call
+            // returns.
+            let started = unsafe { (*this).reader.start_with_current_pipe() };
+            if let bun_sys::Result::Err(err) = started {
                 // Route through the same teardown as a read-callback error
                 // (matches POSIX's register_poll failure path): state=Err,
                 // detach from the Subprocess via on_close_io, release the
                 // start() ref, and let the caller proceed to the sibling pipe.
                 // Returning Err would have the caller throw after try_kill
                 // without unwinding this pipe or the never-started sibling,
                 // and on_process_exit's later drain then double-derefs them.
-                self.on_reader_error(err);
+                //
+                // SAFETY: `_keepalive` keeps `*this` live across the call; no
+                // borrow of `*this` is live.
+                unsafe { Self::on_reader_error(this, err) };
             }
         }
 
@@ -193,35 +225,45 @@ impl PipeReader {
             if lazy {
                 // Defer poll registration until JS first pulls so the kernel
                 // pipe buffer provides backpressure and the child blocks.
-                self.reader.flags.insert(PosixFlags::IS_PAUSED);
+                //
+                // SAFETY: caller contract; the borrow ends at the `;`.
+                unsafe { (*this).reader.flags.insert(PosixFlags::IS_PAUSED) };
             }
-            // PosixBufferedReader.start() always returns Ok(()); if poll
-            // registration fails it synchronously invokes onReaderError() first,
-            // which drops both the Readable.pipe ref (via onCloseIO) and the ref we
-            // just took above. Hold one more ref so `this` survives long enough to
-            // check state after start() returns.
+            // PosixBufferedReader::start() always returns Ok(()); if poll
+            // registration fails it dispatches on_reader_error synchronously,
+            // which releases both the Readable's ref (via on_close_io) and the
+            // ref taken above. The guard keeps `*this` allocated for the state
+            // check below; its drop is then the final release, made through
+            // `this` with no borrow of `*this` live.
             //
-            // SAFETY: `self` is live; ScopedRef bumps the intrusive refcount and
-            // derefs on Drop. The deref may free `*self`, but no borrow of `self`
-            // outlives the guard's drop on return.
-            let _keepalive = unsafe { ScopedRef::new(std::ptr::from_mut::<PipeReader>(self)) };
+            // SAFETY: caller contract.
+            let _keepalive = unsafe { ScopedRef::new(this) };
 
-            let _ = self.reader.start(self.stdio_result.unwrap(), true);
+            // SAFETY: caller contract; the borrow ends at the `;`.
+            let fd = unsafe { (*this).stdio_result.unwrap() };
+            // SAFETY: caller contract. Only `reader` is borrowed for the call;
+            // the on_reader_error it may dispatch reaches the other fields
+            // through the raw parent pointer, and the borrow ends on return.
+            let _ = unsafe { (*this).reader.start(fd, true) };
 
             #[cfg(unix)]
             {
-                if matches!(self.state, State::Err(_)) {
-                    // onReaderError already ran; `_keepalive`'s Drop on return
-                    // will drop the last ref and deinit() closes the handle.
-                    return;
-                }
-                if let Some(poll) = self.reader.handle.get_poll() {
-                    poll.set_flag(FilePollFlag::Socket);
-                    poll.set_flag(FilePollFlag::Nonblocking);
+                // SAFETY: `_keepalive` keeps `*this` live even if on_reader_error
+                // ran inside start(); borrows end at each `;`.
+                unsafe {
+                    if matches!((*this).state, State::Err(_)) {
+                        // on_reader_error already ran; `_keepalive`'s drop
+                        // releases the last ref and deinit() closes the handle.
+                        return;
+                    }
+                    if let Some(poll) = (*this).reader.handle.get_poll() {
+                        poll.set_flag(FilePollFlag::Socket);
+                        poll.set_flag(FilePollFlag::Nonblocking);
+                    }
+                    (*this).reader.flags.insert(
+                        PosixFlags::SOCKET | PosixFlags::NONBLOCKING | PosixFlags::POLLABLE,
+                    );
                 }
-                self.reader
-                    .flags
-                    .insert(PosixFlags::SOCKET | PosixFlags::NONBLOCKING | PosixFlags::POLLABLE);
             }
         }
     }
@@ -231,28 +273,68 @@ impl PipeReader {
         self.to_readable_stream(global_object)
     }
 
-    fn on_reader_done(&mut self) {
-        let owned = self.to_owned_slice();
-        self.state = State::Done(owned);
-        if let Some(process) = self.process.take() {
-            // `process` backref is valid while set; cleared before deref.
-            let kind = self.kind(process.get());
-            process.on_close_io(kind);
+    /// `BufferedReaderParent::on_reader_done`; see [`Self::finish`] for why it
+    /// takes the parent pointer the reader holds rather than `&mut self`.
+    ///
+    /// # Safety
+    /// See [`Self::finish`].
+    unsafe fn on_reader_done(this: *mut Self) {
+        // SAFETY: caller contract; the `&mut` lasts for this call only, and
+        // nothing it reaches touches `*this` through another pointer.
+        let owned = unsafe { (*this).to_owned_slice() };
+        // SAFETY: caller contract.
+        unsafe { Self::finish(this, State::Done(owned)) };
+    }
+
+    /// `BufferedReaderParent::on_reader_error`; also the teardown `start()`
+    /// runs when the pipe cannot be registered.
+    ///
+    /// # Safety
+    /// See [`Self::finish`].
+    unsafe fn on_reader_error(this: *mut Self, err: bun_sys::Error) {
+        // SAFETY: caller contract.
+        unsafe { Self::finish(this, State::Err(err)) };
+    }
+
+    /// Records the terminal `state`, tells the Subprocess this pipe is closed
+    /// (which drops the `Readable::Pipe` ref and takes the buffered output
+    /// through the Readable's own pointer into `*this`), then releases the ref
+    /// `start()` took. That release is normally the last one, so it has to be
+    /// made through the pointer the reader registered as its parent: a `&mut
+    /// self` receiver would still be live (and, as a function argument,
+    /// protected) while the allocation is freed, and `on_close_io`'s access
+    /// would alias it.
+    ///
+    /// # Safety
+    /// `this` must point to a live `PipeReader` whose `start()` ref is still
+    /// held, with no `&`/`&mut` to `*this` live across the call. `*this` may be
+    /// freed on return.
+    unsafe fn finish(this: *mut Self, state: State) {
+        // SAFETY: caller contract: the guard owns the `start()` ref and
+        // releases it when it drops, after the borrows below have ended.
+        let _start_ref = unsafe { ScopedRef::adopt(this) };
+        // A replaced `State::Done` buffer is freed by the assignment.
+        // SAFETY: caller contract; both borrows end inside the block.
+        let process = unsafe {
+            (*this).state = state;
+            (*this).process.take()
+        };
+        if let Some(process) = process {
+            process.on_close_io(Self::kind(this, process.get()));
         }
-        // SAFETY: last use of `self`; caller holds only a raw parent pointer,
-        // so freeing here does not invalidate any live `&mut`.
-        unsafe { PipeReader::deref(self) };
     }
 
-    fn kind(&self, process: &Subprocess<'_>) -> StdioKind {
+    /// Which of the Subprocess's slots holds this reader. Compares addresses
+    /// only, so it never forms a reference to `*this`.
+    fn kind(this: *const Self, process: &Subprocess<'_>) -> StdioKind {
         if let Readable::Pipe(pipe) = process.stdout.get() {
-            if core::ptr::eq(pipe.data.as_ptr(), self) {
+            if core::ptr::eq(pipe.data.as_ptr(), this) {
                 return StdioKind::Stdout;
             }
         }
 
         if let Readable::Pipe(pipe) = process.stderr.get() {
-            if core::ptr::eq(pipe.data.as_ptr(), self) {
+            if core::ptr::eq(pipe.data.as_ptr(), this) {
                 return StdioKind::Stderr;
             }
         }
@@ -347,18 +429,6 @@ impl PipeReader {
         }
     }
 
-    fn on_reader_error(&mut self, err: bun_sys::Error) {
-        // A previous `State::Done` buffer is freed by Drop of the replaced Vec.
-        self.state = State::Err(err);
-        if let Some(process) = self.process.take() {
-            // `process` backref is valid while set; cleared before deref.
-            let kind = self.kind(process.get());
-            process.on_close_io(kind);
-        }
-        // SAFETY: last use of `self`; see `on_reader_done`.
-        unsafe { PipeReader::deref(self) };
-    }
-
     pub(crate) fn close(&mut self) {
         match self.state {
             State::Pending => {
@@ -411,13 +481,14 @@ impl PipeReader {
 
 // BufferedReader vtable parent: `onReaderDone`/`onReaderError`/`loop`/
 // `eventLoop` (no `onReadChunk`).
-// `on_reader_done`/`on_reader_error` are tail-position (the reader is finished
-// with `self`), so `&mut *this` autoref is OK.
+// `on_reader_done`/`on_reader_error` forward the raw `*mut Self` rather than
+// autoref-ing it: they usually free `*this` (see `finish`), which must not
+// happen under a `&mut self` receiver.
 bun_io::impl_buffered_reader_parent! {
     SubprocessPipeReader for PipeReader;
     has_on_read_chunk = false;
-    on_reader_done  = |this| (*this).on_reader_done();
-    on_reader_error = |this, err| (*this).on_reader_error(err);
+    on_reader_done  = |this| PipeReader::on_reader_done(this);
+    on_reader_error = |this, err| PipeReader::on_reader_error(this, err);
     loop_           = |this| (*this).loop_().cast();
     event_loop      = |this| (*this).event_loop_handle.as_event_loop_ctx();
 }