diff --git a/gateway/delivery_ledger.py b/gateway/delivery_ledger.py index 3691e961d4c4..955e1d1d3e95 100644 --- a/gateway/delivery_ledger.py +++ b/gateway/delivery_ledger.py @@ -200,7 +200,11 @@ def _update_state(obligation_id: str, state: str, error: str = "") -> None: ) -def sweep_recoverable(now: Optional[float] = None) -> List[Dict[str, Any]]: +def sweep_recoverable( + now: Optional[float] = None, + *, + deliverable_platforms: Optional[set] = None, +) -> List[Dict[str, Any]]: """Claim undelivered rows owned by dead processes; return them for redelivery. @@ -209,6 +213,13 @@ def sweep_recoverable(now: Optional[float] = None) -> List[Dict[str, Any]]: double-claim (the UPDATE is guarded on the previous owner stamp). Rows over the attempts cap or older than the stale cutoff transition to 'abandoned' instead of being returned. + + ``deliverable_platforms`` (platform value strings) restricts claiming to + platforms the caller can actually send on this boot. ``attempts`` is the + redelivery budget, so it must only be spent on a real send: a platform + that failed to connect would otherwise burn one attempt per boot and hit + the cap having never been sent once. Rows for absent platforms are left + untouched for a later boot; the stale cutoff still bounds them. """ now = now if now is not None else time.time() pid, started = _owner_stamp() @@ -232,6 +243,13 @@ def sweep_recoverable(now: Optional[float] = None) -> List[Dict[str, Any]]: (now, oid), ) continue + if ( + deliverable_platforms is not None + and platform not in deliverable_platforms + ): + # No adapter for this platform this boot — the caller cannot + # send, so claiming would spend an attempt on a no-op. + continue cursor = conn.execute( """UPDATE delivery_obligations SET owner_pid=?, owner_started_at=?, attempts=attempts+1, diff --git a/gateway/run.py b/gateway/run.py index b071e4854f39..9bd6cf4cf862 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -6981,7 +6981,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if not ledger_enabled(): return 0 - claimed = await asyncio.to_thread(sweep_recoverable) + # Only claim rows we can actually send this boot: self.adapters + # holds a platform only after its connect() succeeded, and each + # claim spends one of the row's three redelivery attempts. + _deliverable = { + getattr(p, "value", str(p)) for p in self.adapters + } + claimed = await asyncio.to_thread( + sweep_recoverable, None, deliverable_platforms=_deliverable + ) except Exception: logger.debug("delivery ledger sweep failed", exc_info=True) return 0 diff --git a/tests/gateway/test_delivery_ledger.py b/tests/gateway/test_delivery_ledger.py index 855da10937f0..fcde3a07349e 100644 --- a/tests/gateway/test_delivery_ledger.py +++ b/tests/gateway/test_delivery_ledger.py @@ -281,3 +281,124 @@ class TestGatewayRedeliverySweep: n = await runner._redeliver_pending_obligations() assert n == 0 adapter.send.assert_not_awaited() + + +class TestAttemptsOnlySpentOnRealSends: + """``attempts`` is the redelivery budget — it must buy a send. + + ``self.adapters`` only holds a platform after its ``connect()`` succeeded, + and the sweep claimed every dead-owner row regardless. A platform that + failed to connect this boot therefore burned one attempt per boot while + the caller's ``adapter is None`` branch skipped it without sending — so + after MAX_ATTEMPTS boots the row abandoned having never been sent once, + losing exactly the response the ledger exists to guarantee. That failure + correlates with the crash that created the obligation: the network + trouble that killed the send tends to still be there on the next boot. + """ + + def test_absent_platform_does_not_burn_attempts(self): + _record(platform="telegram") + dl.mark_attempting("ob-1") + + for _ in range(dl.MAX_ATTEMPTS + 2): + _orphan("ob-1") + assert dl.sweep_recoverable(deliverable_platforms={"discord"}) == [] + + row = dl.debug_rows() + assert "abandoned" not in row + with dl._connect() as conn: + state, attempts = conn.execute( + "SELECT state, attempts FROM delivery_obligations " + "WHERE obligation_id=?", ("ob-1",), + ).fetchone() + assert attempts == 0, "an unsendable boot must not spend the budget" + assert state == "attempting" + + def test_row_still_delivers_once_its_platform_returns(self): + _record(platform="telegram") + for _ in range(dl.MAX_ATTEMPTS + 2): + _orphan("ob-1") + dl.sweep_recoverable(deliverable_platforms={"discord"}) + + _orphan("ob-1") + claimed = dl.sweep_recoverable(deliverable_platforms={"telegram"}) + assert len(claimed) == 1 + assert claimed[0]["attempts"] == 1 + + def test_present_platform_still_claims(self): + _record(platform="slack") + _orphan("ob-1") + claimed = dl.sweep_recoverable(deliverable_platforms={"slack"}) + assert len(claimed) == 1 + + def test_omitting_the_filter_claims_everything(self): + """Back-compat: existing callers pass no platform set.""" + _record(platform="telegram") + _orphan("ob-1") + assert len(dl.sweep_recoverable()) == 1 + + def test_stale_rows_abandon_even_when_undeliverable(self): + """The cutoff still bounds rows whose platform never returns.""" + _record(platform="telegram") + _orphan("ob-1") + future = time.time() + dl.STALE_AFTER_SECONDS + 10 + assert dl.sweep_recoverable( + now=future, deliverable_platforms={"discord"} + ) == [] + with dl._connect() as conn: + state = conn.execute( + "SELECT state FROM delivery_obligations WHERE obligation_id=?", + ("ob-1",), + ).fetchone()[0] + assert state == "abandoned" + + +class TestUnconnectedPlatformKeepsItsBudget: + """End-to-end through the real runner: boots where the platform failed to + connect must not consume the row's redelivery budget.""" + + @staticmethod + def _runner_without_slack(): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + runner.adapters = {} # slack failed to connect this boot + _store = MagicMock() + _store.clear_resume_pending = AsyncMock() + _store._store = None + runner.session_store = None + runner._async_session_store = _store + return runner + + @pytest.mark.asyncio + async def test_row_survives_boots_where_its_platform_is_down(self): + _record(platform="slack") + dl.mark_attempting("ob-1") + + for _ in range(dl.MAX_ATTEMPTS + 1): + _orphan("ob-1") + runner = self._runner_without_slack() + assert await runner._redeliver_pending_obligations() == 0 + + assert _row("ob-1")["state"] != "abandoned", ( + "the obligation was abandoned without a single send being attempted" + ) + assert _row("ob-1")["attempts"] == 0 + + @pytest.mark.asyncio + async def test_delivers_when_the_platform_comes_back(self): + from gateway.config import Platform + + _record(platform="slack") + for _ in range(dl.MAX_ATTEMPTS + 1): + _orphan("ob-1") + await self._runner_without_slack()._redeliver_pending_obligations() + + _orphan("ob-1") + adapter = MagicMock() + adapter.send = AsyncMock(return_value=MagicMock(success=True, error="")) + runner = self._runner_without_slack() + runner.adapters = {Platform.SLACK: adapter} + + assert await runner._redeliver_pending_obligations() == 1 + assert _row("ob-1")["state"] == "delivered"