diff --git a/crates/js/src/session.rs b/crates/js/src/session.rs index 1e01674df1..691f3ea7af 100644 --- a/crates/js/src/session.rs +++ b/crates/js/src/session.rs @@ -258,25 +258,43 @@ impl Realm { } } -#[cfg(not(target_arch = "wasm32"))] +#[cfg(all(not(target_arch = "wasm32"), not(test)))] static LIVE_WORKERS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); #[cfg(not(target_arch = "wasm32"))] const MAX_DOCUMENT_WORKERS: usize = if usize::BITS == 32 { 4 } else { 16 }; +/// The count of live document workers that the cap applies to: one per process. +#[cfg(all(not(target_arch = "wasm32"), not(test)))] +fn live_workers() -> &'static std::sync::atomic::AtomicUsize { + &LIVE_WORKERS +} +/// Unit tests run in parallel threads of one process, so one process-wide count lets +/// unrelated tests exhaust the cap together. Each test thread counts its own workers +/// instead, against the same cap; the leak is one counter per test thread. +#[cfg(all(not(target_arch = "wasm32"), test))] +fn live_workers() -> &'static std::sync::atomic::AtomicUsize { + thread_local! { + static LIVE_WORKERS: &'static std::sync::atomic::AtomicUsize = + Box::leak(Box::new(std::sync::atomic::AtomicUsize::new(0))); + } + LIVE_WORKERS.with(|count| *count) +} /// A reservation is released only when its worker exits, including detached timed-out workers. +/// It releases the count it was taken from, whichever thread drops it. #[cfg(not(target_arch = "wasm32"))] -struct WorkerPermit; +struct WorkerPermit(&'static std::sync::atomic::AtomicUsize); #[cfg(not(target_arch = "wasm32"))] impl WorkerPermit { /// Reserve bounded address space without evicting another document's retained state. fn acquire() -> Result { use std::sync::atomic::Ordering::SeqCst; - let mut live = LIVE_WORKERS.load(SeqCst); + let count = live_workers(); + let mut live = count.load(SeqCst); loop { if live >= MAX_DOCUMENT_WORKERS { return Err("document script worker limit reached; close a scripted document and retry".into()); } - match LIVE_WORKERS.compare_exchange(live, live.saturating_add(1), SeqCst, SeqCst) { - Ok(_) => return Ok(Self), + match count.compare_exchange(live, live.saturating_add(1), SeqCst, SeqCst) { + Ok(_) => return Ok(Self(count)), Err(current) => live = current, } } @@ -285,7 +303,7 @@ impl WorkerPermit { #[cfg(not(target_arch = "wasm32"))] impl Drop for WorkerPermit { fn drop(&mut self) { - LIVE_WORKERS.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + self.0.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); } } @@ -915,7 +933,7 @@ mod tests { assert_eq!(session.run(capture, &[]).result.as_deref(), Some("retained")); } - /// Exercise the actual process-wide cap in isolation, then recover a slot without resetting other documents. + /// Exercise the cap in isolation, then recover a slot without resetting other documents. #[cfg(not(target_arch = "wasm32"))] #[test] fn document_worker_capacity_is_bounded_and_reusable() { @@ -939,7 +957,7 @@ mod tests { assert!(extra.start(job("1")).unwrap_err().contains("worker limit")); drop(workers.pop()); let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); - while LIVE_WORKERS.load(std::sync::atomic::Ordering::SeqCst) == MAX_DOCUMENT_WORKERS { + while live_workers().load(std::sync::atomic::Ordering::SeqCst) == MAX_DOCUMENT_WORKERS { assert!(std::time::Instant::now() < deadline); std::thread::sleep(std::time::Duration::from_millis(5)); }