spawn(windows): free the buffer-stdin writer when its uv_write fails inside start()

src/bun_core/env_var.rs

@@ -273,6 +273,9 @@ pub mod feature_flag {
     new_feature_flag!(pub BUN_INTERNAL_BUNX_INSTALL, "BUN_INTERNAL_BUNX_INSTALL", {});
     // Debug-only fault injection for test/js/bun/spawn/spawn-pipe-start-error.test.ts.
     new_feature_flag!(pub BUN_INTERNAL_FAIL_PIPE_READER_START, "BUN_INTERNAL_FAIL_PIPE_READER_START", {});
+    // Debug-only fault injection for test/js/bun/spawn/spawn-pipe-start-error.test.ts:
+    // every stream write the Windows buffered pipe writer issues fails synchronously.
+    new_feature_flag!(pub BUN_INTERNAL_FAIL_PIPE_WRITER_WRITE, "BUN_INTERNAL_FAIL_PIPE_WRITER_WRITE", {});
     // Test-only: bypass the stdin isatty gate in `bun update --interactive` so
     // tests can drive the multi-select by writing keystrokes to a pipe.
     new_feature_flag!(pub BUN_INTERNAL_INTERACTIVE_ASSUME_TTY, "BUN_INTERNAL_INTERACTIVE_ASSUME_TTY", {});

src/install/PackageManager/security_scanner.rs

@@ -944,7 +944,7 @@ pub(crate) type StaticPipeWriter = subprocess::StaticPipeWriter<SecurityScanSubp
 // Wire the writer's `on_close` callback back to this type. Raw `*mut Self`
 // because the call is re-entrant: it may fire synchronously inside
 // `StaticPipeWriter::start()` while `finish_spawn` still has `&mut self` on
-// the stack (small JSON fits the pipe buffer → write completes → close).
+// the stack (Windows: the write fails synchronously and the writer closes).
 impl<'a> subprocess::StaticPipeWriterProcess for SecurityScanSubprocess<'a> {
     const POLL_OWNER_TAG: bun_io::PollTag = bun_io::PollTag::SecurityScanStaticPipeWriter;
     unsafe fn on_close_io(this: *mut Self, kind: subprocess::StdioKind) {
@@ -1288,8 +1288,8 @@ impl<'a> SecurityScanSubprocess<'a> {
             (*parent).process = Some(process);
         }
 
-        // Assign the field BEFORE `start()`. `start()` may complete the write synchronously
-        // (small JSON fits the 64KB pipe buffer on POSIX) and re-enter
+        // Assign the field BEFORE `start()`. On Windows a write that fails
+        // synchronously closes the writer inside `start()`, re-entering
         // `on_close_io` via the `parent` backref; that callback must observe
         // `json_writer.is_some()` to decrement `remaining_fds`, otherwise
         // `is_done()` never returns true and `sleep_until` hangs.
@@ -1316,13 +1316,21 @@ impl<'a> SecurityScanSubprocess<'a> {
         });
 
         let writer_ptr = writer_local.as_ptr();
-        // SAFETY: `writer_local` holds a live ref; `start()` mutates the writer
-        // in place (raw intrusive object — no Rust aliasing across the RefPtr).
-        let start_result = unsafe { (*writer_ptr).start() };
-        // SAFETY: `writer_local` keeps `*writer_ptr` live; we own the `start()` ref.
-        unsafe { RefCount::<StaticPipeWriter>::deref(writer_ptr) };
+        // SAFETY: `writer_local` holds a ref, so the writer stays live across
+        // and after `start()` whatever it does (it may re-enter `on_close_io`
+        // and release its own ref; see `StaticPipeWriter::start`).
+        let start_result = unsafe { StaticPipeWriter::start(writer_ptr) };
+        // Claim start()'s ref through the `started` token, the same token the
+        // writer's own release sites check: `writer_local` covers this frame and
+        // `json_writer` covers the write, so this owner has no use for it.
+        // `start()` leaves the token unset when it already released the ref
+        // itself (it failed, or the write failed synchronously).
         // SAFETY: `writer_local` keeps `*writer_ptr` live.
-        unsafe { (*writer_ptr).started = false };
+        if unsafe { core::mem::replace(&mut (*writer_ptr).started, false) } {
+            // SAFETY: `started` was the token for the outstanding start() ref;
+            // cleared above so no other site releases it. Not the last ref.
+            unsafe { RefCount::<StaticPipeWriter>::deref(writer_ptr) };
+        }
         match start_result {
             Err(e) => {
                 writer_local.deref();

src/io/PipeWriter.rs

@@ -1646,6 +1646,18 @@ impl<Parent: WindowsBufferedWriterParent> WindowsBufferedWriter<Parent> {
         Self::r(this).on_write_complete(uv::ReturnCode::zero());
     }
 
+    /// Debug-only fault injection (`BUN_INTERNAL_FAIL_PIPE_WRITER_WRITE`) for
+    /// test/js/bun/spawn/spawn-pipe-start-error.test.ts: stands in for a
+    /// `uv_write` that fails synchronously, which JS cannot arrange.
+    fn injected_write_error() -> Option<sys::Error> {
+        #[cfg(debug_assertions)]
+        if bun_core::env_var::feature_flag::BUN_INTERNAL_FAIL_PIPE_WRITER_WRITE.get() == Some(true)
+        {
+            return Some(sys::Error::from_code(sys::E::PIPE, sys::Tag::write));
+        }
+        None
+    }
+
     pub fn write(&mut self) {
         let buffer = self.get_buffer_internal();
         // if we are already done or if we have some pending payload we just wait until next write
@@ -1716,15 +1728,18 @@ impl<Parent: WindowsBufferedWriterParent> WindowsBufferedWriter<Parent> {
             self.pending_payload_size = buffer_len;
             self.write_buffer = write_buf;
             let self_ptr = self as *mut Self;
-            if let Some(write_err) = self
-                .write_req
-                // SAFETY: `p` is `self_ptr`; libuv invokes on the loop thread with no
-                // other Rust borrow of `*p` live, so `&mut *p` is the sole alias.
-                .write(stream_raw, &self.write_buffer, self_ptr, |p, s| unsafe {
-                    (*p).on_write_complete(s)
-                })
-                .to_error(sys::Tag::write)
-            {
+            let write_err = if let Some(err) = Self::injected_write_error() {
+                Some(err)
+            } else {
+                self.write_req
+                    // SAFETY: `p` is `self_ptr`; libuv invokes on the loop thread with no
+                    // other Rust borrow of `*p` live, so `&mut *p` is the sole alias.
+                    .write(stream_raw, &self.write_buffer, self_ptr, |p, s| unsafe {
+                        (*p).on_write_complete(s)
+                    })
+                    .to_error(sys::Tag::write)
+            };
+            if let Some(write_err) = write_err {
                 self.close();
                 self.parent_on_error(write_err);
             } else {

src/runtime/api/bun/js_bun_spawn_bindings.rs

@@ -1765,7 +1765,11 @@ fn spawn_maybe_sync<const IS_SYNC: bool>(
     }
 
     if let Writable::Buffer(buffer) = subprocess.stdin.get() {
-        if let Err(err) = Writable::buffer_writer_mut(buffer).start() {
+        // SAFETY: `buffer` holds `create()`'s ref on a live writer. `start()`
+        // may free the writer (a write that fails synchronously on Windows),
+        // in which case `on_close_io` has already replaced this stdin slot;
+        // neither `buffer` nor the writer is used after the call.
+        if let Err(err) = unsafe { Subprocess::StaticPipeWriter::start(buffer.as_ptr()) } {
             let _ = subprocess.try_kill(subprocess.kill_signal);
             let _ = global_this.throw_value(err.to_js(global_this));
             return Err(JsError::Thrown);

src/runtime/shell/subproc.rs

@@ -905,12 +905,18 @@ impl ShellSubprocess {
             }
         }
 
-        // SAFETY: borrow of the stdin slot scoped to this match; single-threaded.
-        let stdin_start_err = match unsafe { &(*subprocess).stdin } {
-            // SAFETY: single-threaded; the writer is uniquely reachable here.
-            Writable::Buffer(buffer) => unsafe { buffer_mut(buffer) }.start().err(),
+        // SAFETY: borrow of the stdin slot scoped to this match; it ends before
+        // `start()` below, which may overwrite the slot through `on_close_io`.
+        let stdin_writer = match unsafe { &(*subprocess).stdin } {
+            Writable::Buffer(buffer) => Some(buffer.as_ptr()),
             _ => None,
         };
+        let stdin_start_err = stdin_writer.and_then(|writer| {
+            // SAFETY: the slot holds `create()`'s ref on a live writer. `start()`
+            // may free it (a write that fails synchronously on Windows, reported
+            // through `on_close_io`); nothing touches it afterwards.
+            unsafe { StaticPipeWriter::start(writer) }.err()
+        });
         if let Some(err) = stdin_start_err {
             let sys_err = err.to_shell_system_error();
             // SAFETY: scoped `&mut` for the kill; `abort_after_failed_start`

src/spawn/static_pipe_writer.rs

@@ -10,6 +10,8 @@ use bun_sys;
 use crate::process::StdioKind;
 use crate::subprocess::{Source, StdioResult};
 
+// `BUN_DEBUG_StaticPipeWriter=1`; test/js/bun/spawn/spawn-pipe-start-error.test.ts
+// counts the `create()` / `deinit()` lines to check writers are freed.
 bun_output::declare_scope!(StaticPipeWriter, hidden);
 
 /// Trait bound for the owning process type `P` of [`StaticPipeWriter`].
@@ -138,58 +140,92 @@ impl<P: StaticPipeWriterProcess> StaticPipeWriter<P> {
             }
         }
         let this = bun_core::heap::into_raw(boxed);
+        bun_output::scoped_log!(
+            StaticPipeWriter,
+            "StaticPipeWriter(0x{:x}) create()",
+            this as usize
+        );
         // SAFETY: `this` was just leaked above; borrow scoped to registering
         // the parent backref.
         unsafe { (*this).writer.set_parent(this) };
         // SAFETY: ownership of the initial ref is transferred to the returned IntrusiveRc.
         unsafe { IntrusiveRc::from_raw(this) }
     }
 
-    pub fn start(&mut self) -> bun_sys::Result<()> {
+    /// Takes start()'s `+1` (tracked by `started`) and begins writing.
+    ///
+    /// Raw `this` rather than `&mut self`: on Windows a `uv_write` that fails
+    /// synchronously closes the writer inside this call. `on_close` then runs
+    /// while `started` is still unset, so it releases nothing, and the owner's
+    /// `on_close_io` empties its slot and drops `create()`'s ref. Nothing can
+    /// claim start()'s `+1` after that, so this function releases it itself,
+    /// which frees `*this`.
+    ///
+    /// # Safety
+    /// `this` must point to a live writer whose owner holds `create()`'s ref.
+    /// `*this` may be freed when this returns (see above), so the caller must
+    /// not touch it afterwards; the owner is notified through `on_close_io`.
+    pub unsafe fn start(this: *mut Self) -> bun_sys::Result<()> {
         bun_output::scoped_log!(
             StaticPipeWriter,
             "StaticPipeWriter(0x{:x}) start()",
-            std::ptr::from_ref(self) as usize
+            this as usize
         );
-        // Intrusive-refcount increment.
-        // SAFETY: `self` is a live `Self` (created via `create()`/`heap::alloc`).
-        unsafe { RefCount::<Self>::ref_(std::ptr::from_mut::<Self>(self)) };
-        // Self-borrow into `self.source` — see `buffer` field invariant.
-        self.buffer = RawSlice::new(self.source.slice());
+        // SAFETY: caller contract: `this` is live.
+        unsafe { RefCount::<Self>::ref_(this) };
+        // Self-borrow into `source`; see the `buffer` field invariant.
+        // SAFETY: live; the borrow is confined to this statement.
+        unsafe { (*this).buffer = RawSlice::new((*this).source.slice()) };
         #[cfg(windows)]
         {
-            let r = self.writer.start_with_current_pipe();
-            self.started = r.is_ok();
-            if r.is_err() {
-                // start() failed: `started` stays false so no release site
-                // fires — release start()'s `+1` here.
-                // SAFETY: `self` is the live `Self` we ref'd at the top of
-                // `start()`; the caller's `IntrusiveRc` keeps it alive and
-                // `started` is false so no other site re-derefs.
-                unsafe { RefCount::<Self>::deref(std::ptr::from_mut::<Self>(self)) };
+            // SAFETY: the ref taken above keeps `*this` live across the call
+            // even when it closes the writer (see the doc comment); the borrow
+            // of the field ends when the call returns.
+            let result = unsafe { (*this).writer.start_with_current_pipe() };
+            // SAFETY: still live, see above.
+            if result.is_err() || unsafe { (*this).writer.is_done() } {
+                // `Err`: nothing was closed; the owner still holds `create()`'s
+                // ref and tears the writer down when the caller reports the
+                // error. Closed: the write failed synchronously and has been
+                // reported through `on_error`/`on_close` just like a write
+                // that fails asynchronously, so the caller sees `Ok`; the
+                // owner's ref is already gone and this release frees the
+                // writer. Either way `started` stays false, so no other site
+                // releases start()'s `+1`.
+                // SAFETY: releases the ref taken above; last use of `this`.
+                unsafe { RefCount::<Self>::deref(this) };
+                return result;
             }
-            return r;
+            // SAFETY: live: the writer is open, so the owner still holds its ref.
+            unsafe { (*this).started = true };
+            bun_sys::Result::Ok(())
         }
         #[cfg(not(windows))]
         {
-            // On POSIX `StdioResult` is an `Option<Fd>`.
-            match self.writer.start(self.stdio_result.unwrap(), true) {
+            // On POSIX `StdioResult` is an `Option<Fd>`. The buffered writer's
+            // `start()` only registers the poll and never reports to the
+            // parent, so nothing in this arm can close or free the writer.
+            // SAFETY: live; the borrow ends before the match body runs.
+            let fd = unsafe { (*this).stdio_result.unwrap() };
+            // SAFETY: live; the borrow of the field ends when the call returns.
+            match unsafe { (*this).writer.start(fd, true) } {
                 bun_sys::Result::Err(err) => {
-                    // start() failed: `started` stays false so no release
-                    // site fires — release start()'s `+1` here.
-                    // SAFETY: `self` is the live `Self` we ref'd at the top
-                    // of `start()`; the caller's `IntrusiveRc` keeps it alive
-                    // and `started` is false so no other site re-derefs.
-                    unsafe { RefCount::<Self>::deref(std::ptr::from_mut::<Self>(self)) };
+                    // `started` stays false so no release site fires; release
+                    // start()'s `+1` here. Not the last ref: the owner still
+                    // holds `create()`'s.
+                    // SAFETY: releases the ref taken above; last use of `this`.
+                    unsafe { RefCount::<Self>::deref(this) };
                     bun_sys::Result::Err(err)
                 }
                 bun_sys::Result::Ok(()) => {
-                    self.started = true;
+                    // SAFETY: live, see above.
+                    unsafe { (*this).started = true };
                     #[cfg(unix)]
                     {
                         // `handle` is `PollOrFd` (enum); flag mutation goes
                         // through the FilePoll vtable shim.
-                        if let Some(poll) = self.writer.handle.get_poll() {
+                        // SAFETY: live, see above.
+                        if let Some(poll) = unsafe { (*this).writer.handle.get_poll() } {
                             poll.set_flag(bun_io::FilePollFlag::Socket);
                         }
                     }
@@ -261,9 +297,11 @@ impl<P: StaticPipeWriterProcess> StaticPipeWriter<P> {
         // reaches here via `close()` without ever calling `Parent::on_write`, so
         // this is the last point `started` can be claimed for that path.
         // `write()`'s +1 (held by that callback's scopeguard) keeps `self` live
-        // past the deref. POSIX must not release here: `drain_buffered_data`
-        // may call `on_error()` -> `close()` -> here and then `on_write()` on
-        // the same object, with no extra ref held.
+        // past the deref. A write that fails synchronously inside `start()`
+        // also passes through here, before `started` is set; `start()` releases
+        // its own ref for that case. POSIX must not release here:
+        // `drain_buffered_data` may call `on_error()` -> `close()` -> here and
+        // then `on_write()` on the same object, with no extra ref held.
         #[cfg(windows)]
         let release_start_ref = core::mem::replace(&mut self.started, false);
         // `buffer` aliases `self.source`'s storage; clear it before detach()
@@ -296,6 +334,11 @@ impl<P: StaticPipeWriterProcess> StaticPipeWriter<P> {
 /// The heap free is handled by `IntrusiveRc` after `drop` returns.
 impl<P: StaticPipeWriterProcess> Drop for StaticPipeWriter<P> {
     fn drop(&mut self) {
+        bun_output::scoped_log!(
+            StaticPipeWriter,
+            "StaticPipeWriter(0x{:x}) deinit()",
+            std::ptr::from_ref(self) as usize
+        );
         self.writer.end();
         // `buffer` aliases `self.source`'s storage; clear it before detach()
         // frees that storage (upholds the field's documented invariant).