From 371ee065fd708ef2e800406549d41443b78c3dad Mon Sep 17 00:00:00 2001 From: Drexuxux Date: Sun, 19 Jul 2026 18:44:34 +0300 Subject: [PATCH] fix(gateway): don't spend a redelivery attempt when the platform is down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The delivery ledger durably records a final response before the send so a crash between finalize and platform ACK can redeliver it on the next boot. attempts is that redelivery budget, capped at MAX_ATTEMPTS=3. sweep_recoverable() claims every dead-owner row and increments attempts before the caller knows whether it can send. self.adapters only holds a platform after its connect() succeeded, so when the platform failed to connect this boot _redeliver_pending_obligations() hits its "adapter is None" branch and continues WITHOUT sending — but the attempt is already spent. Three such boots and the row abandons, having never been sent once. That is the loss the ledger exists to prevent, and the trigger correlates with the crash that created the obligation: the network trouble that killed the send tends to still be there on the next boot. Worse, the message stays lost — once abandoned it is never retried even after the platform recovers. Reproduced against the real runner with an unconnected adapter: boot 1: claimed=1 state='attempting' attempts=1 (0 sends attempted) boot 2: claimed=1 state='attempting' attempts=2 (0 sends attempted) boot 3: claimed=1 state='attempting' attempts=3 (0 sends attempted) boot 4: claimed=0 state='abandoned' attempts=3 (0 sends attempted) Let the caller declare which platforms it can send on, and skip claiming rows for the others. attempts then only ever buys a real send. Rows for a platform that never returns are still bounded by the stale cutoff, so nothing accumulates. The parameter is keyword-only and optional — omitting it keeps the previous claim-everything behaviour for other callers. --- gateway/delivery_ledger.py | 20 ++++- gateway/run.py | 10 ++- tests/gateway/test_delivery_ledger.py | 121 ++++++++++++++++++++++++++ 3 files changed, 149 insertions(+), 2 deletions(-) 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"