From dda7ad47b439f563036b2628eb5faaa164142a52 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Tue, 27 Oct 2026 18:27:18 -0700 Subject: [PATCH] test(history): wait on the writer's state, not short wall-clock pauses (#352) Four 0.08 s pauses before asserting that the worker had not probed yet now wait until the idle worker has run a whole loop pass at the current manual clock value (three clock reads), so a slow runner cannot end the pause before the worker decides. A mutation that probes 20 ms early now fails two tests deterministically. The real-clock bounded close no longer asserts close(0.02) returned in under 0.2 s wall time; it asserts the in-flight write is still blocked and the worker alive, like the output writer test. The manual-clock bound stays exact. The last 0.1 s non-blocking submit check uses the 1 s NON_BLOCKING_CEILING from #351. --- tests/test_execution_history_writer.py | 37 +++++++++++++++++++++----- 1 file changed, 31 insertions(+), 6 deletions(-) diff --git a/tests/test_execution_history_writer.py b/tests/test_execution_history_writer.py index 1d93dcbd..c7ceb6a6 100644 --- a/tests/test_execution_history_writer.py +++ b/tests/test_execution_history_writer.py @@ -35,10 +35,12 @@ NON_BLOCKING_CEILING = 1.0 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): @@ -84,6 +86,17 @@ 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. + + 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). + """ + reads = self.clock.reads + self.wait_for(lambda: self.clock.reads >= reads + 3) + def writer(self, factory=None, ready=True): writer = HistoryWriter(factory or (lambda: self.db), clock=self.clock) self.writers.append(writer) @@ -348,9 +361,12 @@ class HistoryWriterTests(unittest.TestCase): first = self.sealed(writer) charge = first.lease.byte_count + left = threading.Event() + def write(*args, **kwargs): entered.set() release.wait(3) + left.set() return original(*args, **kwargs) with patch.object(self.db, "write_execution_contexts_for_finish", write): @@ -396,8 +412,17 @@ class HistoryWriterTests(unittest.TestCase): before = monotonic() self.assertFalse(writer.close(0.02)) elapsed = monotonic() - before + # close() waits until its own monotonic deadline, so the lower + # bound holds on any runner. self.assertGreaterEqual(elapsed, 0.02) - self.assertLess(elapsed, 0.2) + if early_join: + # 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) + # 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()) + self.assertTrue(writer.status().alive) self.assertEqual(pool.charged_bytes, charge) replacement = HistoryWriter(lambda: self.db, context_budget=pool) self.writers.append(replacement) @@ -771,7 +796,7 @@ class HistoryWriterTests(unittest.TestCase): self.wait_for(lambda: not writer.status().accepting) self.assertFalse(writer.submit(fresh)) self.clock.advance(4.99) - threading.Event().wait(0.08) + self.wait_for_worker_pass() self.assertEqual(probes, [0]) self.clock.advance(0.01) self.wait_for(lambda: writer.status().accepting) @@ -794,12 +819,12 @@ class HistoryWriterTests(unittest.TestCase): writer = self.writer(factory, ready=False) self.wait_for(lambda: len(calls) == 1) self.clock.advance(4.99) - threading.Event().wait(0.08) + self.wait_for_worker_pass() self.assertEqual(calls, [0]) self.clock.advance(0.01) self.wait_for(lambda: len(calls) == 2) self.clock.advance(4.99) - threading.Event().wait(0.08) + self.wait_for_worker_pass() self.assertEqual(calls, [0, 5]) self.clock.advance(0.01) self.wait_for(lambda: writer.status().accepting) @@ -853,7 +878,7 @@ class HistoryWriterTests(unittest.TestCase): ) writer.invalidate() release.set() - threading.Event().wait(0.08) + self.wait_for_worker_pass() self.assertFalse(writer.status().accepting) self.assertIsNone(writer.status().epoch) self.assertFalse(writer.submit(self.start_command(epoch=stale_epoch))) @@ -871,7 +896,7 @@ class HistoryWriterTests(unittest.TestCase): lock.execute("BEGIN IMMEDIATE") before = time.monotonic() self.assertTrue(writer.submit(command)) - self.assertLess(time.monotonic() - before, 0.1) + self.assertLess(time.monotonic() - before, NON_BLOCKING_CEILING) self.wait_for(lambda: not writer.status().accepting) lock.rollback() self.clock.advance(5)