From 0d632d9531199392ca066bc15d1faa1bdd524567 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 14 Aug 2026 04:05:50 +0000 Subject: [PATCH] threading: publish into a Channel without holding a reference into it past the release Channel::write_item(&self) kept &self (and, through the MutexGuard, &Mutex) alive until after the store that lets a blocked read_item return. prefetch_remote_images reads from a channel on its own stack and returns, ending the channel's storage, as soon as the last download's tick arrives, so the HTTP thread's publish of that tick could still be holding those references when the storage died. Add Channel::write_item_raw(this: *const Self, item), whose last access to the channel is the releasing store inside Mutex::unlock_raw; write_item delegates to it. RemoteImageDownload::on_done publishes through it as its last statement. Mutex::unlock_raw / Futex::wake_raw are the same hunks as in the WaitGroup change (#38330). The crate test models the caller (reader frees each channel as soon as it has its items) and is rejected by miri for the &self shape; threading-channel-miri.test.ts runs it under Tree Borrows. The markdown test covers more downloads than the channel has slots. --- src/runtime/cli/run_command.rs | 10 +- src/threading/Futex.rs | 61 ++++-- src/threading/Mutex.rs | 71 +++++-- src/threading/channel.rs | 213 ++++++++++++++----- test/cli/run/markdown-entrypoint.test.ts | 42 ++++ test/internal/threading-channel-miri.test.ts | 73 +++++++ 6 files changed, 381 insertions(+), 89 deletions(-) create mode 100644 test/internal/threading-channel-miri.test.ts 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, +);