fix(js): each unit-test thread counts its own document script workers

Parallel unit tests in one process shared the process-wide count of live
document script workers, so unrelated tests together reached the 16-worker
cap and failed at random with "document script worker limit reached".
Under cfg(test) each test thread now has its own count against the same cap;
a permit releases the count it was taken from. The production cap and its
process-wide count are unchanged.
This commit is contained in:
mtlcl410 committed 2026-11-12 07:04:56 +08:00
1 parent 8c99ec4e08
commit 795bc1144b
1 file changed
+26 -8
+26 -8
View File
@@ -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<Self, String> {
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));
}