diff --git a/src/runtime/api/bun/js_bun_spawn_bindings.rs b/src/runtime/api/bun/js_bun_spawn_bindings.rs index 5397137e80f4..940f0381d2d5 100644 --- a/src/runtime/api/bun/js_bun_spawn_bindings.rs +++ b/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,27 @@ fn spawn_maybe_sync( // 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. + // Either call may free the reader (a failed start, EOF inside read_all), + // which also clears the slot, so the slot is re-read rather than `pipe` reused. + let lazy_reads = !IS_SYNC && lazy; if let Readable::Pipe(pipe) = subprocess.stdout.get() { - // Note: pass `subprocess_nn` (the `NonNull>` - // 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 the reader and nothing borrows it here. + unsafe { PipeReader::start(pipe.as_ptr(), subprocess_nn, event_loop_nn, lazy_reads) }; + if !lazy_reads { if let Readable::Pipe(pipe) = subprocess.stdout.get() { - Readable::pipe_reader_mut(pipe).read_all(); + // SAFETY: as above; the slot still reading `Pipe` means start() succeeded. + 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, lazy_reads) }; + if !lazy_reads { 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()) }; } } } diff --git a/src/runtime/api/bun/subprocess/SubprocessPipeReader.rs b/src/runtime/api/bun/subprocess/SubprocessPipeReader.rs index a5863b92fe43..f698093f5457 100644 --- a/src/runtime/api/bun/subprocess/SubprocessPipeReader.rs +++ b/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,47 +136,64 @@ 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) }; + /// EOF or an error inside the read reaches [`Self::finish`], so `*this` + /// may be freed on return. + /// + /// # Safety + /// `this` is a live, started `PipeReader` with no `&`/`&mut` to it live. + pub(crate) unsafe fn read_all(this: *mut Self) { + // SAFETY: caller contract; no borrow of `*this` is held across `read`. + unsafe { + if matches!((*this).state, State::Pending) { + IOReader::read(&raw mut (*this).reader); + } } } - pub(crate) fn start( - &mut self, + /// Takes the ref [`Self::finish`] releases. A pipe that cannot be + /// registered is finished synchronously, so `*this` may be freed on return. + /// + /// # Safety + /// `this` is a live `PipeReader` from `create()` with no `&`/`&mut` to it live. + pub(crate) unsafe fn start( + this: *mut Self, process: NonNull>, event_loop: NonNull, 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::(); - if let Some(source) = self.reader.source.as_mut() { - source.set_data(reader_ptr); + // SAFETY: caller contract. + unsafe { + let reader = &raw mut (*this).reader; + if let Some(source) = (*reader).source.as_mut() { + source.set_data(reader.cast::()); + } + (*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. - // - // 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::(self)) }; - if let bun_sys::Result::Err(err) = self.reader.start_with_current_pipe() { + // A failed start releases every other ref below; the guard's drop + // on return is then the last one. + // SAFETY: caller contract. + let _keepalive = unsafe { ScopedRef::new(this) }; + // SAFETY: caller contract; the `reader` borrow ends with the call. + 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 @@ -184,7 +201,9 @@ impl PipeReader { // 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; no borrow of it is live. + unsafe { Self::on_reader_error(this, err) }; } } @@ -193,35 +212,39 @@ 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. - // - // 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::(self)) }; - - let _ = self.reader.start(self.stdio_result.unwrap(), true); + // PosixBufferedReader::start() always returns Ok(()); a failed poll + // registration dispatches on_reader_error synchronously instead, + // releasing every other ref. The guard keeps `*this` alive for the + // state check below; its drop on return is then the last release. + // SAFETY: caller contract. + let _keepalive = unsafe { ScopedRef::new(this) }; + + // SAFETY: caller contract. + let fd = unsafe { (*this).stdio_result.unwrap() }; + // SAFETY: caller contract; only `reader` is borrowed, for the call. + 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; + // SAFETY: `_keepalive` keeps `*this` live. + 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, + ); } - if let Some(poll) = self.reader.handle.get_poll() { - poll.set_flag(FilePollFlag::Socket); - poll.set_flag(FilePollFlag::Nonblocking); - } - self.reader - .flags - .insert(PosixFlags::SOCKET | PosixFlags::NONBLOCKING | PosixFlags::POLLABLE); } } } @@ -231,28 +254,54 @@ 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); + /// # Safety + /// See [`Self::finish`]. + unsafe fn on_reader_done(this: *mut Self) { + // SAFETY: caller contract; the `&mut` lasts for this call only. + let owned = unsafe { (*this).to_owned_slice() }; + // SAFETY: caller contract. + unsafe { Self::finish(this, State::Done(owned)) }; + } + + /// # 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`, has the Subprocess drop its `Readable::Pipe` + /// ref, then releases the `start()` ref, which is normally the last one. + /// Raw `this`, not `&mut self`: `on_close_io` reaches back into `*this` + /// through the Readable's pointer, and the release frees `*this`; neither + /// may happen while a receiver borrow of it is live. + /// + /// # Safety + /// `this` is a live `PipeReader` still holding its `start()` ref, with no + /// `&`/`&mut` to it live. `*this` may be freed on return. + unsafe fn finish(this: *mut Self, state: State) { + // SAFETY: caller contract; the guard releases after the borrows below end. + let _start_ref = unsafe { ScopedRef::adopt(this) }; + // SAFETY: caller contract. + 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 { + /// Address comparison only; 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 +396,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 +448,13 @@ 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` usually free `*this` (see `finish`), so +// they get the raw pointer rather than a `&mut *this` autoref. 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(); } diff --git a/test/internal/source-lints/self-receiver-deref.test.ts b/test/internal/source-lints/self-receiver-deref.test.ts new file mode 100644 index 000000000000..8bb1ff904260 --- /dev/null +++ b/test/internal/source-lints/self-receiver-deref.test.ts @@ -0,0 +1,192 @@ +import { file } from "bun"; +import { expect, test } from "bun:test"; +import { realpathSync } from "fs"; +import path from "path"; +import { globAllSources } from "../../../scripts/glob-sources.ts"; + +// An intrusive-refcount release handed the method's own receiver, +// +// unsafe { PipeReader::deref(self) }; // `&mut self` coerces to `*mut Self` +// unsafe { Self::deref_with_context(self, ctx) }; +// let _ref = unsafe { ScopedRef::adopt(&mut *self) }; +// +// is banned. A release may be the object's last one, and then the destructor +// frees the allocation the receiver still points at. The receiver is a +// function argument, so it is protected for the whole call, and both aliasing +// models reject freeing protected memory even when `self` is never touched +// again: Tree Borrows (what `bun run rust:miri` uses) reports "deallocation +// through is forbidden ... the strongly protected tag disallows +// deallocations", pointing at the `&mut self`, and Stacked Borrows reports +// "deallocating while item [Unique] is strongly protected". The subprocess +// `PipeReader::on_reader_done`/`on_reader_error` were the canonical instance: +// every `Bun.spawn` stdout/stderr pipe ended by releasing the reader's last +// ref from inside its own `&mut self` (the Subprocess had dropped the other +// ref a line earlier, in `on_close_io`). +// +// The object was allocated as a raw pointer and every caller of such a +// function has that pointer (the `*mut Self` a vtable registered, a task's +// ctx, an `IntrusiveRc`'s `as_ptr()`), so the fix is to take `this: *mut Self`, +// do the `&mut` work through call-scoped reborrows, and release through `this` +// once they have ended, ideally by adopting the ref into a `bun_ptr::ScopedRef` +// first (src/runtime/api/bun/subprocess/SubprocessPipeReader.rs `finish`, +// src/runtime/shell/subproc.rs `PipeReader::on_reader_done`). +// +// Scope: the receiver itself as the argument, either coerced (`self`) or +// reborrowed in place (`&mut *self`), to the `deref` family of release +// functions or to `ScopedRef::adopt`. A pointer spelled out from the receiver +// (`deref(ptr::from_mut(self))`, `deref_nn(NonNull::from(self))`) is the same +// bug in a different spelling and has its own lint (#37703); the `&self` +// release methods some types expose (`self.deref()`) are a separate +// population. `Deref::deref(self)` is a borrow, not a release, and is ignored. +// +// Sibling guards: self-receiver-reclaim.test.ts (heap::take / Box::from_raw of +// the receiver), fn-long-mut-reborrow.test.ts. + +const root = path.resolve(import.meta.dir, "..", "..", ".."); +const rustSources = globAllSources().rust.filter(p => p.endsWith(".rs")); + +// Only scan files tracked in HEAD (a `git stash` round-trip can leave stray +// `.rs` files in the working tree; CI runs on a clean checkout). Same guard as +// dead-code-escapes.test.ts. +const tracked: Set | null = (() => { + const r = Bun.spawnSync({ + cmd: ["git", "-C", root, "ls-tree", "-r", "--name-only", "-z", "HEAD"], + stdout: "pipe", + stderr: "ignore", + }); + if (!r.success) return null; + return new Set(r.stdout.toString().split("\0").filter(Boolean)); +})(); + +// `deref(`, `deref_nn(`, `deref_from_thread(`, `deref_with_context(`, and the +// `rc_` variants, however path-qualified, plus `ScopedRef::adopt(` / +// `ScopedRef::::adopt(`. +const RELEASE = String.raw`(?:\b(?:rc_)?deref(?:_nn|_from_thread|_with_context)?|\bScopedRef(?:::<[^>]*>)?::adopt)\s*\(\s*`; + +// The receiver as the first argument: bare `self` or `&mut *self`, followed by +// the end of the argument. The terminator keeps `self.field`, `self.as_ptr()` +// and `&mut *self.inner` (things the receiver owns) out. +const RECEIVER = String.raw`(?:&\s*mut\s+\*\s*)?self\s*[,)]`; + +const BANNED = new RegExp(RELEASE + RECEIVER, "g"); + +// What may precede a match that is not a release: `fn deref(self)` is an item +// with a by-value receiver, and `Deref::deref(self)` / `::deref(self)` +// borrow. Checked on the text before the match rather than with lookbehinds +// at the head of BANNED, which would cost the regex its literal-prefix scan +// (about 6 s instead of under 0.1 s over the tree). +const NOT_A_RELEASE_BEFORE = /(?:\bfn\s+|Deref>?::)$/; + +function isRelease(stripped: string, index: number): boolean { + return !NOT_A_RELEASE_BEFORE.test(stripped.slice(Math.max(0, index - 64), index)); +} + +// Documented, ratcheted exceptions: files allowed to keep exactly N of the +// shape. Each is a release through the receiver that is being converted +// separately; lower the count when one is converted, do not add entries. +const ALLOW: Record = { + // `on_close` and `maybe_release` release the socket-ext ref through their + // `&mut self`; with the registry ref already gone that is the session's last + // ref (or the `ref_scope` guard built from the same receiver drops it a + // line later). + "src/http/h2_client/ClientSession.rs": 2, + // `detach` releases the per-stream ref through its receiver. + "src/http/h3_client/ClientSession.rs": 1, + // `detach_and_deref` releases the ref its caller transferred in; its own + // comment describes the call as the dealloc path. + "src/http/ProxyTunnel.rs": 1, +}; + +function findReleases(stripped: string): number[] { + return Array.from(stripped.matchAll(BANNED), m => m.index).filter(index => isRelease(stripped, index)); +} + +function lineOf(text: string, offset: number): number { + return text.slice(0, offset).split("\n").length; +} + +const counts: Record = {}; +const offenders: string[] = []; +let scanned = 0; +for (const abs of rustSources) { + const source = path.relative(root, abs).replaceAll(path.sep, "/"); + // `src/cli` is a symlink into `src/runtime/cli`; count each file once under + // its canonical path. + if (path.relative(root, realpathSync(abs)).replaceAll(path.sep, "/") !== source) continue; + if (tracked !== null && !tracked.has(source)) continue; + scanned++; + const content = await file(abs).text(); + // Strip full-line comments so prose mentions (including doc comments + // describing this shape) don't count. `[ \t]*`, not `\s*`, so blank lines + // survive and reported line numbers stay right. + const stripped = content.replace(/^[ \t]*\/\/.*$/gm, ""); + for (const offset of findReleases(stripped)) { + counts[source] = (counts[source] ?? 0) + 1; + if (counts[source] > (ALLOW[source] ?? 0)) { + offenders.push(`${source}:${lineOf(stripped, offset)}`); + } + } +} + +test("scans a non-empty set of tracked Rust sources", () => { + // Guards against the tracked/realpath filters above over-firing and leaving + // nothing to scan, which would make the ban below pass vacuously. + expect(scanned).toBeGreaterThan(0); +}); + +test("the pattern matches the banned spellings and nothing else", () => { + const banned = [ + // `PipeReader::on_reader_done(&mut self)` before it took the pointer. + "unsafe { PipeReader::deref(self) };", + "unsafe { ClientSession::deref(self) };", + "unsafe { RefCount::::deref(self) }", + "unsafe { Self::deref_with_context(self, ctx) };", + "unsafe { T::rc_deref(self) };", + "unsafe { ::rc_deref_with_context(self, ()) };", + "Self::deref_nn(self);", + "Self::deref_from_thread(self);", + "unsafe { Tunnel::deref(&mut *self) };", + "let _ref = unsafe { ScopedRef::adopt(self) };", + "let _ref = unsafe { bun_ptr::ScopedRef::::adopt(&mut *self) };", + // rustfmt-wrapped call. + "unsafe {\n PipeReader::deref(\n self,\n )\n};", + ]; + const allowed = [ + // Releasing through a raw pointer or a handle is the intended shape. + "unsafe { PipeReader::deref(this) };", + "unsafe { PipeReader::deref(pipe.as_ptr()) };", + "unsafe { RefCount::deref(writer) };", + "let _ref = unsafe { ScopedRef::adopt(this) };", + "let _ref = unsafe { ScopedRef::adopt(self.as_ctx_ptr()) };", + "unsafe { Self::deref(self.inner) };", + "unsafe { Self::deref(&mut *self.inner) };", + // Method-call spellings and `ScopedRef::new` (takes its own ref; balanced). + "pipe.deref();", + "self.deref();", + "let _keepalive = unsafe { ScopedRef::new(this) };", + // Definitions, `Deref` impls, unrelated names. + "pub fn deref(self) {", + "pub unsafe fn deref(this: *mut Self) {", + "fn deref(&self) -> &T {", + "core::ops::Deref::deref(self)", + "Deref::deref(self)", + "::deref(self)", + "ThreadSafe::adopt(self)", + "some_other_deref(self)", + "self.deref_mut()", + ]; + expect(banned.map(s => findReleases(s).length)).toEqual(banned.map(() => 1)); + expect(allowed.map(s => findReleases(s).length)).toEqual(allowed.map(() => 0)); +}); + +test("no method releases its own receiver's refcount", () => { + expect(offenders).toEqual([]); +}); + +test("allowlisted files still carry exactly their documented count", () => { + // Ratchet: when an allowlisted release is converted, lower its entry so the + // shape cannot come back into that file. + for (const [source, n] of Object.entries(ALLOW)) { + expect({ source, count: counts[source] ?? 0 }).toEqual({ source, count: n }); + } +});