diff --git a/src/runtime/cli/run_command.rs b/src/runtime/cli/run_command.rs index eb970946ba9d..cda128219156 100644 --- a/src/runtime/cli/run_command.rs +++ b/src/runtime/cli/run_command.rs @@ -3096,7 +3096,10 @@ impl RemoteImageDownload { // to the channel. // SAFETY: `this` was passed as the callback ctx in `prefetch_remote_images`; // `async_http` is the worker-thread temporary whose `.real` points back at - // `this.async_http`. + // `this.async_http`. `prefetch_remote_images` reads one tick per download + // before it returns (freeing `this` and ending `done`'s storage), and this + // download's tick is the last statement below, so everything here runs + // while both are live, and the channel is live until that publish's unlock. unsafe { let this = &mut *this; let async_http = &mut *async_http; @@ -3110,7 +3113,10 @@ impl RemoteImageDownload { result.body_into(&mut this.response_buffer.list); // Channel payload is a placeholder tick — the main thread // walks `downloads[]` to read per-task state after N wakeups. - let _ = (*this.done).write_item(0); + // If this is the last tick, the main thread may free the channel + // as soon as it has taken it, which `(*this.done).write_item(0)` + // would still hold a reference into (see `Channel::write_item_raw`). + let _ = DoneChannel::write_item_raw(this.done, 0); } } } 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/channel.rs b/src/threading/channel.rs index 93c6b7d3cec8..5a66e228309e 100644 --- a/src/threading/channel.rs +++ b/src/threading/channel.rs @@ -51,8 +51,9 @@ impl Channel> { } } -// `T: Copy` because `LinearFifo::write`/`read` are slice-copy based. All -// in-tree channel payloads are POD; revisit if a non-`Copy` T appears. +// `T: Copy` because `read_items` assigns into slots that are still +// uninitialized (a destructor would run on garbage there). All in-tree +// channel payloads are POD; revisit if a non-`Copy` T appears. impl> Channel { fn with_buffer(buffer: LinearFifo) -> Self { Self { @@ -64,10 +65,74 @@ impl> Channel { } } + /// Publishes `item`, blocking while the buffer is full. Only for a channel + /// that something other than the matching [`read_item`](Self::read_item) + /// keeps alive past this call: `&self` asserts the channel's storage until + /// this returns, and the reader this call unblocks may return before then. + /// When that read returning is what lets the owner free the channel, use + /// [`write_item_raw`](Self::write_item_raw). pub fn write_item(&self, item: T) -> Result<(), ChannelError> { - self.write_all(core::slice::from_ref(&item)) + // SAFETY: the channel outlives this call (fn contract). + unsafe { Self::write_item_raw(self, item) } } + /// [`write_item`](Self::write_item) for a channel whose owner may free it + /// as soon as the matching `read_item` returns (`RemoteImageDownload` in + /// run_command.rs publishes into a channel on the reading thread's stack). + /// The reader takes the item under the mutex, so it cannot return before + /// this thread's unlock; the store inside that unlock which releases the + /// mutex is this thread's last access to the channel, and no frame between + /// here and that store holds a reference into the channel. + /// + /// # Safety + /// `this` must point to a live channel, and the channel must stay live + /// until this call's final unlock. No reader can take `item` before that + /// unlock, so an owner that frees the channel only after the `read_item` + /// that returned `item` has returned satisfies this. + pub unsafe fn write_item_raw(this: *const Self, item: T) -> Result<(), ChannelError> { + // SAFETY: `item` has not been published, so the channel is live (fn + // contract). + unsafe { (*this).mutex.lock() }; + // SAFETY: as above while it waits for space; once it has published + // `item`, the reader still has to acquire the mutex, which this thread + // holds until the unlock below. The `&Self` this forms is gone before + // that unlock. + let result = unsafe { (*this).write_item_locked(item) }; + // SAFETY: the mutex is still held, so the channel is still live. The + // releasing store inside is the last access to it; `mutex.unlock()` + // would keep a `&Mutex` alive past that store. + unsafe { Mutex::unlock_raw(&raw const (*this).mutex) }; + result + } + + /// The critical section of [`write_item_raw`](Self::write_item_raw). The + /// caller holds `mutex` on entry and gets it back on return; `putters.wait` + /// releases it only while parked. + fn write_item_locked(&self, item: T) -> Result<(), ChannelError> { + loop { + // `is_closed` is a `Cell`, so this is a fresh load after every wait. + if self.is_closed.get() { + return Err(ChannelError::Closed); + } + // SAFETY: the mutex is held, and this `&mut` is dead before the + // wait() below releases it. + let buffer = unsafe { &mut *self.buffer.get() }; + match buffer.write_item(item) { + Ok(()) => { + self.getters.signal(); + return Ok(()); + } + // A dynamic buffer only fails to grow on OOM; a static one only + // fails while full. + Err(err) if B::DYNAMIC => return Err(err.into()), + Err(_) => self.putters.wait(&self.mutex), + } + } + } + + /// Blocks until an item is available. Once this returns, the + /// [`write_item_raw`](Self::write_item_raw) that published the item is done + /// with the channel, so a caller waiting for the last item may free it. pub fn read_item(&self) -> Result { let mut items: [MaybeUninit; 1] = [MaybeUninit::uninit()]; // SAFETY: see try_read_item. @@ -77,68 +142,19 @@ impl> Channel { Ok(unsafe { items[0].assume_init_read() }) } - pub(crate) fn write_all(&self, items: &[T]) -> Result<(), ChannelError> { - let n = self.write_items(items, true)?; - debug_assert!(n == items.len()); - Ok(()) - } - pub(crate) fn read_all(&self, items: &mut [T]) -> Result<(), ChannelError> { let n = self.read_items(items, true)?; debug_assert!(n == items.len()); Ok(()) } - fn write_items(&self, items: &[T], should_block: bool) -> Result { - let _guard = self.mutex.lock_guard(); - - let mut pushed: usize = 0; - while pushed < items.len() { - // Re-derive the `&mut buffer` each iteration: `Condition::wait` - // below releases the mutex, so a long-lived `&mut buffer` held - // across wait() would alias another thread's `&mut` (UB). - // `is_closed` is a `Cell` so `.get()` is already a fresh load each - // iteration (cannot be hoisted past the interior-mutable wait). - let did_push = 'blk: { - if self.is_closed.get() { - return Err(ChannelError::Closed); - } - // SAFETY: mutex is held; this &mut does not live across wait(). - let buffer = unsafe { &mut *self.buffer.get() }; - match buffer.write(items) { - Ok(()) => {} - Err(err) => { - if B::DYNAMIC { - return Err(err.into()); - } - break 'blk false; - } - } - self.getters.signal(); - break 'blk true; - }; - - if did_push { - pushed += 1; - } else if should_block { - // wait() releases the mutex while parked, reacquires before - // returning. No long-lived UnsafeCell borrows are live here. - self.putters.wait(&self.mutex); - } else { - break; - } - } - - Ok(pushed) - } - fn read_items(&self, items: &mut [T], should_block: bool) -> Result { let _guard = self.mutex.lock_guard(); let mut popped: usize = 0; while popped < items.len() { - // See write_items: re-derive UnsafeCell refs each iteration so no - // borrow lives across `getters.wait()` (which releases the mutex). + // Re-derive the UnsafeCell refs each iteration so no borrow lives + // across `getters.wait()` (which releases the mutex). let new_item: Option = 'blk: { // SAFETY: mutex is held; this &mut does not live across wait(). let buffer = unsafe { &mut *self.buffer.get() }; @@ -168,3 +184,92 @@ impl> Channel { Ok(popped) } } + +#[cfg(test)] +mod tests { + use super::*; + + type OneSlot = Channel>; + + struct SendPtr(*const OneSlot); + // SAFETY: `Channel` is `Sync`; the writer only uses the pointer under + // `write_item_raw`'s contract, which the reader below upholds. + unsafe impl Send for SendPtr {} + + // The reader frees a channel as soon as `read_item` has handed it the last + // item (`RemoteImageDownload` in run_command.rs: the channel is a local of + // the function that reads it), so the writer must neither touch nor still + // hold a reference into the channel once its final unlock has let that + // read return. Under Miri (`bun run rust:miri`) the `Box` drop is rejected + // whenever a frame of the writer still holds a reference into the channel; + // natively this is a use-after-free race. + // + // Two items through a one-slot buffer make the writer wait for the reader + // on every channel, so the reader is normally parked in `read_item` when + // the publishing write happens and its free races that write's return on + // every channel, not only on a writer thread's first one (with one item per + // channel the writer runs ahead and the reader frees channels the writer + // left long ago). One writer thread serves a batch of channels because + // under Miri a spawn costs as much as a couple of channels. Miri's + // scheduler lets the free win about once per 100 channels for the `&self` + // shape; a run is 1024 channels. + #[test] + fn reader_may_free_a_channel_once_it_has_the_items() { + const CHANNELS: usize = 32; + #[cfg(miri)] + const BATCHES: usize = 32; + #[cfg(not(miri))] + const BATCHES: usize = 1_000; + + for _ in 0..BATCHES { + let channels: Vec<*const OneSlot> = (0..CHANNELS) + .map(|_| Box::into_raw(Box::new(OneSlot::init_static())).cast_const()) + .collect(); + let writer = { + let channels: Vec = channels.iter().map(|&c| SendPtr(c)).collect(); + std::thread::Builder::new() + .spawn(move || { + for SendPtr(channel) in channels { + // SAFETY: the reader frees this channel only after + // `read_item` has returned the second item, so the + // channel is live until each call's final unlock. + unsafe { + OneSlot::write_item_raw(channel, 1).unwrap(); + OneSlot::write_item_raw(channel, 2).unwrap(); + } + } + }) + .unwrap() + }; + for channel in channels { + // SAFETY: `channel` is a freshly boxed allocation this thread + // alone owns; the second `read_item` returning means the writer + // is done with it (the property under test). + let items = unsafe { + let items = [(*channel).read_item(), (*channel).read_item()]; + drop(Box::from_raw(channel.cast_mut())); + items + }; + assert_eq!(items, [Ok(1), Ok(2)]); + } + writer.join().unwrap(); + } + } + + // `write_item(&self)` is for a channel that outlives the call for some + // reason other than the matching read; here the scope joins the writer + // before the channel goes away. + #[test] + fn items_arrive_in_order_through_a_full_buffer() { + let channel = OneSlot::init_static(); + let received = std::thread::scope(|scope| { + scope.spawn(|| { + for item in 1..=3 { + channel.write_item(item).unwrap(); + } + }); + [(); 3].map(|()| channel.read_item().unwrap()) + }); + assert_eq!(received, [1, 2, 3]); + } +} diff --git a/test/cli/run/markdown-entrypoint.test.ts b/test/cli/run/markdown-entrypoint.test.ts index 653046cf8457..1dfec9450fcb 100644 --- a/test/cli/run/markdown-entrypoint.test.ts +++ b/test/cli/run/markdown-entrypoint.test.ts @@ -401,4 +401,46 @@ describe("bun ", () => { expect(secondExit).toBe(0); }, ); + + // Every download reports back over a channel with 256 slots + // (`DoneChannel` in src/runtime/cli/run_command.rs), so more images than + // that is the case where the HTTP thread's publish can find the channel full + // and has to wait for the main thread to drain it. + test.skipIf(!isPosix)("prefetches more remote images than the done channel has slots", async () => { + const images = 300; + const payload = Buffer.alloc(64, "z"); + const requested = new Set(); + await using server = Bun.serve({ + port: 0, + fetch(req) { + requested.add(new URL(req.url).pathname); + return new Response(payload, { headers: { "content-type": "image/png" } }); + }, + }); + const paths = Array.from({ length: images }, (_, i) => `/${i}.png`); + using dir = tempDir("md-remote-img-many-", { + "doc.md": paths.map(p => `![](http://127.0.0.1:${server.port}${p})`).join("\n\n") + "\n", + "tmp/.keep": "", + }); + const tmp = join(String(dir), "tmp"); + await using proc = Bun.spawn({ + cmd: [bunExe(), "./doc.md"], + env: { + ...bunEnv, + FORCE_COLOR: "1", + TERM: "xterm-kitty", + KITTY_WINDOW_ID: "1", + BUN_TMPDIR: tmp, + TMPDIR: tmp, + }, + cwd: String(dir), + terminal: { cols: 80, rows: 24, data() {} }, + }); + const exitCode = await proc.exited; + proc.terminal?.close(); + + expect([...requested].sort()).toEqual([...paths].sort()); + expect(readdirSync(tmp).filter(name => name.startsWith("bun-md-")).length).toBe(images); + expect(exitCode).toBe(0); + }); }); diff --git a/test/internal/threading-channel-miri.test.ts b/test/internal/threading-channel-miri.test.ts new file mode 100644 index 000000000000..6dc3be304aba --- /dev/null +++ b/test/internal/threading-channel-miri.test.ts @@ -0,0 +1,73 @@ +/** + * `bun_threading::Channel` (src/threading/channel.rs) must stay clean under + * `cargo miri test`. + * + * The property this guards is the one `RunCommand::prefetch_remote_images` + * (src/runtime/cli/run_command.rs) relies on: its done-channel is a local of + * the function that reads it, so the HTTP thread's publish of the last tick + * has to be finished with the channel by the time `read_item` returns. With + * the publish going through `write_item(&self)`, the `&self` argument still + * asserts the channel's storage while the reader is already free to return, + * and the crate's own model of that caller + * (`channel::tests::reader_may_free_a_channel_once_it_has_the_items`) is + * rejected at its `Box` drop under both Tree Borrows (pinned here, as in + * `bun run rust:miri`) and the default Stacked Borrows. `Channel::write_item_raw` + * holds no reference into the channel past the store that releases the reader. + * Natively the unfixed shape has nothing left to do after that store but + * return, so there is nothing a bun-level test can observe; miri itself is + * the discriminator. + * + * Only the channel tests run here: the crate's `wait_group` test has the same + * bug in `WaitGroup::finish` and is fixed separately. Once the whole crate is + * clean this filter can go and the crate can join `MIRI_CRATES` in + * scripts/rust-miri.ts. + * + * 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)( + "a Channel reader may free the channel as soon as read_item returns (Tree Borrows miri)", + async () => { + await using proc = Bun.spawn({ + cmd: [cargoBin!, "miri", "test", "--locked", "-p", "bun_threading", "--", "channel::"], + 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"); + // The filter matching nothing would also exit 0; the model has to have run. + expect(stdout).toContain("test channel::tests::reader_may_free_a_channel_once_it_has_the_items ... ok"); + expect(exitCode).toBe(0); + }, + // Compiles the crate's dependencies for miri, then interprets 1024 + // channel hand-offs: ~20s of interpretation on top of a warm build here, + // about a minute on a cold one. + 180_000, +);