test(history): count worker parks, hold saturation blocks until released (#352)

wait_for_worker_pass now counts the idle worker's parks on its wake
event (two parks from now = one whole pass begun after the call)
instead of assuming a fixed number of clock reads per pass, which a
future clock read would have made vacuous. The real-clock bounded
close keeps a loose upper bound (0.02 s + NON_BLOCKING_CEILING).

The two saturation tests (history and output writer) blocked the
worker with release.wait(3) while they queued 1023 more items. On a
loaded runner the fill took longer than 3 s, the worker unblocked on
its own and drained part of the queue (pending 868..1017 != 1024),
and cleanup then reported a leaked worker. The block now holds until
the test releases it; 60 s is only a hang guard, and cleanup always
sets the release. Under one core with 8 busy loops: 8/8 runs failed
before, 6/6 pass after.

tests/test_execution_history_writer.py

@@ -31,16 +31,20 @@ from utils.execution_history_writer import HistoryWriter
 # 0.19 s on macOS py3.13 (PR #351).
 NON_BLOCKING_CEILING = 1.0
 
+# Saturation tests block the worker while they queue 1023 more items. The
+# block must hold until the test releases it (#346/#352): a 3 s timer let a
+# loaded runner unblock the worker mid-fill and drain it. This is only a hang
+# guard; the test, or cleanup, always sets the release.
+SATURATION_HOLD_SECONDS = 60
+
 
 class ManualClock:
     def __init__(self):
         self.value = 0.0
-        self.reads = 0
         self.lock = threading.Lock()
 
     def __call__(self):
         with self.lock:
-            self.reads += 1
             return self.value
 
     def advance(self, seconds):
@@ -86,16 +90,25 @@ class HistoryWriterTests(unittest.TestCase):
             threading.Event().wait(0.005)
         self.fail("finite deadline expired waiting for worker behavior")
 
-    def wait_for_worker_pass(self):
-        """Wait until the idle worker has run one whole loop pass on the clock.
+    def wait_for_worker_pass(self, writer):
+        """Wait until the idle worker has run one whole loop pass from now.
 
-        An idle pass reads the clock for the probe decision and again before
-        it parks, so three reads from now include one complete pass: whatever
-        the worker decides at the current clock value has happened (state,
-        not a wall-clock pause that a loaded runner can outlast).
+        A worker with no accepted work ends every pass by parking on its wake
+        event. The first park counted may end a pass already under way; the
+        second ends a pass that began after this call, so whatever the worker
+        decides at the current clock value has happened (state, not a
+        wall-clock pause that a loaded runner can outlast).
         """
-        reads = self.clock.reads
-        self.wait_for(lambda: self.clock.reads >= reads + 3)
+        wake = writer._wake
+        park = wake.wait
+        parks = []
+
+        def counted(timeout=None):
+            parks.append(timeout)
+            return park(timeout)
+
+        with patch.object(wake, "wait", counted):
+            self.wait_for(lambda: len(parks) >= 2)
 
     def writer(self, factory=None, ready=True):
         writer = HistoryWriter(factory or (lambda: self.db), clock=self.clock)
@@ -314,7 +327,7 @@ class HistoryWriterTests(unittest.TestCase):
 
         def write(command):
             entered.set()
-            release.wait(3)
+            release.wait(SATURATION_HOLD_SECONDS)
             return original(command)
 
         with patch.object(self.db, "write_execution_start", write):
@@ -419,6 +432,10 @@ class HistoryWriterTests(unittest.TestCase):
                 # Manual clock: only the patched join/sleep advance it, so this
                 # exact bound proves early joins do not spin forever.
                 self.assertLess(elapsed, 0.2)
+            else:
+                # Real clock: a loaded runner may run late, but close() must
+                # not wait anywhere near the 3 s the blocked write holds.
+                self.assertLess(elapsed, 0.02 + NON_BLOCKING_CEILING)
             # Bounded on the real clock too: close() gave up while the
             # in-flight write was still blocked, instead of waiting it out.
             self.assertFalse(left.is_set())
@@ -796,7 +813,7 @@ class HistoryWriterTests(unittest.TestCase):
             self.wait_for(lambda: not writer.status().accepting)
             self.assertFalse(writer.submit(fresh))
             self.clock.advance(4.99)
-            self.wait_for_worker_pass()
+            self.wait_for_worker_pass(writer)
             self.assertEqual(probes, [0])
             self.clock.advance(0.01)
             self.wait_for(lambda: writer.status().accepting)
@@ -819,12 +836,12 @@ class HistoryWriterTests(unittest.TestCase):
         writer = self.writer(factory, ready=False)
         self.wait_for(lambda: len(calls) == 1)
         self.clock.advance(4.99)
-        self.wait_for_worker_pass()
+        self.wait_for_worker_pass(writer)
         self.assertEqual(calls, [0])
         self.clock.advance(0.01)
         self.wait_for(lambda: len(calls) == 2)
         self.clock.advance(4.99)
-        self.wait_for_worker_pass()
+        self.wait_for_worker_pass(writer)
         self.assertEqual(calls, [0, 5])
         self.clock.advance(0.01)
         self.wait_for(lambda: writer.status().accepting)
@@ -878,7 +895,7 @@ class HistoryWriterTests(unittest.TestCase):
                 )
             writer.invalidate()
             release.set()
-            self.wait_for_worker_pass()
+            self.wait_for_worker_pass(writer)
             self.assertFalse(writer.status().accepting)
             self.assertIsNone(writer.status().epoch)
             self.assertFalse(writer.submit(self.start_command(epoch=stale_epoch)))

tests/test_execution_output_writer.py

@@ -28,6 +28,13 @@ except ImportError:
 # Upper bound a blocked fake write waits for its release. Tests prove ordering
 # with events, so this only caps a failing run and never decides a pass.
 BLOCK_SECONDS = 10
+
+# Saturation tests block the worker while they queue 1023 more items. The
+# block must hold until the test releases it (#346/#352): a 3 s timer let a
+# loaded runner unblock the worker mid-fill and drain it. This is only a hang
+# guard; the test, or cleanup, always sets the release.
+SATURATION_HOLD_SECONDS = 60
+
 # A restore is a full verify/migrate/backup/replace cycle and is bound by the
 # disk: a busy Windows runner took over 10 s (merge queue, #382). Waits that
 # span a restore use this ceiling, which stays under the runner's 120 s hang
@@ -591,7 +598,7 @@ class OutputWriterTests(fixtures.unittest.TestCase):
 
         def write(command):
             entered.set()
-            release.wait(3)
+            release.wait(SATURATION_HOLD_SECONDS)
             return original(command)
 
         with patch.object(self.db, "write_execution_start", write):