diff --git a/src/io/PipeWriter.rs b/src/io/PipeWriter.rs index 4b949b9ddcfd..b18de4eb6731 100644 --- a/src/io/PipeWriter.rs +++ b/src/io/PipeWriter.rs @@ -271,6 +271,33 @@ fn write_to_socket(fd: Fd, buf: &[u8]) -> sys::Result { }) } +/// 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(*mut Parent); + +impl ParentKeepAlive { + /// `parent` must point to a live `Parent`. + unsafe fn new(parent: *mut Parent) -> Self { + // SAFETY: caller contract. + unsafe { Parent::ref_(parent) }; + Self(parent) + } +} + +impl Drop for ParentKeepAlive { + fn drop(&mut self) { + // SAFETY: the ref taken in `new` kept the parent live until here. + unsafe { Parent::deref(self.0) }; + } +} + // ────────────────────────────────────────────────────────────────────────── // PosixBufferedWriter // ────────────────────────────────────────────────────────────────────────── @@ -283,7 +310,7 @@ fn write_to_socket(fd: Fd, buf: &[u8]) -> sys::Result { /// 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` from this. @@ -291,6 +318,7 @@ pub trait PosixBufferedWriterParent { /// # 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 PosixBufferedWriter { 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,7 +642,7 @@ impl PosixBufferedWriter { /// 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` from this. @@ -619,6 +650,7 @@ pub trait PosixStreamingWriterParent { /// # 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 PosixStreamingWriter { 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 PosixStreamingWriter { // 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)*)) => { ::$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] diff --git a/src/runtime/api/bun/Terminal.rs b/src/runtime/api/bun/Terminal.rs index a41db046700e..b680000c514e 100644 --- a/src/runtime/api/bun/Terminal.rs +++ b/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::::ref_(this) }; + } + unsafe fn deref(this: *mut Self) { + // SAFETY: see ref_. May free `this`. + unsafe { bun_ptr::RefCount::::deref(this) }; + } +} + #[cfg(windows)] impl bun_io::pipe_writer::WindowsWriterParent for Terminal { unsafe fn loop_(this: *mut Self) -> *mut bun_libuv_sys::Loop { diff --git a/src/runtime/shell/IOWriter.rs b/src/runtime/shell/IOWriter.rs index 199f617f2afb..9dd37ae29c35 100644 --- a/src/runtime/shell/IOWriter.rs +++ b/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()), } // ────────────────────────────────────────────────────────────────────────── diff --git a/src/runtime/webcore/FileSink.rs b/src/runtime/webcore/FileSink.rs index cd1d3348604b..5abdedef43ac 100644 --- a/src/runtime/webcore/FileSink.rs +++ b/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())); diff --git a/src/spawn/static_pipe_writer.rs b/src/spawn/static_pipe_writer.rs index 68db15028e23..91fd8690018a 100644 --- a/src/spawn/static_pipe_writer.rs +++ b/src/spawn/static_pipe_writer.rs @@ -280,9 +280,8 @@ impl StaticPipeWriter

{ // 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::::deref(std::ptr::from_mut::(self)) }; } } diff --git a/test/js/bun/util/filesink.test.ts b/test/js/bun/util/filesink.test.ts index 89648f414baa..618936479e4f 100644 --- a/test/js/bun/util/filesink.test.ts +++ b/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