From cfcd3d5b9f669085f6af0769ad6d11427f159caf Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Tue, 27 Oct 2026 20:03:28 -0700 Subject: [PATCH] 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 | 47 ++++++++++++++++++-------- tests/test_execution_output_writer.py | 9 ++++- 2 files changed, 40 insertions(+), 16 deletions(-) diff --git a/tests/test_execution_history_writer.py b/tests/test_execution_history_writer.py index c7ceb6a6..c0b11633 100644 --- a/tests/test_execution_history_writer.py +++ b/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))) diff --git a/tests/test_execution_output_writer.py b/tests/test_execution_output_writer.py index 06475d24..532b0dde 100644 --- a/tests/test_execution_output_writer.py +++ b/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):