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/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/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..e3c1aff46a83 100644 --- a/src/threading/Mutex.rs +++ b/src/threading/Mutex.rs @@ -64,7 +64,32 @@ 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)). + /// + /// 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 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 +210,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 +271,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 +311,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 +339,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 +422,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 +431,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 +457,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/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([]); +}); 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, +);