fix(reaper): surface and quarantine poison rows instead of silent recirculation (#91, #107)

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,
     )
 
 

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)."""

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,

src/tollgate/workers/runner.py

@@ -20,13 +20,45 @@
 
 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``)."""
 
     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)

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).

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