fix(test): stop the spawn-heavy tooling files contending for one runner's vCPUs (BACKLOG #1304)

tests/test_coord_lock.py

@@ -27,6 +27,8 @@
 
 import pytest
 
+from tests._spawn_lock import spawn_burst
+
 LOCK = Path(__file__).resolve().parents[1] / "scripts" / "coord" / "lock.ps1"
 
 pytestmark = pytest.mark.skipif(
@@ -141,7 +143,13 @@ def test_eight_concurrent_claimants_never_hold_it_at_once(repo: Path, tmp_path:
     barrier = tmp_path / "barrier"
     barrier.mkdir()
 
-    with ThreadPoolExecutor(max_workers=8) as pool:
+    # Eight concurrent pwsh, and the winner holds for 8 s -- the same shape as the storm BACKLOG #1304
+    # is about, found by sweeping the tier rather than by CI evidence. Nothing about the claimants
+    # changes; the lock only keeps single-launch tests off these vCPUs while they run.
+    with (
+        spawn_burst("coord_lock.mutex_under_load x8"),
+        ThreadPoolExecutor(max_workers=8) as pool,
+    ):
         claims = [
             pool.submit(
                 acquire,

tests/test_session_mail.py

@@ -45,6 +45,8 @@
 
 import pytest
 
+from tests._spawn_lock import spawn_burst
+
 ROOT = Path(__file__).resolve().parents[1]
 COORD = ROOT / "scripts" / "coord"
 HOOKS = ROOT / "scripts" / "hooks"
@@ -653,7 +655,12 @@ def one(i: int) -> subprocess.CompletedProcess[str]:
             capture_output=True, text=True, timeout=240, check=False,
         )  # fmt: skip
 
-    with concurrent.futures.ThreadPoolExecutor(max_workers=RACERS) as ex:
+    # RACERS stays 16 -- see tests/_spawn_lock.py for why cutting it is the wrong fix. The burst lock
+    # keeps single-launch tests off these vCPUs while the storm runs, without changing what runs here.
+    with (
+        spawn_burst(f"session_mail._race {mode} x{RACERS}"),
+        concurrent.futures.ThreadPoolExecutor(max_workers=RACERS) as ex,
+    ):
         procs = [f.result() for f in [ex.submit(one, i) for i in range(1, RACERS + 1)]]
     rows: list[dict[str, Any]] = []
     for p in procs:
@@ -948,7 +955,10 @@ def test_concurrent_drains_deliver_one_message_once(repo: Path, tmp_path: Path)
     def one(_: int) -> subprocess.CompletedProcess[str]:
         return run_drain(repo)
 
-    with concurrent.futures.ThreadPoolExecutor(max_workers=DRAINS) as ex:
+    with (
+        spawn_burst(f"session_mail.concurrent_drains x{DRAINS}"),
+        concurrent.futures.ThreadPoolExecutor(max_workers=DRAINS) as ex,
+    ):
         procs = [f.result() for f in [ex.submit(one, i) for i in range(DRAINS)]]
 
     texts = [injection(p) for p in procs]