diff --git a/tests/history_waits.py b/tests/history_waits.py new file mode 100644 index 0000000..3e72ab9 --- /dev/null +++ b/tests/history_waits.py @@ -0,0 +1,36 @@ +"""Finite test waits for real SQLite batches without a throughput requirement.""" + +import threading +import time +from typing import Callable + +from utils.execution_history_writer import HistoryWriter + + +def drain_writer( + writer: HistoryWriter, *, clock: Callable[[], float] = time.monotonic +) -> None: + initial = current = writer.status() + started = last_progress = clock() + lowest_pending = initial.pending + while current.pending: + now = clock() + if current.pending < lowest_pending: + lowest_pending = current.pending + last_progress = now + reason = None + if not current.alive: + reason = "worker stopped" + elif now - started >= 30: + reason = "batch deadline" + elif now - last_progress >= 3: + reason = "worker stalled" + if reason is not None: + raise AssertionError( + f"history drain {reason}: elapsed={now - started:.3f}s, " + f"stalled={now - last_progress:.3f}s, " + f"initial={initial}, current={current}, " + f"charged_bytes={writer.context_budget.charged_bytes}" + ) + threading.Event().wait(0.005) + current = writer.status() diff --git a/tests/test_execution_history_hooks.py b/tests/test_execution_history_hooks.py index 840735e..ae790d2 100644 --- a/tests/test_execution_history_hooks.py +++ b/tests/test_execution_history_hooks.py @@ -16,6 +16,7 @@ from types import SimpleNamespace import unittest from unittest.mock import patch +from tests.history_waits import drain_writer from database import PromptDatabase from utils import usage_tracking from utils.execution_history_writer import HistoryWriter @@ -117,7 +118,7 @@ class HistoryHooksTests(unittest.TestCase): ] def drain(self, registration): - self.wait_for(lambda: registration.writer.status().pending == 0) + drain_writer(registration.writer, clock=time.monotonic) def test_package_history_initializes_when_monitor_and_usage_setup_fail(self): try: diff --git a/tests/test_execution_history_writer.py b/tests/test_execution_history_writer.py index babe30b..df1c5b2 100644 --- a/tests/test_execution_history_writer.py +++ b/tests/test_execution_history_writer.py @@ -13,6 +13,7 @@ import unittest from unittest.mock import patch from uuid import uuid4 +from tests.history_waits import drain_writer from database import PromptDatabase from utils.execution_history_events import ( LifecycleEvent, @@ -696,7 +697,7 @@ class HistoryWriterTests(unittest.TestCase): "prompt_manager.execution_history_writer", "WARNING" ) as logs: release.set() - self.wait_for(lambda: writer.status().pending == 0) + drain_writer(writer) self.assertTrue(writer.close(2)) self.assertEqual(len(logs.records), 1) diff --git a/tests/test_history_waits.py b/tests/test_history_waits.py new file mode 100644 index 0000000..1c4a7a3 --- /dev/null +++ b/tests/test_history_waits.py @@ -0,0 +1,90 @@ +"""Batch drain detects stalls without imposing a SQLite throughput target.""" + +from dataclasses import replace +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +from tests import test_execution_history_hooks as hooks +from utils.execution_history_writer import HistoryWriterStatus + + +class StepClock: + def __init__(self, step: float): + self.value = 0.0 + self.step = step + + def __call__(self) -> float: + self.value += self.step + return self.value + + +class ObservedWriter: + def __init__(self, pending: int, *, progress: bool = False, alive: bool = True): + self.pending = pending + self.progress = progress + self.alive = alive + self.context_budget = SimpleNamespace(charged_bytes=281) + + def status(self) -> HistoryWriterStatus: + status = HistoryWriterStatus("epoch", True, self.pending, 0, self.alive, 0) + if self.progress and self.pending: + self.pending -= 1 + return status + + +class HistoryDrainTests(unittest.TestCase): + wait_for = hooks.HistoryHooksTests.wait_for + + def drain(self, writer: ObservedWriter, clock: StepClock) -> None: + with patch.object(hooks, "time", SimpleNamespace(monotonic=clock)): + hooks.HistoryHooksTests.drain(self, SimpleNamespace(writer=writer)) + + def test_batch_with_progress_can_exceed_short_stall_deadline(self): + writer = ObservedWriter(128, progress=True) + clock = StepClock(0.04) + self.drain(writer, clock) + self.assertEqual(writer.pending, 0) + self.assertGreater(clock.value, 3) + self.assertLess(clock.value, 30) + + def test_no_progress_fails_with_worker_and_lease_diagnostics(self): + writer = ObservedWriter(52) + clock = StepClock(0.5) + with self.assertRaisesRegex( + AssertionError, r"worker stalled.*pending=52.*charged_bytes=281" + ): + self.drain(writer, clock) + self.assertLessEqual(clock.value, 4) + + def test_progress_cannot_extend_batch_past_hard_ceiling(self): + writer = ObservedWriter(1000, progress=True) + clock = StepClock(0.5) + with self.assertRaisesRegex(AssertionError, "batch deadline"): + self.drain(writer, clock) + self.assertGreater(writer.pending, 0) + self.assertGreaterEqual(clock.value, 30) + self.assertLessEqual(clock.value, 31) + + def test_dead_worker_with_pending_work_fails_immediately(self): + writer = ObservedWriter(1, alive=False) + clock = StepClock(0.5) + with self.assertRaisesRegex(AssertionError, r"worker stopped.*alive=False"): + self.drain(writer, clock) + self.assertLess(clock.value, 3) + + def test_oscillating_pending_is_not_new_progress(self): + writer = ObservedWriter(1) + original = writer.status + calls = 0 + + def oscillating() -> HistoryWriterStatus: + nonlocal calls + calls += 1 + return replace(original(), pending=1 + calls % 2) + + writer.status = oscillating + clock = StepClock(0.5) + with self.assertRaisesRegex(AssertionError, "worker stalled"): + self.drain(writer, clock) + self.assertLessEqual(clock.value, 5)