diff --git a/src/tollgate/app.py b/src/tollgate/app.py index 51b0795..c22c570 100644 --- a/src/tollgate/app.py +++ b/src/tollgate/app.py @@ -130,6 +130,7 @@ def _reservation_reaper(engine: AsyncEngine, settings: Settings) -> ReservationR clock=SystemClock(), ids=Uuid7IdGenerator(), batch_size=settings.reaper_batch_size, + max_reap_attempts=settings.reaper_max_reap_attempts, ) diff --git a/src/tollgate/application/handlers/reap.py b/src/tollgate/application/handlers/reap.py index 6588e13..147b0eb 100644 --- a/src/tollgate/application/handlers/reap.py +++ b/src/tollgate/application/handlers/reap.py @@ -34,11 +34,27 @@ class ReapReport: class ReservationReaperHandler: """Releases held reservations past their TTL, one bounded transaction each (§5.4, §5.5).""" - def __init__(self, *, uow: UnitOfWork, clock: Clock, ids: IdGenerator, batch_size: int) -> None: + def __init__( + self, + *, + uow: UnitOfWork, + clock: Clock, + ids: IdGenerator, + batch_size: int, + max_reap_attempts: int = 5, + ) -> None: self._uow = uow self._clock = clock self._ids = ids self._batch_size = batch_size + self._max_reap_attempts = max_reap_attempts + # Cross-tick poison-row bookkeeping (#91). Attempt counts accumulate across ticks; a row + # that fails ``max_reap_attempts`` times is quarantined — excluded from every future claim + # this process — so it stops recirculating at the queue head and stranding its estimate + # unseen. This state is per-process (a restart re-attempts, a fresh chance if contention + # cleared); a durable dead-letter would need a persisted attempt column. + self._reap_attempts: dict[ReservationId, int] = {} + self._quarantined: set[ReservationId] = set() async def run_once(self) -> ReapReport: """Reap expired reservations, each in its own SKIP LOCKED tx, isolating per-item failures. @@ -49,14 +65,18 @@ async def run_once(self) -> ReapReport: atomically. A reap that raises is caught so one persistently failing reservation cannot abort the whole tick and starve everything behind it (#74): its id is excluded from the rest of this tick (its rolled-back status flip would otherwise put it back at the queue - head) and the reaper moves to the next candidate. A failure in the *claim* itself (no id - yet) is a datastore problem, not one poison row, so it propagates for the runner to handle. - The tick is bounded to ``batch_size`` attempts; the next tick continues. + head) and the reaper moves to the next candidate. A row that fails this way across + ``max_reap_attempts`` ticks is *quarantined* — excluded from every future claim and logged + as an error — so a permanent poison row surfaces instead of recirculating forever and + silently stranding its estimate (#91). A failure in the *claim* itself (no id yet) is a + datastore problem, not one poison row, so it propagates for the runner to handle. The tick + is bounded to ``batch_size`` attempts; the next tick continues. """ now = self._clock.now() reaped = 0 failed = 0 - skip: list[ReservationId] = [] + # Start from the durable quarantine set so poison rows never re-enter the candidate window. + skip: list[ReservationId] = list(self._quarantined) while reaped + failed < self._batch_size: reservation_id: ReservationId | None = None try: @@ -85,16 +105,36 @@ async def run_once(self) -> ReapReport: ) await tx.ledger.append(entries) reaped += 1 + self._reap_attempts.pop(reservation_id, None) # a clean reap clears its history except Exception: if reservation_id is None: raise # the claim failed, not a single reap; let the runner handle it - logger.exception( - "reservation reap failed for %s; skipping it this tick", reservation_id - ) + self._record_reap_failure(reservation_id) skip.append(reservation_id) failed += 1 return ReapReport(reaped=reaped, failed=failed) + def _record_reap_failure(self, reservation_id: ReservationId) -> None: + """Count a per-item reap failure and quarantine the row once it exhausts its attempts.""" + attempts = self._reap_attempts.get(reservation_id, 0) + 1 + if attempts >= self._max_reap_attempts: + self._quarantined.add(reservation_id) + self._reap_attempts.pop(reservation_id, None) + logger.error( + "reservation reap failed for %s %d times; quarantining it (its reserved estimate " + "is stranded until it is resolved) — investigate", + reservation_id, + attempts, + ) + else: + self._reap_attempts[reservation_id] = attempts + logger.exception( + "reservation reap failed for %s (attempt %d/%d); skipping it this tick", + reservation_id, + attempts, + self._max_reap_attempts, + ) + class IdempotencyReaperHandler: """Batch-deletes idempotency keys past their TTL, one bounded transaction each (§5.5).""" diff --git a/src/tollgate/config/settings.py b/src/tollgate/config/settings.py index 81baec6..58c2c8e 100644 --- a/src/tollgate/config/settings.py +++ b/src/tollgate/config/settings.py @@ -73,6 +73,15 @@ class Settings(BaseSettings): ge=1, description="Max reservations the reservation reaper reaps per tick (bounded work).", ) + reaper_max_reap_attempts: int = Field( + default=5, + ge=1, + description=( + "Consecutive failed reap attempts before the reservation reaper quarantines a poison " + "row — excluding it from future claims and logging an error — so it stops " + "recirculating at the queue head and stranding its estimate unseen (#91)." + ), + ) idempotency_ttl_hours: int = Field( default=24, ge=1, diff --git a/src/tollgate/workers/runner.py b/src/tollgate/workers/runner.py index f8d427a..88ba435 100644 --- a/src/tollgate/workers/runner.py +++ b/src/tollgate/workers/runner.py @@ -20,6 +20,20 @@ logger = logging.getLogger(__name__) +#: Cap the backoff exponent before it is applied. ``worker_max_consecutive_failures`` only enforces +#: ``ge=1`` (no upper bound), so an operator can set it high enough that ``2 ** (n - 1)`` overflows +#: when multiplied by the float base — before the ``min`` with the ceiling ever runs. 2**30 s is +#: already ~34 years, far above any sane ceiling, so clamping the exponent here is lossless (#107). +_MAX_BACKOFF_EXPONENT = 30 + + +class WorkerStalled(RuntimeError): + """A worker made no forward progress for too many consecutive ticks (#91). + + Distinct from a tick that *raised*: the ticks returned normally but reported failures without + reaping anything, so the loop escalates by raising this for the orchestrator to restart/alert. + """ + class SupportsRunOnce(Protocol): """One bounded, idempotent unit of polled work (a reaper's ``run_once``).""" @@ -27,6 +41,24 @@ class SupportsRunOnce(Protocol): async def run_once(self) -> object: ... +def _backoff_seconds(consecutive_failures: int, *, base: float, cap: float) -> float: + """Exponential backoff for the *n*-th consecutive failure, clamped to ``cap`` (#75, #107).""" + exponent = min(consecutive_failures - 1, _MAX_BACKOFF_EXPONENT) + return min(base * 2.0**exponent, cap) + + +def _reported_failure_without_progress(result: object) -> bool: + """Whether a structured tick result reaped nothing while hitting per-item failures (#91). + + Duck-typed so the loop stays generic and stdlib-only: a reservation-reaper ``ReapReport`` + exposes ``reaped``/``failed``, whereas the idempotency reaper returns a plain ``int`` (no such + fields) and is therefore never judged stalled. + """ + failed = getattr(result, "failed", 0) + reaped = getattr(result, "reaped", 0) + return isinstance(failed, int) and isinstance(reaped, int) and failed > 0 and reaped == 0 + + async def run_forever( tick: SupportsRunOnce, *, @@ -40,18 +72,19 @@ async def run_forever( """Poll ``tick.run_once`` every ``interval_seconds`` until ``stop`` is set (§5.5). Between ticks the loop waits the interval but wakes immediately when ``stop`` is set (graceful - shutdown). A tick that raises is logged and the loop continues after an exponential backoff - (``backoff_base_seconds`` doubling up to ``backoff_max_seconds``); a success resets the backoff. - After ``max_consecutive_failures`` in a row the loop re-raises, so the process exits non-zero - and the orchestrator restarts and alerts rather than a wedged worker looking healthy (#75). + shutdown). A tick counts as a failure when it *raises* or when it returns a structured result + that reaped nothing while reporting per-item failures — the latter is the reaper stuck behind + poison rows, which a normal return would otherwise hide from the escalation (#91). A failed + tick is logged and the loop continues after an exponential backoff (``backoff_base_seconds`` + doubling up to ``backoff_max_seconds``); any forward progress resets the counter. After + ``max_consecutive_failures`` in a row the loop exits non-zero (re-raising a raised tick, or + raising :class:`WorkerStalled`), so the orchestrator restarts and alerts rather than a wedged + worker looking healthy (#75). """ consecutive_failures = 0 while not stop.is_set(): try: result = await tick.run_once() - logger.info("%s tick complete: %r", name, result) - consecutive_failures = 0 - wait = interval_seconds except Exception: consecutive_failures += 1 logger.exception( @@ -64,6 +97,33 @@ async def run_forever( consecutive_failures, ) raise - wait = min(backoff_base_seconds * 2 ** (consecutive_failures - 1), backoff_max_seconds) + wait = _backoff_seconds( + consecutive_failures, base=backoff_base_seconds, cap=backoff_max_seconds + ) + else: + logger.info("%s tick complete: %r", name, result) + if _reported_failure_without_progress(result): + consecutive_failures += 1 + logger.error( + "%s reaped nothing but reported failures (%d in a row); backing off", + name, + consecutive_failures, + ) + if consecutive_failures >= max_consecutive_failures: + logger.error( + "%s made no progress for %d ticks in a row; exiting for the orchestrator " + "to restart", + name, + consecutive_failures, + ) + raise WorkerStalled( + f"{name} made no reaping progress for {consecutive_failures} ticks" + ) + wait = _backoff_seconds( + consecutive_failures, base=backoff_base_seconds, cap=backoff_max_seconds + ) + else: + consecutive_failures = 0 + wait = interval_seconds with contextlib.suppress(TimeoutError): await asyncio.wait_for(stop.wait(), timeout=wait) diff --git a/tests/unit/test_reap.py b/tests/unit/test_reap.py index 75ecd0f..5f5c150 100644 --- a/tests/unit/test_reap.py +++ b/tests/unit/test_reap.py @@ -316,6 +316,53 @@ async def test_reaper_isolates_a_poison_reservation_and_reaps_the_rest() -> None assert {e.reservation_id for e in ctx.ledger.appended} == {"r-good-1", "r-good-2"} +async def test_reaper_quarantines_a_persistent_poison_row_after_max_attempts( + caplog: pytest.LogCaptureFixture, +) -> None: + # A reservation whose reap fails every tick would otherwise recirculate at the queue head + # forever — its rolled-back status flip leaves it held with the oldest deadline — stranding its + # estimate silently (#91). After max_reap_attempts failed ticks the reaper quarantines it: + # excludes it from future claims (so it stops blocking and wasting attempts) and logs an error + # so a permanent poison row surfaces instead of recirculating unseen. + import logging + + poison = ReservationId("r-poison") + reservations = _StarvingReservations(poison=poison, healthy=[]) + handler = ReservationReaperHandler( + uow=_Uow(_Ctx(reservations, counter_store=_PoisonCounterStore())), + clock=_ClockAt(_NOW), + ids=_SeqIds(), + batch_size=100, + max_reap_attempts=3, + ) + # Three ticks each fail on the poison (attempts 1, 2, 3); the third crosses the threshold. + with caplog.at_level(logging.ERROR): + for _ in range(3): + report = await handler.run_once() + assert report.failed == 1 + assert any("quarantin" in r.message.lower() for r in caplog.records) + + # From now on the poison is excluded from the claim, so a tick finds nothing and does not fail. + reservations.claim_excludes.clear() + report = await handler.run_once() + assert report == ReapReport(reaped=0, failed=0) + assert reservations.claim_excludes[0] == [poison] # quarantined id excluded on the next claim + + +async def test_reaper_does_not_quarantine_healthy_reservations() -> None: + # A tick that reaps cleanly never accrues attempts, so healthy reservations are never excluded. + reservations = _FakeReservations([_stored("r-1"), _stored("r-2")], lines=(_USER_LINE,)) + handler = ReservationReaperHandler( + uow=_Uow(_Ctx(reservations)), + clock=_ClockAt(_NOW), + ids=_SeqIds(), + batch_size=100, + max_reap_attempts=3, + ) + report = await handler.run_once() + assert report == ReapReport(reaped=2, failed=0) + + async def test_reaper_propagates_a_claim_failure() -> None: # A failure in the claim itself (no reservation id yet) is a datastore problem, not one poison # row, so it propagates for the runner's backoff/escalation to handle (#74/#75). diff --git a/tests/unit/test_worker_runner.py b/tests/unit/test_worker_runner.py index d6a8c5a..726d1a7 100644 --- a/tests/unit/test_worker_runner.py +++ b/tests/unit/test_worker_runner.py @@ -3,10 +3,33 @@ from __future__ import annotations import asyncio +from dataclasses import dataclass import pytest -from tollgate.workers.runner import run_forever +from tollgate.workers.runner import WorkerStalled, _backoff_seconds, run_forever + + +@dataclass(frozen=True) +class _Report: + """Stand-in for a ReapReport-shaped tick result the runner inspects by duck type.""" + + reaped: int + failed: int = 0 + + +class _ReportingTick: + def __init__(self, stop: asyncio.Event, results: list[_Report]) -> None: + self._stop = stop + self._results = results + self.calls = 0 + + async def run_once(self) -> object: + result = self._results[self.calls] + self.calls += 1 + if self.calls >= len(self._results): + self._stop.set() + return result class _CountingTick: @@ -106,3 +129,61 @@ async def test_run_forever_resets_the_failure_counter_on_success() -> None: backoff_max_seconds=0, ) assert tick.calls == 6 # never three failures in a row, so it ran to the stop + + +async def test_run_forever_escalates_when_a_tick_reports_failures_without_progress() -> None: + # A per-item reap failure returns a normal report (it does not raise), so the #75 escalation + # would never fire and a poison row could recirculate forever while the worker looks healthy + # (#91). A tick that reaped nothing but hit failures made no forward progress: treat it like a + # raised tick so backoff/escalation surface the stall. + stop = asyncio.Event() + tick = _ReportingTick(stop, [_Report(reaped=0, failed=5)] * 10) + with pytest.raises(WorkerStalled): + await run_forever( + tick, + interval_seconds=0, + stop=stop, + name="test", + max_consecutive_failures=3, + backoff_base_seconds=0, + backoff_max_seconds=0, + ) + assert tick.calls == 3 # exits on the third stalled tick + + +async def test_run_forever_tolerates_failures_while_making_progress() -> None: + # As long as the reaper still clears healthy reservations (reaped > 0), per-item failures must + # not take the whole reaper down — progress resets the counter (#91). + stop = asyncio.Event() + tick = _ReportingTick(stop, [_Report(reaped=2, failed=1)] * 5) + await run_forever( + tick, + interval_seconds=0, + stop=stop, + name="test", + max_consecutive_failures=3, + backoff_base_seconds=0, + backoff_max_seconds=0, + ) + assert tick.calls == 5 # never escalates; runs to the stop + + +async def test_run_forever_ignores_failed_field_on_scalar_results() -> None: + # The idempotency reaper returns a plain int; it has no reaped/failed fields, so it is never + # judged stalled. + stop = asyncio.Event() + tick = _CountingTick(stop, stop_after=3) + await run_forever(tick, interval_seconds=0, stop=stop, name="test") + assert tick.calls == 3 + + +def test_backoff_seconds_does_not_overflow_for_a_large_failure_count() -> None: + # worker_max_consecutive_failures only enforces ge=1, so a large value must not make + # base * 2 ** (n - 1) raise OverflowError before it is clamped to the ceiling (#107). + assert _backoff_seconds(5000, base=1.0, cap=60.0) == 60.0 + + +def test_backoff_seconds_grows_then_clamps() -> None: + assert _backoff_seconds(1, base=1.0, cap=60.0) == 1.0 + assert _backoff_seconds(3, base=1.0, cap=60.0) == 4.0 + assert _backoff_seconds(10, base=1.0, cap=60.0) == 60.0 # clamped