From c18e2e12429d6ba47c99829c41c7331bb9c0c767 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 14 Aug 2026 02:00:11 +0000 Subject: [PATCH 1/2] threading: finish a WaitGroup without holding a reference into it past the release wait() returning lets the owner free the group, so the thread that finishes the last task may not touch the group after the store that lets wait() return. #34458 moved the last real access before that store, but finish(), Mutex::unlock() and the unlock's Futex::wake() all still took the group (or its mutex word) by reference, which asserts the memory for the whole call; Miri rejects the waiter's free under both aliasing models, and in the contended case the unlock tail really does form &self.state after the waiter may have freed it. WaitGroup::finish_raw(this: *const Self) does the work through raw pointers; its last access is Mutex::unlock_raw's releasing store, and the contended-path wake goes through Futex::wake_raw, which takes the word by address because every backend's wake side only uses the address as a key. finish(&self) stays for groups something else keeps alive (ThreadPool joins its workers before it drops, the Windows install queue is a static) and delegates to finish_raw; unlock(&self) delegates to unlock_raw; the per-OS unlock impls take the pointer as well, and Bun__unlock uses them directly. The os_unfair_lock_unlock extern takes the address for the same reason. LinkerContext's two source-map groups are the callers whose waiter frees right after wait() (the task slab in generate_chunks_in_parallel, the whole BundleV2 on the Bun.build error path); they use finish_raw. The crate's WaitGroup test now finishes through finish_raw and runs 500 iterations under Miri (the unfixed shape fails within 2..200 iterations on each of 12 seeds). Miri answers a FUTEX_WAKE on a freed word with EFAULT, which the Linux backend tolerates under cfg(miri) only; a real kernel never reports that for a private wake. bun_threading joins MIRI_CRATES, and test/internal/threading-miri.test.ts runs the crate under miri the same way. --- .github/workflows/miri.yml | 3 +- scripts/rust-miri.ts | 16 ++-- src/bundler/LinkerContext.rs | 20 ++++- src/threading/Futex.rs | 61 ++++++++++---- src/threading/Mutex.rs | 71 ++++++++++++---- src/threading/WaitGroup.rs | 122 +++++++++++++++++++-------- test/internal/threading-miri.test.ts | 62 ++++++++++++++ 7 files changed, 275 insertions(+), 80 deletions(-) create mode 100644 test/internal/threading-miri.test.ts diff --git a/.github/workflows/miri.yml b/.github/workflows/miri.yml index 1ee1447252a2..216a61b3a3b0 100644 --- a/.github/workflows/miri.yml +++ b/.github/workflows/miri.yml @@ -8,7 +8,7 @@ on: workflow_dispatch: pull_request: paths: - # The FFI-free crate set covered by MIRI_CRATES in scripts/rust-miri.ts + # The crate set covered by MIRI_CRATES in scripts/rust-miri.ts - "src/ast/**" - "src/base64/**" - "src/clap/**" @@ -22,6 +22,7 @@ on: - "src/ptr/**" - "src/resolve_builtins/**" - "src/shell_parser/**" + - "src/threading/**" - "src/wyhash/**" - "scripts/rust-miri.ts" - "Cargo.toml" diff --git a/scripts/rust-miri.ts b/scripts/rust-miri.ts index 6d690e4a82cf..43a0bbf3d402 100644 --- a/scripts/rust-miri.ts +++ b/scripts/rust-miri.ts @@ -1,11 +1,12 @@ #!/usr/bin/env bun /** - * `cargo miri test` for the FFI-free crate set. + * `cargo miri test` for the crates Miri can interpret end to end. * * Miri interprets MIR and catches UB (use-after-free, out-of-bounds, * uninit reads, data races, aliasing violations) at runtime. It cannot call - * foreign functions, so this only covers the pure-Rust corner of the - * workspace — which is also where `unsafe` density is highest. + * foreign functions beyond the libc subset it ships shims for, so this only + * covers the (nearly) pure-Rust corner of the workspace — which is also where + * `unsafe` density is highest. * * Aliasing model: `-Zmiri-tree-borrows`, not the default Stacked Borrows. * Stacked Borrows invalidates every raw pointer derived from `&mut self` the @@ -27,9 +28,11 @@ import { resolve } from "node:path"; const repo = resolve(import.meta.dirname, ".."); // Crates that pass `cargo miri test` under Tree Borrows. To add one it must -// (a) have at least one `#[test]`, (b) compile under `--cfg test`, (c) not -// call into `extern "C"` at test runtime — Miri reports -// `unsupported operation: can't call foreign function` if it does. +// (a) have at least one `#[test]`, (b) compile under `--cfg test`, (c) only +// call `extern "C"` functions Miri ships shims for at test runtime (libc's +// futex syscall and thread APIs, as bun_threading does, are fine; anything +// vendored is not) — Miri reports +// `unsupported operation: can't call foreign function` otherwise. const MIRI_CRATES = [ "bun_ast", "bun_base64", @@ -44,6 +47,7 @@ const MIRI_CRATES = [ "bun_ptr", "bun_resolve_builtins", "bun_shell_parser", + "bun_threading", "bun_wyhash", ]; diff --git a/src/bundler/LinkerContext.rs b/src/bundler/LinkerContext.rs index 3706c548f8ce..e3af3c9e6151 100644 --- a/src/bundler/LinkerContext.rs +++ b/src/bundler/LinkerContext.rs @@ -1366,9 +1366,17 @@ impl SourceMapDataTask { // pointee outlives every task (joined via `line_offset_wait_group`). let ctx = task.ctx.expect("SourceMapDataTask.ctx"); scopeguard::defer! { - // Both `&self` methods (atomic ops) — safe via `ParentRef::Deref`. ctx.mark_pending_task_done(); - ctx.source_maps.line_offset_wait_group.finish(); + // SAFETY: the linker is blocked in `line_offset_wait_group.wait()` + // (or will be) until this finish, so the group is live; it frees the + // tasks as soon as `wait()` returns (`generate_chunks_in_parallel`), + // which is why this goes through `finish_raw` and is the last + // statement to touch `ctx`. + unsafe { + WaitGroup::finish_raw( + &raw const (*ctx.as_const_ptr()).source_maps.line_offset_wait_group, + ) + }; } // SAFETY: ctx is BundleV2.linker; container_of recovers the parent. We @@ -1403,9 +1411,13 @@ impl SourceMapDataTask { // pointee outlives every task (joined via `quoted_contents_wait_group`). let ctx = task.ctx.expect("SourceMapDataTask.ctx"); scopeguard::defer! { - // Both `&self` methods (atomic ops) — safe via `ParentRef::Deref`. ctx.mark_pending_task_done(); - ctx.source_maps.quoted_contents_wait_group.finish(); + // SAFETY: as in `run_line_offset`, for `quoted_contents_wait_group`. + unsafe { + WaitGroup::finish_raw( + &raw const (*ctx.as_const_ptr()).source_maps.quoted_contents_wait_group, + ) + }; } // SAFETY: see `run_line_offset` — raw-ptr container_of, no `&mut` diff --git a/src/threading/Futex.rs b/src/threading/Futex.rs index 3d2b23767fe2..e038d360cc87 100644 --- a/src/threading/Futex.rs +++ b/src/threading/Futex.rs @@ -55,6 +55,19 @@ pub fn wait_forever(ptr: &AtomicU32, expect: u32) { /// Unblocks at most `max_waiters` callers blocked in a `wait()` call on `ptr`. #[cold] pub fn wake(ptr: &AtomicU32, max_waiters: u32) { + wake_raw(core::ptr::from_ref(ptr), max_waiters); +} + +/// [`wake`] for a word that may be freed before this runs: a mutex's unlock +/// tail (`Mutex::unlock_raw`), where the store that released the lock has to +/// be the primitive's last access to its own memory because the thread it +/// released may free the primitive at once. A `&AtomicU32` would assert the +/// word live until this returns. Every backend's wake side uses the address +/// only as a key and never reads the word, so this needs no `unsafe`: the +/// worst a freed or reused address yields is a spurious wakeup, which every +/// `wait()` loop tolerates. +#[cold] +pub(crate) fn wake_raw(ptr: *const AtomicU32, max_waiters: u32) { // Avoid calling into the OS if there's nothing to wake up. if max_waiters == 0 { return; @@ -106,7 +119,7 @@ mod unsupported_impl { unsupported() } - pub(super) fn wake(_ptr: &AtomicU32, _max_waiters: u32) { + pub(super) fn wake(_ptr: *const AtomicU32, _max_waiters: u32) { unsupported() } @@ -162,11 +175,12 @@ mod windows_impl { } } - pub(super) fn wake(ptr: &AtomicU32, max_waiters: u32) { - let address: *const c_void = ptr.as_ptr().cast(); + pub(super) fn wake(ptr: *const AtomicU32, max_waiters: u32) { + let address: *const c_void = ptr.cast(); debug_assert!(max_waiters != 0); - // SAFETY: address points at a live AtomicU32. + // SAFETY: RtlWakeAddress* only look `address` up in the waiter table; + // they never access the memory, so it need not be live (see `super::wake_raw`). unsafe { match max_waiters { 1 => windows::ntdll::RtlWakeAddressSingle(address), @@ -261,7 +275,7 @@ mod darwin_impl { } } - pub(super) fn wake(ptr: &AtomicU32, max_waiters: u32) { + pub(super) fn wake(ptr: *const AtomicU32, max_waiters: u32) { let flags = c::UL { op: c::ULOp::COMPARE_AND_WAIT, no_errno: true, @@ -270,8 +284,10 @@ mod darwin_impl { }; loop { - let addr: *const c_void = ptr.as_ptr().cast(); - // SAFETY: addr points at a live AtomicU32. + let addr: *const c_void = ptr.cast(); + // SAFETY: __ulock_wake only keys the waiter lookup on `addr`; it never + // accesses the memory (hence no EFAULT below), so it need not be live + // (see `super::wake_raw`). let status = unsafe { c::__ulock_wake(flags, addr, 0) }; if status >= 0 { @@ -346,16 +362,18 @@ mod linux_impl { } } - pub(super) fn wake(ptr: &AtomicU32, max_waiters: u32) { + pub(super) fn wake(ptr: *const AtomicU32, max_waiters: u32) { use bun_sys::linux; let val: u32 = match i32::try_from(max_waiters) { Ok(v) => v as u32, Err(_) => i32::MAX as u32, }; - // SAFETY: ptr.as_ptr() is a valid *const u32 for the duration of the call. + // SAFETY: a private FUTEX_WAKE keys the waiter lookup on the address + // alone (`get_futex_key` does not touch the memory), so `ptr` need not + // point to live memory (see `super::wake_raw`). let rc = unsafe { linux::futex_3arg( - ptr.as_ptr().cast(), + ptr.cast(), linux::FutexOp { cmd: linux::FutexCmd::WAKE, private: true, @@ -367,7 +385,13 @@ mod linux_impl { match linux::E::init(rc) { linux::E::SUCCESS => {} // successful wake up linux::E::INVAL => {} // invalid futex_wait() on ptr done elsewhere - linux::E::FAULT => panic!("futex_wake() returned EFAULT unexpectedly"), // pointer became invalid while doing the wake + // The kernel only reports this for an address outside user space. + #[cfg(not(miri))] + linux::E::FAULT => panic!("futex_wake() returned EFAULT unexpectedly"), + // Miri reports it for a word that has already been freed, which + // `super::wake_raw` allows. + #[cfg(miri)] + linux::E::FAULT => {} _ => panic!("Unexpected futex_wake() return code"), } } @@ -427,15 +451,16 @@ mod freebsd_impl { } } - pub(super) fn wake(ptr: &AtomicU32, max_waiters: u32) { + pub(super) fn wake(ptr: *const AtomicU32, max_waiters: u32) { // The kernel reads n_wake as `int`; passing maxInt(u32) truncates to // -1 and umtxq_signal_queue's `++ret >= n_wake` returns after one // wakeup. _umtx_op(2): "Specify INT_MAX to wake up all waiters." let n: c_ulong = max_waiters.min(c_int::MAX as u32) as c_ulong; - // SAFETY: ptr.as_ptr() is valid for the duration of the call. + // SAFETY: a private WAKE only keys the waiter lookup on the address; it + // never accesses the memory, so it need not be live (see `super::wake_raw`). let rc = unsafe { libc::_umtx_op( - ptr.as_ptr().cast::(), + ptr.cast::().cast_mut(), libc::UMTX_OP_WAKE_PRIVATE, n, core::ptr::null_mut(), // there is no timeout struct @@ -480,14 +505,16 @@ mod wasm_impl { } } - pub fn wake(ptr: &AtomicU32, max_waiters: u32) { + pub fn wake(ptr: *const AtomicU32, max_waiters: u32) { #[cfg(not(target_feature = "atomics"))] compile_error!("WASI target missing cpu feature 'atomics'"); debug_assert!(max_waiters != 0); - // SAFETY: ptr.as_ptr() is a valid aligned *mut i32 (AtomicU32 has the same layout). + // SAFETY: memory.atomic.notify only keys on the (aligned, in-bounds) + // address; linear memory is never unmapped, so a freed word is still a + // valid key (see `super::wake_raw`). AtomicU32 has the layout of i32. let woken_count = unsafe { - core::arch::wasm32::memory_atomic_notify(ptr.as_ptr().cast::(), max_waiters) + core::arch::wasm32::memory_atomic_notify(ptr.cast::().cast_mut(), max_waiters) }; let _ = woken_count; // can be 0 when linker flag 'shared-memory' is not enabled } diff --git a/src/threading/Mutex.rs b/src/threading/Mutex.rs index c4378a0455a8..b036f6af3b81 100644 --- a/src/threading/Mutex.rs +++ b/src/threading/Mutex.rs @@ -64,7 +64,24 @@ impl Mutex { /// Releases the mutex which was previously acquired with `lock()` or `try_lock()`. /// It is undefined behavior if the mutex is unlocked from a different thread that it was locked from. pub fn unlock(&self) { - self.impl_.unlock() + // SAFETY: `self` is held by this thread (fn contract) and live for the + // whole call. + unsafe { Self::unlock_raw(self) } + } + + /// [`unlock`](Self::unlock) for a critical section whose exit is what lets + /// another thread free the mutex's owner (`WaitGroup::finish_raw`). A `&self` + /// argument would assert the mutex's storage, padding included, until this + /// returns; here the store that releases the lock is the last access to + /// `*this`, and the futex wake that may follow it goes by address only + /// ([`Futex::wake_raw`](crate::futex::wake_raw)). + /// + /// # Safety + /// `this` must point to a mutex this thread holds. It stays valid until the + /// lock is released; from then on another thread may free it. + pub(crate) unsafe fn unlock_raw(this: *const Self) { + // SAFETY: the lock is still held, so `*this` is live (fn contract). + unsafe { Impl::unlock_raw(&raw const (*this).impl_) } } /// Debug-only check that the calling thread already holds this mutex. @@ -185,11 +202,15 @@ impl DebugImpl { self.locking_thread.store(current_id, Ordering::Relaxed); } + /// See [`Mutex::unlock_raw`] for the contract. #[inline] - fn unlock(&self) { - debug_assert!(self.locking_thread.load(Ordering::Relaxed) == current_thread_id()); - self.locking_thread.store(0, Ordering::Relaxed); - self.impl_.unlock(); + unsafe fn unlock_raw(this: *const Self) { + // SAFETY: the lock is still held, so `*this` is live (fn contract). + unsafe { + debug_assert!((*this).locking_thread.load(Ordering::Relaxed) == current_thread_id()); + (*this).locking_thread.store(0, Ordering::Relaxed); + ReleaseImpl::unlock_raw(&raw const (*this).impl_); + } } } @@ -242,10 +263,15 @@ impl WindowsImpl { AcquireSRWLockExclusive(&self.srwlock) } - fn unlock(&self) { - // SAFETY: caller acquired the lock on this thread (`Mutex::unlock` - // contract); releasing without ownership is documented UB on Windows. - unsafe { bun_sys::windows::kernel32::ReleaseSRWLockExclusive(self.srwlock.get()) } + /// See [`Mutex::unlock_raw`] for the contract. + unsafe fn unlock_raw(this: *const Self) { + // SAFETY: this thread holds the lock (fn contract), so `*this` is live + // up to the release inside the call; releasing without ownership is + // documented UB on Windows. + unsafe { + let srwlock = core::cell::UnsafeCell::raw_get(&raw const (*this).srwlock); + bun_sys::windows::kernel32::ReleaseSRWLockExclusive(srwlock) + } } } @@ -277,12 +303,15 @@ pub(crate) struct OsUnfairLock { // The type encodes the only pointer-validity precondition, and Apple's runtime // detects misuse (recursive lock / unowned unlock) by aborting — which is safe // — so `safe fn` discharges the link-time proof and callers need no `unsafe`. +// `os_unfair_lock_unlock` is the exception: a reference would assert the word +// live until the call returns, but once it releases the lock another thread +// may free the word (`Mutex::unlock_raw`), so it takes the address. #[cfg(target_vendor = "apple")] unsafe extern "C" { #[cfg(debug_assertions)] safe fn os_unfair_lock_trylock(lock: &core::cell::UnsafeCell) -> bool; safe fn os_unfair_lock_lock(lock: &core::cell::UnsafeCell); - safe fn os_unfair_lock_unlock(lock: &core::cell::UnsafeCell); + fn os_unfair_lock_unlock(lock: *mut OsUnfairLock); } #[cfg(target_vendor = "apple")] @@ -302,8 +331,11 @@ impl DarwinImpl { os_unfair_lock_lock(&self.oul) } - fn unlock(&self) { - os_unfair_lock_unlock(&self.oul) + /// See [`Mutex::unlock_raw`] for the contract. + unsafe fn unlock_raw(this: *const Self) { + // SAFETY: this thread holds the lock (fn contract), so `*this` is live + // up to the release inside the call, which is its last access. + unsafe { os_unfair_lock_unlock(core::cell::UnsafeCell::raw_get(&raw const (*this).oul)) } } } @@ -382,7 +414,8 @@ impl FutexImpl { } } - fn unlock(&self) { + /// See [`Mutex::unlock_raw`] for the contract. + unsafe fn unlock_raw(this: *const Self) { // Unlock the mutex and wake up a waiting thread if any. // // A waiting thread will acquire with `contended` instead of `locked` @@ -390,11 +423,17 @@ impl FutexImpl { // // Release barrier ensures the critical section happens before we let go of the lock // and that our critical section happens before the next lock holder grabs the lock. - let state = self.state.swap(Self::UNLOCKED, Ordering::Release); + // + // SAFETY: the lock is still held, so `*this` is live (fn contract). + let state_ptr = unsafe { &raw const (*this).state }; + // SAFETY: as above; the swap is what releases the lock, and the last + // access to `*this`. The wake below goes by address because the thread + // the swap releases may have freed the mutex by the time it runs. + let state = unsafe { (*state_ptr).swap(Self::UNLOCKED, Ordering::Release) }; debug_assert!(state != Self::UNLOCKED); if state == Self::CONTENDED { - Futex::wake(&self.state, 1); + Futex::wake_raw(state_ptr, 1); } } } @@ -410,7 +449,7 @@ unsafe extern "C" fn Bun__lock(ptr: *mut ReleaseImpl) { #[unsafe(no_mangle)] unsafe extern "C" fn Bun__unlock(ptr: *mut ReleaseImpl) { // SAFETY: C caller passes a valid, initialized ReleaseImpl pointer that this thread locked. - unsafe { (*ptr).unlock() } + unsafe { ReleaseImpl::unlock_raw(ptr) } } #[unsafe(no_mangle)] diff --git a/src/threading/WaitGroup.rs b/src/threading/WaitGroup.rs index 4f45613b1555..63a9dd5306f3 100644 --- a/src/threading/WaitGroup.rs +++ b/src/threading/WaitGroup.rs @@ -42,37 +42,75 @@ impl WaitGroup { self.add(1); } + /// Counts one task as done. Only for a group that something other than + /// [`wait`](Self::wait) keeps alive past this call (`ThreadPool` joins its + /// workers before it is dropped; the install queue on Windows is a + /// `static`): `&self` asserts the group's storage until this returns, and + /// a `wait()` this call releases may return before then. When `wait()` + /// returning is what lets the owner free the group, use + /// [`finish_raw`](Self::finish_raw). pub fn finish(&self) { + // SAFETY: the group outlives this call (fn contract). + unsafe { Self::finish_raw(self) } + } + + /// [`finish`](Self::finish) for a group whose owner may free it as soon as + /// `wait()` returns (`LinkerContext`'s source-map groups: the waiter frees + /// the tasks' storage on the line after `wait()`). `wait()` can return once + /// the count is published as 0 and the mutex is released, so this thread's + /// last access to the group is the store that releases the mutex; no frame + /// between here and that store holds a reference into the group. + /// + /// # Safety + /// `this` must point to a live `WaitGroup` whose count includes the task + /// being finished. Once this lets a `wait()` return, the owner may free it. + pub unsafe fn finish_raw(this: *const Self) { // Fast path: decrement lock-free while there are other outstanding // tasks. We cannot unconditionally `fetch_sub(1)` and then lock/signal // for the last one: the moment `raw_count` reaches 0 a concurrent - // `wait()` can observe it, return, and the caller drop the `WaitGroup`, - // so any later `self.mutex`/`self.cond` access is a use-after-free. - let mut old = self.raw_count.load(Ordering::Relaxed); - while old > 1 { - match self.raw_count.compare_exchange_weak( - old, - old - 1, - Ordering::AcqRel, - Ordering::Relaxed, - ) { - Ok(_) => return, - Err(cur) => old = cur, + // `wait()` can observe it, return, and the owner free the group, so any + // later `mutex`/`cond` access would be a use-after-free. + // + // SAFETY: the group is live until a finisher publishes 0 (fn contract); + // an exchange here leaves the count at >= 1, so that finisher is a later + // call and this one is done with `*this` once the exchange lands. + unsafe { + let mut old = (*this).raw_count.load(Ordering::Relaxed); + while old > 1 { + match (*this).raw_count.compare_exchange_weak( + old, + old - 1, + Ordering::AcqRel, + Ordering::Relaxed, + ) { + Ok(_) => return, + Err(cur) => old = cur, + } } } // We are (or a concurrent `add` may yet make us not) the last one. // Publish `raw_count == 0` only while holding the mutex so `wait()`, - // which checks the count under the same mutex, cannot return until our - // `unlock()` below. Signal before unlocking so the waiter's reacquire - // serializes after every `self` access we make. - self.mutex.lock(); - let old_count = self.raw_count.fetch_sub(1, Ordering::AcqRel); - debug_assert!(old_count >= 1); - self.cond.signal(); - self.mutex.unlock(); + // which checks the count under the same mutex, cannot return until the + // unlock below. Signal before unlocking so the waiter's reacquire + // serializes after every access we make. + // + // SAFETY: `wait()` cannot return, so the group is live (fn contract), + // until the unlock releases the mutex; `unlock_raw` makes that release + // the last access to `*this`, where `(*this).mutex.unlock()` would hold + // `&Mutex` past it. + unsafe { + (*this).mutex.lock(); + let old_count = (*this).raw_count.fetch_sub(1, Ordering::AcqRel); + debug_assert!(old_count >= 1); + (*this).cond.signal(); + Mutex::unlock_raw(&raw const (*this).mutex); + } } + /// Blocks until the count reaches 0. Once this returns, every + /// [`finish_raw`](Self::finish_raw) that contributed to that is done with + /// the group, so the caller may free it. pub fn wait(&self) { self.mutex.lock(); // crate::Mutex is a raw lock/unlock wrapper (no RAII guard), so unlock @@ -90,26 +128,38 @@ impl WaitGroup { mod tests { use super::*; - // After `wait()` returns the caller may drop the `WaitGroup`; `finish()` - // must therefore not touch `self` once it has published `raw_count == 0`. + // `wait()` returning lets the owner free the group (see `finish_raw`), so + // `finish_raw()` must neither touch nor hold a reference into the group + // once it has let `wait()` return. Under Miri (`bun run rust:miri`) the + // `Box` drop is rejected whenever a frame of the finishing thread still + // holds a reference into the group; natively it is a use-after-free race. #[test] - fn wait_returning_means_finish_is_done_with_self() { - for _ in 0..10_000 { + fn wait_returning_means_finish_raw_is_done_with_the_group() { + // Miri takes ~30ms per iteration and its scheduler produces the + // offending interleaving about once per 60 iterations (12 seeds: worst + // case 200), so 500 keeps the run short without losing the failure. + #[cfg(miri)] + const ITERATIONS: usize = 500; + #[cfg(not(miri))] + const ITERATIONS: usize = 10_000; + + for _ in 0..ITERATIONS { let wg = Box::into_raw(Box::new(WaitGroup::init_with_count(1))); - struct SendPtr(*mut WaitGroup); - // SAFETY: `WaitGroup` is `Sync`; the raw pointer is only ever - // dereferenced while the pointee is live (joined below). + struct SendPtr(*const WaitGroup); + // SAFETY: `WaitGroup` is `Sync`; the pointer is only used under + // `finish_raw`'s contract, which the `wait()` below upholds. unsafe impl Send for SendPtr {} let p = SendPtr(wg); - let t = std::thread::spawn(move || { - let p = p; - // SAFETY: `wg` is live until `drop(Box::from_raw(..))` below, - // which happens-before `join()`, so the pointee outlives this - // deref iff `finish()` is done with `self` by the time - // `wait()` returns — the property under test. - unsafe { (*p.0).finish() }; - }); - // SAFETY: `wg` is the freshly-boxed allocation; sole owner here. + let t = std::thread::Builder::new() + .spawn(move || { + let p = p; + // SAFETY: `wg` stays live until `wait()` returns on the main + // thread, and this call is what lets it return (fn contract). + unsafe { WaitGroup::finish_raw(p.0) }; + }) + .unwrap(); + // SAFETY: `wg` is the freshly-boxed allocation and this is its sole + // owner; `wait()` returning means `finish_raw` is done with it. unsafe { (*wg).wait(); drop(Box::from_raw(wg)); diff --git a/test/internal/threading-miri.test.ts b/test/internal/threading-miri.test.ts new file mode 100644 index 000000000000..49a080fc58a1 --- /dev/null +++ b/test/internal/threading-miri.test.ts @@ -0,0 +1,62 @@ +/** + * `bun_threading` (src/threading/) must stay clean under `cargo miri test`; it + * is one of the crates `bun run rust:miri` (scripts/rust-miri.ts) covers. + * + * The property this was added for is `WaitGroup`'s: `wait()` returning lets the + * owner free the group, so the thread that finishes the last task must neither + * touch the group nor still hold a reference into it once it has let `wait()` + * return (`WaitGroup::finish_raw`, and the `Mutex::unlock_raw` / + * `Futex::wake_raw` it rests on). Natively the unfixed shape's only extra work + * after the releasing store is a futex wake keyed by the freed address, which + * the kernel ignores, so there is nothing a bun-level test can observe; the + * discriminator is miri itself. With `finish()` taking `&self` through the + * release, the crate's own `wait_group` test is rejected at its `Box` drop under + * both Tree Borrows (pinned here, as in `rust:miri`) and the default Stacked + * Borrows. + * + * Skipped where miri is not installed or the cargo workspace is not resolvable + * (test-only CI lanes run a prebuilt binary and lack vendor/lolhtml); same + * prerequisite check as linear-fifo.test.ts and scripts/rust-miri.ts. + */ +import { expect, test } from "bun:test"; +import { existsSync } from "node:fs"; +import path from "node:path"; + +const cargoBin = Bun.which("cargo"); +const repoRoot = path.resolve(import.meta.dir, "..", ".."); +const workspaceResolvable = + existsSync(path.join(repoRoot, "vendor", "lolhtml", "Cargo.toml")) && + existsSync(path.join(repoRoot, "build", "debug", "codegen", "build_options.rs")); +const miriAvailable = + !!cargoBin && + workspaceResolvable && + Bun.spawnSync({ + cmd: [cargoBin, "miri", "--version"], + cwd: repoRoot, + stdout: "ignore", + stderr: "ignore", + timeout: 30_000, + }).exitCode === 0; + +test.skipIf(!miriAvailable)( + "bun_threading unit tests are clean under Tree Borrows miri", + async () => { + await using proc = Bun.spawn({ + cmd: [cargoBin!, "miri", "test", "--locked", "-p", "bun_threading"], + cwd: repoRoot, + env: { ...process.env, MIRIFLAGS: "-Zmiri-tree-borrows" }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + if (exitCode !== 0) { + // Surface miri's diagnostic so the gate/CI log shows the actual UB. + console.error(stderr || stdout); + } + expect(stderr).not.toContain("Undefined Behavior"); + expect(exitCode).toBe(0); + }, + // Compiles the crate's dependencies for miri, then interprets the WaitGroup + // test's 500 thread spawns: ~30s on a warm tree, more on a cold one. + 180_000, +); From 59bec166bcb5cabb92eb2da2377cd1f52061bbdd Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 14 Aug 2026 03:24:23 +0000 Subject: [PATCH 2/2] s3: reach the streaming download task through its raw pointer on both threads S3HttpDownloadStreamingTask is used by the HTTP thread and the JS thread at the same time for its whole life. Its &mut self methods (process_http_callback and update_state on the HTTP thread, report_progress on the JS thread) each asserted the whole task, mutex word included, for the duration of the call while the other thread may be taking the lock, and process_http_callback's receiver and lock guard were still live when its final unlock let the JS thread free the task. Convert those methods to this: *mut Self functions that form references one field or one call at a time, fold release_portable into Drop (which runs once the HTTP thread is done), and make both HTTP-thread critical sections end in Mutex::unlock_raw, which becomes pub for this, so the releasing store is the HTTP thread's last access to the task. A source lint bans &mut self receivers in the task's impl block and guard / receiver spellings of the mutex in the s3 directory. --- src/runtime/webcore/s3/download_stream.rs | 361 ++++++++++-------- src/threading/Mutex.rs | 10 +- .../s3-download-task-raw-access.test.ts | 245 ++++++++++++ 3 files changed, 450 insertions(+), 166 deletions(-) create mode 100644 test/internal/source-lints/s3-download-task-raw-access.test.ts diff --git a/src/runtime/webcore/s3/download_stream.rs b/src/runtime/webcore/s3/download_stream.rs index a07a56e11f1b..1d593a28a5bc 100644 --- a/src/runtime/webcore/s3/download_stream.rs +++ b/src/runtime/webcore/s3/download_stream.rs @@ -15,6 +15,21 @@ use bun_threading::Mutex; bun_core::declare_scope!(S3, hidden); +/// One streaming GET, shared by the HTTP thread (which appends each chunk under `mutex`) and +/// the JS thread (which reports the chunks from `on_response` under the same mutex and frees +/// the task once it has read `has_more == false`), from the moment it is scheduled until that +/// free. +/// +/// While both threads are live the task is reached only through its raw pointer, with any +/// reference scoped to a single field or a single call: a `&mut self` method would assert +/// the whole task, `mutex` and the atomics included, for the duration of the call, while the +/// other thread may be writing to exactly those words (every `lock()` attempt is one), and on +/// the HTTP thread it would still be asserted when the final unlock lets the JS thread free the +/// task. `&self` helpers are fine: they assert only the non-atomic fields, which only the lock +/// holder writes. Both HTTP-thread critical sections release through `Mutex::unlock_raw` for +/// the same reason (see `process_http_callback`). Enforced by +/// test/internal/source-lints/s3-download-task-raw-access.test.ts; `Drop` runs once the HTTP +/// thread is done with the task, so its `&mut self` is not subject to this. pub struct S3HttpDownloadStreamingTask { // `MaybeUninit` because `AsyncHTTP` contains non-null references, so // `mem::zeroed()` can't be used here (mirrors `S3HttpSimpleTask`). @@ -67,76 +82,85 @@ impl S3HttpDownloadStreamingTask { self.state.store(state.0, Ordering::Relaxed); } - fn report_progress(&mut self, state: State) { + /// JS thread, under `mutex`: delivers what the HTTP thread has recorded so far to + /// `callback`. + /// + /// # Safety + /// `this` is live and this thread holds its `mutex`. The HTTP thread may be trying to take + /// the mutex throughout, and the callback may reach the task again (`on_stream_cancelled`), + /// so no reference here spans more than one field or one call; see the type docs. + unsafe fn report_progress(this: *mut Self, state: State) { let has_more = state.has_more(); let failed = match state.status_code() { 200 | 204 | 206 => state.request_error() != 0, _ => true, }; - bun_core::scoped_log!( - S3, - "reportProgres failed: {} has_more: {} len: {}", - failed, - has_more, - self.reported_response_buffer.list.len() - ); + // SAFETY: fn contract. What the callbacks borrow is either a local (`empty`, `chunk`, + // taken out of the task first) or, for `message`, the buffer's heap bytes, which the + // HTTP thread cannot touch until this thread unlocks. + unsafe { + bun_core::scoped_log!( + S3, + "reportProgres failed: {} has_more: {} len: {}", + failed, + has_more, + (*this).reported_response_buffer.list.len() + ); - if failed { - if has_more { - return; - } - let empty = MutableString::default(); - let mut code: &[u8] = b"UnknownError"; - let mut message: &[u8] = b"an unexpected error has occurred"; - let parsed; - if let Some(req_err) = self.request_error { - code = req_err.name().as_bytes(); - } else { - let bytes = self.reported_response_buffer.list.as_slice(); - if !bytes.is_empty() { - message = bytes; + if failed { + if has_more { + return; } - parsed = xml_response::parse_error(bytes); - if let Some(error) = &parsed { - code = error.code.as_deref().unwrap_or(code); - message = error.message.as_deref().unwrap_or(message); + let empty = MutableString::default(); + let mut code: &[u8] = b"UnknownError"; + let mut message: &[u8] = b"an unexpected error has occurred"; + let parsed; + if let Some(req_err) = (*this).request_error { + code = req_err.name().as_bytes(); + } else { + let bytes = (*this).reported_response_buffer.list.as_slice(); + if !bytes.is_empty() { + message = bytes; + } + parsed = xml_response::parse_error(bytes); + if let Some(error) = &parsed { + code = error.code.as_deref().unwrap_or(code); + message = error.message.as_deref().unwrap_or(message); + } } + ((*this).callback)( + &empty, + false, + Some(S3Error { code, message }), + (*this).callback_context.as_ptr().cast(), + ); + return; } - (self.callback)( - &empty, - false, - Some(S3Error { code, message }), - self.callback_context.as_ptr().cast(), - ); - return; - } - // dont report empty chunks if we have more data to read - if !has_more || self.reported_response_buffer.list.len() > 0 { - // `core::mem::take` transfers ownership of the buffer, leaving an - // empty MutableString behind. - let chunk = core::mem::take(&mut self.reported_response_buffer); - (self.callback)( - &chunk, - has_more, - None, - self.callback_context.as_ptr().cast(), - ); - self.reported_response_buffer.reset(); + // dont report empty chunks if we have more data to read + if !has_more || (*this).reported_response_buffer.list.len() > 0 { + let chunk = core::mem::take(&mut (*this).reported_response_buffer); + ((*this).callback)( + &chunk, + has_more, + None, + (*this).callback_context.as_ptr().cast(), + ); + (*this).reported_response_buffer.reset(); + } } } /// this is the task callback from the last task result and is always in the main thread /// /// # Safety - /// `this` must be a live heap pointer produced by `Self::new`; the event loop guarantees - /// exclusive main-thread access for the duration of this call. When the loaded state's - /// `has_more` is false this call reclaims and drops the allocation exactly once. + /// `this` must be a live heap pointer produced by `Self::new`, and this must be the run of + /// the task posted for it (one per post). When the loaded state's `has_more` is false this + /// call reclaims and drops the allocation exactly once. pub(crate) fn on_response(this: *mut Self) { - // SAFETY: `this` is a live heap allocation created via `Self::new`; the event loop - // guarantees exclusive access on the main thread for the duration of this callback. - // Each access below is scoped so no borrow spans `report_progress` (which invokes - // the chunk callback). + // SAFETY: fn contract. The HTTP thread may be using the task concurrently until it + // publishes the final state, so every access below is a field access or a single call + // (see the type docs). unsafe { (*this).mutex.lock() }; // the state is atomic let's load it once // SAFETY: as above. @@ -146,8 +170,10 @@ impl S3HttpDownloadStreamingTask { // `report_progress` still unlocks + deinits. let this_ptr = this; scopeguard::defer! { - // SAFETY: `this_ptr` was allocated via `Box::new` in `Self::new`; once - // `has_more == false` we are the sole owner (HTTP thread will not call back again). + // SAFETY: `this_ptr` was allocated via `Box::new` in `Self::new`. Once we have read + // `has_more == false` under the mutex, the HTTP thread is done with the task (its + // last access was the release we acquired from; it will not call back again), so we + // are the sole owner. unsafe { (*this_ptr).mutex.unlock(); if !has_more { @@ -166,128 +192,131 @@ impl S3HttpDownloadStreamingTask { .store(false, Ordering::Relaxed) }; } - // SAFETY: as above; exclusive borrow scoped to the call. - unsafe { (*this).report_progress(state) }; + // SAFETY: as above, and we hold the mutex. + unsafe { Self::report_progress(this, state) }; } - /// this function is only called from the http callback in the HTTPThread and returns true if we - /// should wait until we are done buffering the response body to report - /// should only be called when already locked - fn update_state( - &mut self, + /// HTTP thread, under `mutex` (from `process_http_callback`): folds `result` into `state`, + /// publishes it and takes over the request's current `AsyncHTTP`. Returns true if we should + /// wait until we are done buffering the response body to report (the body of a failed + /// request is its error document). + /// + /// # Safety + /// `this` is live and this thread holds its `mutex`; `async_http` is the HTTP thread's live + /// copy of the request. + unsafe fn update_state( + this: *mut Self, async_http: &mut AsyncHTTP<'static>, - // borrowed so the caller (process_http_callback) can still read - // `result.body` afterward. + // borrowed so the caller can still take `result`'s body afterward. result: &HTTPClientResult, state: &mut State, ) -> bool { let is_done = !result.has_more; - // if we got a error or fail wait until we are done buffering the response body to report - let wait_until_done; - { - state.set_has_more(!is_done); - - self.request_error = result.fail; - state.set_request_error(if result.fail.is_some() { 1 } else { 0 }); - if state.status_code() == 0 { - // `certificate_info` / `metadata` free their owned buffers via `Drop` - // when `HTTPClientResult` is dropped by the caller after this returns. - if let Some(m) = &result.metadata { - state.set_status_code(m.response.status_code); - } - } - match state.status_code() { - 200 | 204 | 206 => wait_until_done = state.request_error() != 0, - _ => wait_until_done = true, + state.set_has_more(!is_done); + state.set_request_error(if result.fail.is_some() { 1 } else { 0 }); + if state.status_code() == 0 { + // `certificate_info` / `metadata` free their owned buffers via `Drop` + // when `HTTPClientResult` is dropped by the caller after this returns. + if let Some(m) = &result.metadata { + state.set_status_code(m.response.status_code); } - // store the new state - self.set_state(*state); - // SAFETY: `async_http` points to a live AsyncHTTP owned by the HTTP thread; a - // bitwise read+write copies its current state into `self.http` without running - // destructors (the HTTP thread retains ownership of the source until the request - // completes). `self.http` was previously initialised in - // `client::download_stream`. - unsafe { core::ptr::write(self.http.as_mut_ptr(), core::ptr::read(async_http)) }; + } + let wait_until_done = match state.status_code() { + 200 | 204 | 206 => state.request_error() != 0, + _ => true, + }; + // SAFETY: fn contract (field accesses only, see the type docs). The bitwise read+write + // copies `async_http`'s current state into `http` without running destructors: the + // HTTP thread retains ownership of the source until the request completes, and `http` + // was initialised in `client::download_stream`. + unsafe { + (*this).request_error = result.fail; + (*this).set_state(*state); + core::ptr::write((*this).http.as_mut_ptr(), core::ptr::read(async_http)); } wait_until_done } - /// this functions is only called from the http callback in the HTTPThread and returns true if - /// we should enqueue another task - fn process_http_callback( - &mut self, + /// HTTP thread: records one result of the request under `mutex` and returns true if the + /// caller should post the task to the JS thread. + /// + /// On the final result the unlock at the end can be what frees the task: if a task is + /// already queued (`has_schedule_callback` is set, so nothing gets posted), the JS thread + /// may be blocked in `on_response`'s `lock()`, and once let in it sees `has_more == false` + /// and frees the task while this thread is still returning. Hence, on top of the type's + /// field-access rule, the section is not a `lock_guard()` one (its guard would unlock + /// through a `&Mutex` into the task) and is released by `Mutex::unlock_raw`, which makes + /// the releasing store this thread's last access to the task. + /// + /// # Safety + /// `this` is the task registered with the `AsyncHTTP` whose result this is; it is live + /// because this result has not been published yet. `async_http` is the HTTP thread's live + /// copy of the request. + unsafe fn process_http_callback( + this: *mut Self, async_http: &mut AsyncHTTP<'static>, mut result: HTTPClientResult, ) -> bool { - // lets lock and unlock to be safe we know the state is not in the middle of a callback when locked - // The RAII guard unlocks on every - // return path. The guard holds the mutex by raw pointer (see - // `Mutex::lock_guard`), so `&mut self` stays freely usable while - // locked, and it drops before this fn returns — strictly before the - // task can be freed by the main thread. - let _guard = self.mutex.lock_guard(); - - // remember the state is atomic load it once, and store it again - let mut state = self.get_state(); - // old state should have more otherwise it's an HTTP-client bug - debug_assert!(state.has_more()); let is_done = !result.has_more; - let wait_until_done = self.update_state(async_http, &result, &mut state); - let should_enqueue = !wait_until_done || is_done; - bun_core::scoped_log!( - S3, - "state err: {} status_code: {} has_more: {} should_enqueue: {}", - state.request_error(), - state.status_code(), - state.has_more(), + // SAFETY: fn contract. Every access is a field access or a single call (type docs), and + // nothing touches the task after `unlock_raw` (`result`, dropped on return, is this + // thread's own). + unsafe { + (*this).mutex.lock(); + // remember the state is atomic load it once, and store it again + let mut state = (*this).get_state(); + // old state should have more otherwise it's an HTTP-client bug + debug_assert!(state.has_more()); + let wait_until_done = Self::update_state(this, async_http, &result, &mut state); + bun_core::scoped_log!( + S3, + "state err: {} status_code: {} has_more: {} should_enqueue: {}", + state.request_error(), + state.status_code(), + state.has_more(), + !wait_until_done || is_done + ); + result.body_into(&mut (*this).reported_response_buffer.list); + let should_enqueue = (!wait_until_done || is_done) + // dont report empty chunks if we have more data to read + && (is_done || !(*this).reported_response_buffer.list.is_empty()) + // if a task is already queued it will pick this state up; the exchange only + // happens when we would post, since a set flag without a post stalls the stream. + && (*this) + .has_schedule_callback + .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed) + .is_ok(); + Mutex::unlock_raw(&raw const (*this).mutex); should_enqueue - ); - - result.body_into(&mut self.reported_response_buffer.list); - if should_enqueue { - if self.reported_response_buffer.list.is_empty() && !is_done { - return false; - } - if let Err(has_schedule_callback) = self.has_schedule_callback.compare_exchange( - false, - true, - Ordering::Acquire, - Ordering::Relaxed, - ) { - if has_schedule_callback { - return false; - } - } - return true; } - false } /// this is the AsyncHTTP callback and is always called from the HTTPThread /// /// # Safety - /// `this` must be a live heap pointer produced by `Self::new`, valid for the duration of the - /// HTTP request; `mutex` serializes against `on_response`. `async_http` must be a valid - /// pointer to an initialised `AsyncHTTP` for the duration of this call. + /// `this` must be the heap task produced by `Self::new` for this request, live until this + /// thread publishes the request's final result (the JS thread frees it after that; see the + /// type docs). `async_http` must be a valid pointer to an initialised `AsyncHTTP` for the + /// duration of this call. pub(crate) fn http_callback( this: *mut Self, async_http: *mut AsyncHTTP<'static>, result: HTTPClientResult, ) { - // SAFETY: `this` is live for the duration of the HTTP request; HTTPThread holds the only - // concurrent reference and `mutex` serializes against `on_response`. `async_http` is the - // live HTTP-thread copy, non-null for the callback's duration. Borrows scoped to the call. + // SAFETY: fn contract; nothing below holds a reference into the task beyond a single + // field access or call, and the final result is published inside `process_http_callback`. let is_done = !result.has_more; // The final callback is where the HTTP thread hands the request back // (`embedded_work_finished` below, after `this` may have been freed). // SAFETY: `this` is live for the duration of the request. let done_handle = is_done.then(|| unsafe { (*this).loop_handle.clone() }); - // SAFETY: as above; the HTTP thread is the only one touching it here. - if unsafe { (*this).process_http_callback(&mut *async_http, result) } { + // SAFETY: as above; `async_http` is the HTTP thread's live copy of the request. + if unsafe { Self::process_http_callback(this, &mut *async_http, result) } { // we are always unlocked here and its safe to enqueue - // SAFETY: same exclusivity as above; `task` is the inline `concurrent_task` field of - // this heap request and the queue takes ownership of its `next` link. The VM waits - // for its S3 requests (embedded work) before closing its handle: always queued. + // SAFETY: `true` means no task was queued, so the JS thread cannot free the task + // before this post; `task` is the inline `concurrent_task` field of this heap + // request and the queue takes ownership of its `next` link. The VM waits for its S3 + // requests (embedded work) before closing its handle: always queued. unsafe { let task = core::ptr::NonNull::from( (*this).concurrent_task.from(this, AutoDeinit::ManualDeinit), @@ -314,21 +343,25 @@ impl S3HttpDownloadStreamingTask { pub(crate) unsafe fn release_at_shutdown(this: *mut ()) { let this = this.cast::(); // SAFETY: fn contract — nothing else touches the task now (the JS - // thread is waiting in the HTTP shutdown). + // thread is waiting in the HTTP shutdown), so unlike in + // `process_http_callback` nothing can free the task while this runs. + // The critical section still ends in `unlock_raw` so that every + // HTTP-thread use of `mutex` has the one shape that is sound even when + // the JS thread is waiting for the lock. The post below only happens + // when the exchange won, i.e. when no task is queued. unsafe { let handle = (*this).loop_handle.clone(); - let should_enqueue = { - let _guard = (*this).mutex.lock_guard(); - let mut state = (*this).get_state(); - state.set_has_more(false); - (*this).request_error = Some(bun_http::Error::Aborted); - state.set_request_error(1); - (*this).set_state(state); - (*this) - .has_schedule_callback - .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed) - .is_ok() - }; + (*this).mutex.lock(); + let mut state = (*this).get_state(); + state.set_has_more(false); + (*this).request_error = Some(bun_http::Error::Aborted); + state.set_request_error(1); + (*this).set_state(state); + let should_enqueue = (*this) + .has_schedule_callback + .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed) + .is_ok(); + Mutex::unlock_raw(&raw const (*this).mutex); if should_enqueue { let task = core::ptr::NonNull::from( (*this).concurrent_task.from(this, AutoDeinit::ManualDeinit), @@ -355,16 +388,10 @@ impl S3HttpDownloadStreamingTask { bun_http::http_thread().schedule_shutdown((*this).http.assume_init_ref()); } } - - fn release_portable(&mut self) { - // SAFETY: `http` is always initialised before the task is scheduled / dropped. - let http = unsafe { self.http.assume_init_mut() }; - http.clear_data(); - http.request_headers = Default::default(); - http.client.header_entries = Default::default(); - } } +/// Runs on the JS thread once the HTTP thread is done with the task (`on_response`), so unlike +/// the functions above this may take the whole task by reference. impl Drop for S3HttpDownloadStreamingTask { fn drop(&mut self) { // KeepAlive::unref now takes an aio EventLoopCtx; the JS-loop ctx is fetched @@ -375,7 +402,11 @@ impl Drop for S3HttpDownloadStreamingTask { )); // reported_response_buffer, headers, sign_result, range, proxy_url: // dropped automatically (Box/Vec-backed fields). - self.release_portable(); + // SAFETY: `http` is always initialised before the task is scheduled / dropped. + let http = unsafe { self.http.assume_init_mut() }; + http.clear_data(); + http.request_headers = Default::default(); + http.client.header_entries = Default::default(); } } diff --git a/src/threading/Mutex.rs b/src/threading/Mutex.rs index b036f6af3b81..e3c1aff46a83 100644 --- a/src/threading/Mutex.rs +++ b/src/threading/Mutex.rs @@ -76,10 +76,18 @@ impl Mutex { /// `*this`, and the futex wake that may follow it goes by address only /// ([`Futex::wake_raw`](crate::futex::wake_raw)). /// + /// The same goes for the caller's frames: such a section has to reach the + /// owner through a raw pointer rather than a `&self` / `&mut self` of it + /// (protected until that function returns, too), and cannot be a + /// [`lock_guard`](Self::lock_guard) section, whose guard unlocks through + /// `&Mutex`. `pub` for the HTTP thread's side of `S3HttpDownloadStreamingTask` + /// (src/runtime/webcore/s3/download_stream.rs), whose final unlock lets the + /// JS thread free the task. + /// /// # Safety /// `this` must point to a mutex this thread holds. It stays valid until the /// lock is released; from then on another thread may free it. - pub(crate) unsafe fn unlock_raw(this: *const Self) { + pub unsafe fn unlock_raw(this: *const Self) { // SAFETY: the lock is still held, so `*this` is live (fn contract). unsafe { Impl::unlock_raw(&raw const (*this).impl_) } } diff --git a/test/internal/source-lints/s3-download-task-raw-access.test.ts b/test/internal/source-lints/s3-download-task-raw-access.test.ts new file mode 100644 index 000000000000..8a11b925e1fd --- /dev/null +++ b/test/internal/source-lints/s3-download-task-raw-access.test.ts @@ -0,0 +1,245 @@ +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"; + +// `S3HttpDownloadStreamingTask` (src/runtime/webcore/s3/download_stream.rs) is +// one heap object used by two threads at once for its whole life: the HTTP +// thread records each chunk into it under the task's `mutex`, and the JS thread +// reports the chunks from `on_response` under the same mutex and frees the task +// once it has read `has_more == false`. Two consequences for how its code may +// be written, both enforced here: +// +// 1. No `&mut self` methods on the task: that receiver is banned in +// `impl S3HttpDownloadStreamingTask`. A reference argument is protected for +// the duration of the call under Rust's aliasing models (Tree Borrows, which +// `bun run rust:miri` uses, and Stacked Borrows), and a `&mut` one claims the +// whole object, atomics and mutex word included. The other thread writes +// those words whenever it tries to take the lock (and `on_stream_cancelled` +// writes `signal_store` from inside the chunk callback), and Miri reports the +// first such write during the call as UB ("this foreign write access would +// cause the protected tag to become Disabled"). On the HTTP thread the same +// `&mut self` would additionally still be protected when the final unlock +// lets the JS thread free the task. So every function that runs while both +// threads are live takes `this: *mut Self` and forms references one field or +// one call at a time; `&self` helpers (`get_state`, `set_state`) are fine +// because a shared reference leaves the interior-mutable words to the other +// thread and the remaining fields are only written by whoever holds the lock. +// `Drop` runs after the HTTP thread is done, so it is outside the impl block +// this applies to. +// +// 2. The mutex is left through `Mutex::unlock_raw`, never through a guard or a +// receiver: `.mutex.lock_guard()` in any spelling and `self.mutex.lock()` / +// `unlock()` / `try_lock()` are banned in src/runtime/webcore/s3/. On the final +// chunk the HTTP thread's unlock is what frees the task when a task is already +// queued (the JS thread is blocked in `on_response`'s `lock()` and frees the +// task as soon as it gets in), so the releasing store has to be the HTTP +// thread's last access to the task: `MutexGuard::drop` unlocks through a +// `&Mutex` into the task (and `Mutex::lock_guard` documents that the mutex +// must outlive the guard), and a section entered through `self.mutex` sits in +// a method whose receiver is live across the release. The shape both +// HTTP-thread sections use (`process_http_callback`, `release_at_shutdown`): +// +// (*this).mutex.lock(); +// ... +// Mutex::unlock_raw(&raw const (*this).mutex); +// +// Before the conversion this reported `report_progress`, `update_state`, +// `process_http_callback` and `release_portable` for (1) and the two +// `lock_guard()` sections, `process_http_callback` and `release_at_shutdown`, +// for (2). +// +// Not covered: an HTTP-thread section ending in `(*this).mutex.unlock()` is the +// same bug as (2) spelled differently and cannot be told apart by regex from +// `on_response`, whose `unlock()` runs on the JS thread ahead of that thread's +// own free; convert it on sight. Sibling guards for other spellings of "the +// callee outlives, or frees, what a reference argument covers": +// self-receiver-reclaim.test.ts, unsound-erased-box.test.ts. + +const root = path.resolve(import.meta.dir, "..", "..", ".."); +const SCOPE = "src/runtime/webcore/s3/"; +const TASK_FILE = `${SCOPE}download_stream.rs`; +const TASK_IMPL_HEADER = "impl S3HttpDownloadStreamingTask {"; +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)); +})(); + +// (1): a `&mut self` receiver, in either rustfmt layout (`fn f(&mut self, ..)` +// or `fn f(\n &mut self,\n`). +const MUT_SELF_RECEIVER = /\bfn\s+(\w+)\s*(?:<[^>]*>)?\s*\(\s*&\s*(?:'\w+\s+)?mut\s+self\b/g; + +// (2): the two spellings of a critical section that is not the raw shape. +const BANNED_MUTEX_USE = new RegExp( + [ + // `.mutex.lock_guard(..)`: `self.mutex`, `(*this).mutex`, `task.mutex`, + // including rustfmt's one-segment-per-line wrapping of the chain. + String.raw`\.\s*mutex\s*\.\s*lock_guard\s*\(`, + // Entering or leaving the section through the receiver. + String.raw`\bself\s*\.\s*mutex\s*\.\s*(?:try_)?(?:un)?lock\s*\(`, + ].join("|"), + "g", +); + +// What keeps the bans from passing vacuously: the task file still has its +// inherent impl block, still has a `bun_threading::Mutex` field called `mutex` +// (a rename would otherwise blind the regexes above) and still locks it. +const MUTEX_FIELD = /\bmutex\s*:\s*(?:[\w:]+::)?Mutex\b/g; +const LOCK = /\.\s*mutex\s*\.\s*lock\s*\(/g; + +// Strip full-line comments so the comments describing this hazard don't +// count. `[ \t]*`, not `\s*`: `\s` crosses newlines and would swallow blank +// lines, shifting the reported line numbers. +function stripComments(source: string): string { + return source.replace(/^[ \t]*\/\/.*$/gm, ""); +} + +function lineOf(text: string, index: number): number { + return text.slice(0, index).split("\n").length; +} + +// The inherent impl block: from its header to the next `}` at column 0 +// (rustfmt closes every item there), as [start, end) offsets into `stripped`. +function taskImplRange(stripped: string): [number, number] | null { + const start = stripped.indexOf(TASK_IMPL_HEADER); + if (start < 0) return null; + const end = stripped.indexOf("\n}", start); + return end < 0 ? null : [start, end]; +} + +function mutSelfMethods(stripped: string, range: [number, number]): string[] { + const hits: string[] = []; + for (const m of stripped.matchAll(MUT_SELF_RECEIVER)) { + const at = m.index ?? 0; + if (at >= range[0] && at < range[1]) hits.push(`${TASK_FILE}:${lineOf(stripped, at)}: ${m[1]}`); + } + return hits; +} + +const receiverOffenders: string[] = []; +const mutexOffenders: string[] = []; +const scanned: string[] = []; +let taskImplFound = false; +let mutexFields = 0; +let locks = 0; +for (const abs of rustSources) { + const source = path.relative(root, abs).replaceAll(path.sep, "/"); + if (!source.startsWith(SCOPE)) continue; + if (path.relative(root, realpathSync(abs)).replaceAll(path.sep, "/") !== source) continue; + if (tracked !== null && !tracked.has(source)) continue; + scanned.push(source); + const stripped = stripComments(await file(abs).text()); + mutexFields += [...stripped.matchAll(MUTEX_FIELD)].length; + locks += [...stripped.matchAll(LOCK)].length; + for (const m of stripped.matchAll(BANNED_MUTEX_USE)) { + mutexOffenders.push(`${source}:${lineOf(stripped, m.index ?? 0)}: ${m[0].replace(/\s+/g, " ")}`); + } + if (source === TASK_FILE) { + const range = taskImplRange(stripped); + if (range !== null) { + taskImplFound = true; + receiverOffenders.push(...mutSelfMethods(stripped, range)); + } + } +} + +function matchesMutexBan(snippet: string): boolean { + BANNED_MUTEX_USE.lastIndex = 0; + return BANNED_MUTEX_USE.test(snippet); +} + +test("scans the task it is about", () => { + expect(scanned).toContain(TASK_FILE); + expect(taskImplFound).toBe(true); + expect(mutexFields).toBeGreaterThan(0); + expect(locks).toBeGreaterThan(0); +}); + +test("the receiver pattern sees `&mut self` methods inside the impl block only", () => { + const sample = stripComments( + [ + "impl State {", + " fn set_has_more(&mut self, v: bool) {}", + "}", + "", + TASK_IMPL_HEADER, + " pub(crate) fn get_state(&self) -> State { todo!() }", + " // `report_progress`, as it was.", + " fn report_progress(&mut self, state: State) {}", + " fn update_state(", + " &mut self,", + " state: &mut State,", + " ) -> bool { todo!() }", + " fn with_lifetime<'a>(&'a mut self) {}", + " unsafe fn process_http_callback(this: *mut Self) -> bool { todo!() }", + " fn takes_another(&mut other: &mut u32) {}", + "}", + "", + "impl Drop for S3HttpDownloadStreamingTask {", + " fn drop(&mut self) {}", + "}", + ].join("\n"), + ); + const range = taskImplRange(sample); + expect(range).not.toBeNull(); + expect(mutSelfMethods(sample, range!).map(hit => hit.split(": ")[1])).toEqual([ + "report_progress", + "update_state", + "with_lifetime", + ]); +}); + +test("the mutex pattern recognizes the spellings it claims to", () => { + const banned = [ + // `process_http_callback(&mut self, ..)`, as it was. + "let _guard = self.mutex.lock_guard();", + // `release_at_shutdown`, as it was. + "let _guard = (*this).mutex.lock_guard();", + "let guard = task.mutex.lock_guard();", + "drop(self.mutex.lock_guard());", + "self.mutex.lock();", + "self.mutex.unlock();", + "if self.mutex.try_lock() {", + // rustfmt-wrapped chains. + "let _guard = (*this)\n .mutex\n .lock_guard();", + "self\n .mutex\n .unlock();", + ]; + const allowed = [ + // The required shape. + "(*this).mutex.lock();", + "Mutex::unlock_raw(&raw const (*this).mutex);", + "bun_threading::Mutex::unlock_raw(&raw const (*this).mutex);", + // `on_response` (JS thread; this thread frees the task itself, after the unlock). + "unsafe { (*this).mutex.lock() };", + "(*this_ptr).mutex.unlock();", + // A debug assertion neither enters nor leaves the section. + "debug_assert!(self.mutex.is_held_by_current_thread());", + // Declaring and initialising the field. + "pub(crate) mutex: Mutex,", + "mutex: Default::default(),", + // Guards on some other object's lock (`.mutex` is the field this lint is about). + "let _guard = self.queued_writes_lock.lock_guard();", + "let _guard = other.state_mutex.lock_guard();", + ]; + expect(banned.filter(s => !matchesMutexBan(s))).toEqual([]); + expect(allowed.filter(matchesMutexBan)).toEqual([]); +}); + +test("the streaming task has no `&mut self` methods", () => { + expect(receiverOffenders).toEqual([]); +}); + +test("S3 task mutexes are taken through the raw pointer and never held by a guard", () => { + expect(mutexOffenders).toEqual([]); +});