io: POSIX pipe writers hold a ref on their parent from on_error until close() returns

src/io/PipeWriter.rs

@@ -271,6 +271,33 @@ fn write_to_socket(fd: Fd, buf: &[u8]) -> sys::Result<usize> {
     })
 }
 
+/// The refcount of the parent that a POSIX writer is a field of.
+pub trait PosixWriterParent {
+    /// `this` must point to a live `Self`.
+    unsafe fn ref_(this: *mut Self);
+    /// `this` must point to a live `Self` that the caller holds a ref on. May free it.
+    unsafe fn deref(this: *mut Self);
+}
+
+/// A ref on the parent of a writer. Its drop can free the parent and the writer in it.
+struct ParentKeepAlive<Parent: PosixWriterParent>(*mut Parent);
+
+impl<Parent: PosixWriterParent> ParentKeepAlive<Parent> {
+    /// `parent` must point to a live `Parent`.
+    unsafe fn new(parent: *mut Parent) -> Self {
+        // SAFETY: caller contract.
+        unsafe { Parent::ref_(parent) };
+        Self(parent)
+    }
+}
+
+impl<Parent: PosixWriterParent> Drop for ParentKeepAlive<Parent> {
+    fn drop(&mut self) {
+        // SAFETY: the ref taken in `new` kept the parent live until here.
+        unsafe { Parent::deref(self.0) };
+    }
+}
+
 // ──────────────────────────────────────────────────────────────────────────
 // PosixBufferedWriter
 // ──────────────────────────────────────────────────────────────────────────
@@ -283,14 +310,15 @@ fn write_to_socket(fd: Fd, buf: &[u8]) -> sys::Result<usize> {
 /// Materializing `&mut Parent` while a `&mut writer` is live would alias under
 /// Stacked Borrows, so we use raw
 /// pointers and never form a `&mut Parent` inside the writer.
-pub trait PosixBufferedWriterParent {
+pub trait PosixBufferedWriterParent: PosixWriterParent {
     /// `bun_io::poll_tag` constant for this writer's `FilePoll` owner. The
     /// per-tag dispatch in `bun_runtime::dispatch::__bun_run_file_poll`
     /// recovers `*mut PosixBufferedWriter<Self>` from this.
     const POLL_OWNER_TAG: PollTag;
     /// # Safety
     /// `this` must point to a live `Self`.
     unsafe fn on_write(this: *mut Self, amount: usize, status: WriteStatus);
+    /// May release every other ref on `this`: the writer holds one until its `close()` returns.
     /// # Safety
     /// `this` must point to a live `Self`.
     unsafe fn on_error(this: *mut Self, err: sys::Error);
@@ -433,6 +461,9 @@ impl<Parent: PosixBufferedWriterParent> PosixBufferedWriter<Parent> {
     fn _on_error(&mut self, err: sys::Error) {
         debug_assert!(!err.is_retry());
 
+        // `on_error` can release every other ref, and `close()` is still to run.
+        // SAFETY: type invariant — set-once parent backref outlives writer.
+        let _keep_alive = unsafe { ParentKeepAlive::new(self.parent()) };
         self.parent_on_error(err);
 
         self.close();
@@ -611,14 +642,15 @@ impl<Parent: PosixBufferedWriterParent> PosixBufferedWriter<Parent> {
 /// Materializing `&mut Parent` while a `&mut writer` is live would alias under
 /// Stacked Borrows, so we use raw
 /// pointers and never form a `&mut Parent` inside the writer.
-pub trait PosixStreamingWriterParent {
+pub trait PosixStreamingWriterParent: PosixWriterParent {
     /// `bun_io::poll_tag` constant for this writer's `FilePoll` owner. The
     /// per-tag dispatch in `bun_runtime::dispatch::__bun_run_file_poll`
     /// recovers `*mut PosixStreamingWriter<Self>` from this.
     const POLL_OWNER_TAG: PollTag;
     /// # Safety
     /// `this` must point to a live `Self`.
     unsafe fn on_write(this: *mut Self, amount: usize, status: WriteStatus);
+    /// May release every other ref on `this`: the writer holds one until its `close()` returns.
     /// # Safety
     /// `this` must point to a live `Self`.
     unsafe fn on_error(this: *mut Self, err: sys::Error);
@@ -785,8 +817,12 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
         self.is_done = true;
         self.outgoing.reset();
 
+        let parent = self.parent();
+        // `on_error` can release every other ref on the parent, and `close()` is still to run.
         // SAFETY: parent BACKREF set via set_parent; outlives this writer.
-        unsafe { Parent::on_error(self.parent(), err) };
+        let _keep_alive = unsafe { ParentKeepAlive::new(parent) };
+        // SAFETY: as above.
+        unsafe { Parent::on_error(parent, err) };
         self.close();
     }
 
@@ -837,16 +873,25 @@ impl<Parent: PosixStreamingWriterParent> PosixStreamingWriter<Parent> {
                 // ASM-verified PROVEN_CACHED on the `self.close()` path's
                 // field reads. Launder so `close()` sees fresh state.
                 let this: *mut Self = core::hint::black_box(core::ptr::from_mut(self));
-                // SAFETY: parent BACKREF valid.
-                unsafe { Parent::on_error(Self::r(this).parent(), err) };
-                // `this` is still live (parent owns this writer; an on_error
-                // handler may end/detach but never frees mid-call).
-                Self::r(this).close();
+                Self::on_register_poll_error(this, err);
             }
             sys::Result::Ok(()) => {}
         }
     }
 
+    /// Out of line so that `register_poll`, which runs on every buffered write, stays small.
+    #[cold]
+    #[inline(never)]
+    fn on_register_poll_error(this: *mut Self, err: sys::Error) {
+        let parent = Self::r(this).parent();
+        // As in `_on_error`: `close()` is still to run.
+        // SAFETY: parent BACKREF valid.
+        let _keep_alive = unsafe { ParentKeepAlive::new(parent) };
+        // SAFETY: parent BACKREF valid.
+        unsafe { Parent::on_error(parent, err) };
+        Self::r(this).close();
+    }
+
     pub fn write_utf16(&mut self, buf: &[u16]) -> WriteResult {
         if self.is_done || self.closed_without_reporting {
             return WriteResult::Done(0);
@@ -2657,7 +2702,7 @@ pub mod __parent_macro {
     pub use ::bun_uws_sys::Loop as UwsLoop;
 }
 
-/// Stamp `PosixStreamingWriterParent` + `WindowsWriterParent` +
+/// Stamp `PosixWriterParent` + `PosixStreamingWriterParent` + `WindowsWriterParent` +
 /// `WindowsStreamingWriterParent` for a parent type. See module comment above.
 #[macro_export]
 macro_rules! impl_streaming_writer_parent {
@@ -2666,7 +2711,7 @@ macro_rules! impl_streaming_writer_parent {
     (@call shared $p:expr; $m:ident($($a:tt)*)) => { (&*$p).$m($($a)*) };
     (@call ptr    $p:expr; $m:ident($($a:tt)*)) => { <Self>::$m($p, $($a)*) };
 
-    // Internal: expand the three impls once generics are normalized.
+    // Internal: expand the impls once generics are normalized.
     (@emit
         [$($gen:tt)*] $Ty:ty;
         poll_tag   = $poll_tag:expr,
@@ -2724,6 +2769,24 @@ macro_rules! impl_streaming_writer_parent {
             }
         }
 
+        #[cfg(unix)]
+        impl $($gen)* $crate::pipe_writer::PosixWriterParent for $Ty {
+            #[inline]
+            unsafe fn ref_(this: *mut Self) {
+                // SAFETY: BACKREF set via `set_parent`. Intrusive refcount bump.
+                let $ref_this = this;
+                #[allow(unused_unsafe)]
+                unsafe { $ref_ };
+            }
+            #[inline]
+            unsafe fn deref(this: *mut Self) {
+                // SAFETY: see ref_. May free `this`.
+                let $deref_this = this;
+                #[allow(unused_unsafe)]
+                unsafe { $deref };
+            }
+        }
+
         #[cfg(windows)]
         impl $($gen)* $crate::pipe_writer::WindowsWriterParent for $Ty {
             #[inline]
@@ -2795,7 +2858,7 @@ macro_rules! impl_streaming_writer_parent {
     };
 }
 
-/// Stamp `PosixBufferedWriterParent` + `WindowsWriterParent` +
+/// Stamp `PosixWriterParent` + `PosixBufferedWriterParent` + `WindowsWriterParent` +
 /// `WindowsBufferedWriterParent` for a parent type. See module comment above.
 #[macro_export]
 macro_rules! impl_buffered_writer_parent {
@@ -2852,6 +2915,24 @@ macro_rules! impl_buffered_writer_parent {
             }
         }
 
+        #[cfg(not(windows))]
+        impl $($gen)* $crate::pipe_writer::PosixWriterParent for $Ty {
+            #[inline]
+            unsafe fn ref_(this: *mut Self) {
+                // SAFETY: BACKREF set via `set_parent`. Intrusive refcount bump.
+                let $ref_this = this;
+                #[allow(unused_unsafe)]
+                unsafe { $ref_ };
+            }
+            #[inline]
+            unsafe fn deref(this: *mut Self) {
+                // SAFETY: see ref_. May free `this`.
+                let $deref_this = this;
+                #[allow(unused_unsafe)]
+                unsafe { $deref };
+            }
+        }
+
         #[cfg(windows)]
         impl $($gen)* $crate::pipe_writer::WindowsWriterParent for $Ty {
             #[inline]

src/runtime/api/bun/Terminal.rs

@@ -2077,6 +2077,19 @@ impl bun_io::pipe_writer::PosixStreamingWriterParent for Terminal {
     }
 }
 
+#[cfg(unix)]
+impl bun_io::pipe_writer::PosixWriterParent for Terminal {
+    unsafe fn ref_(this: *mut Self) {
+        // SAFETY: BACKREF set via writer.parent. Through the raw pointer, as the
+        // Windows `WindowsWriterParent::ref_` below: a `&mut self.writer` is live.
+        unsafe { bun_ptr::RefCount::<Terminal>::ref_(this) };
+    }
+    unsafe fn deref(this: *mut Self) {
+        // SAFETY: see ref_. May free `this`.
+        unsafe { bun_ptr::RefCount::<Terminal>::deref(this) };
+    }
+}
+
 #[cfg(windows)]
 impl bun_io::pipe_writer::WindowsWriterParent for Terminal {
     unsafe fn loop_(this: *mut Self) -> *mut bun_libuv_sys::Loop {

src/runtime/shell/IOWriter.rs

@@ -1088,8 +1088,8 @@ bun_io::impl_buffered_writer_parent! {
     uv_loop    = |this| (*(*this).evtloop().loop_()).uv_loop,
     // INVARIANT: `this` is `Arc::as_ptr` stashed via `writer.set_parent` in
     // `IOWriter::init` (sole constructor); passing a non-Arc ptr is UB.
-    ref_       = |this| std::sync::Arc::increment_strong_count(this as *const Self),
-    deref      = |this| std::sync::Arc::decrement_strong_count(this as *const Self),
+    ref_       = |this| std::sync::Arc::increment_strong_count(this.cast_const()),
+    deref      = |this| std::sync::Arc::decrement_strong_count(this.cast_const()),
 }
 
 // ──────────────────────────────────────────────────────────────────────────

src/runtime/webcore/FileSink.rs

@@ -464,8 +464,7 @@ impl FileSink {
         bun_core::scoped_log!(FileSink, "onError({:?})", err);
         // The streaming writer follows every `onError` with `close()` →
         // `onClose` (on both platforms), which fires `source.close()` and
-        // releases the keep-alive ref. Releasing the ref here instead could
-        // drop the last reference and free `this` before that `close()` runs.
+        // releases the keep-alive ref. `run_pending` can release every other ref on `this`.
         // SAFETY: caller contract — `this` is live with write+dealloc provenance.
         unsafe {
             (*this).record_stream_error(streams::StreamError::Error(err.clone()));

src/spawn/static_pipe_writer.rs

@@ -280,9 +280,8 @@ impl<P: StaticPipeWriterProcess> StaticPipeWriter<P> {
         // for the lifetime of this writer (the process owns/outlives its stdio writers).
         unsafe { P::on_close_io(self.process, StdioKind::Stdin) };
         if release_start_ref {
-            // SAFETY: token taken above. On POSIX this frees `self`: it is the
-            // last use here, and the writer's `close()` frames below do nothing
-            // after this callback. On Windows the in-flight write's ref outlives it.
+            // SAFETY: token taken above. Not the last ref: on POSIX the writer's `_on_error`
+            // holds one until its `close()` returns, on Windows the in-flight write does.
             unsafe { RefCount::<Self>::deref(std::ptr::from_mut::<Self>(self)) };
         }
     }

test/js/bun/util/filesink.test.ts

@@ -1120,6 +1120,18 @@ describe("a stream piped into a FileSink on a pipe is pumped to its end", () =>
         c.close();
       },
     }`,
+    // pull() waits for the sink to write the whole chunk, so the sink holds a promise that
+    // it has to reject when the reader goes away.
+    directFlush: `{
+      type: "direct",
+      async pull(c) {
+        c.write(first);
+        const flushed = c.flush();
+        console.error("parked");
+        await flushed;
+        c.close();
+      },
+    }`,
   };
 
   function fixture(source: string, dest: string, countBeforeExit = false) {
@@ -1158,8 +1170,9 @@ describe("a stream piped into a FileSink on a pipe is pumped to its end", () =>
   }
 
   // read(2) in a poll on the non-blocking read end the test holds: it returns 0 once the
-  // writer is gone, on every POSIX, and needs no readiness notification.
-  async function drain(fd: number) {
+  // writer is gone, on every POSIX, and needs no readiness notification. With a `limit` it
+  // stops once that many bytes have arrived.
+  async function drain(fd: number, limit = Infinity) {
     const buffer = Buffer.alloc(64 * 1024);
     const parts: Buffer[] = [];
     let total = 0;
@@ -1180,6 +1193,7 @@ describe("a stream piped into a FileSink on a pipe is pumped to its end", () =>
       parts.push(Buffer.from(buffer.subarray(0, n)));
       total += n;
       progressAt = performance.now();
+      if (total >= limit) return Buffer.concat(parts);
     }
   }
 
@@ -1330,6 +1344,56 @@ describe("a stream piped into a FileSink on a pipe is pumped to its end", () =>
       fs.closeSync(readFd);
     }
   });
+
+  // The reader takes part of the stream and goes away while pull() awaits flush(). The child
+  // finds out in its poll of stdout, and the sink rejects the flush from that callback.
+  // pull() then fails, which closes the sink and drops the last refs on it, and the pipe
+  // writer, a field of the sink, still had a close of its own to make (ASAN:
+  // heap-use-after-free in PosixStreamingWriter::close).
+  //
+  // The flush that ends each turn of the child's loop writes too, and a failure there is
+  // reported from another place. So the read end has to close while the child waits in its
+  // poll, and no signal says when it does. The reader takes a few refills, which puts the
+  // child past the turn that printed "parked", and then gives it a moment to handle the last.
+  it.concurrent.skipIf(!isPosix)(
+    "a reader that goes away while pull() awaits flush() rejects the promise",
+    async () => {
+      using dir = tempDir("filesink-piped-reader-gone", {});
+      const path = join(String(dir), "stdout.fifo");
+      mkfifo(path, 0o666);
+      const readFd = fs.openSync(path, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK);
+      let readFdOpen = true;
+      let writeFd: number | undefined;
+      try {
+        writeFd = fs.openSync(path, fs.constants.O_WRONLY);
+        await using proc = Bun.spawn({
+          cmd: [bunExe(), "-e", fixture(sources.directFlush, "Bun.stdout")],
+          env: bunEnv,
+          stdout: writeFd,
+          stderr: "pipe",
+        });
+        // The child holds the only write end now: a read of 0 bytes means it is gone.
+        fs.closeSync(writeFd);
+        writeFd = undefined;
+
+        const rest = await parked(proc.stderr);
+        const received = await drain(readFd, 4 * 64 * 1024);
+        await Bun.sleep(10);
+        fs.closeSync(readFd);
+        readFdOpen = false;
+
+        const [stderr, exitCode] = await Promise.all([rest(), proc.exited]);
+        expect({ intact: expected.subarray(0, received.length).equals(received), stderr, exitCode }).toEqual({
+          intact: true,
+          stderr: "parked\n" + JSON.stringify({ settled: "rejected EPIPE: broken pipe, write", code: 0 }) + "\n",
+          exitCode: 0,
+        });
+      } finally {
+        if (writeFd !== undefined) fs.closeSync(writeFd);
+        if (readFdOpen) fs.closeSync(readFd);
+      }
+    },
+  );
 });
 
 // FileSink::on_close tells the owner of the sink that it closed, and a Subprocess then drops its ref