From 506cc14def0a4bba496c17a2cb220579f68f4a73 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sun, 27 Sep 2026 12:14:13 +0530 Subject: [PATCH] fix(gateway): back off busy re-queued events inside the drain task The runner busy-demotion puts the event back into the adapter pending slot and returns None; the in-band drain re-dispatched it at once, ~400 times/s for the whole busy window (typing churn, log flood, platform connection storm). Count consecutive unanswered re-queues per session on the adapter (not by event identity, so a pre_gateway_dispatch rewrite via dataclasses.replace is still caught), reset when a handler returns a response. First re-queue stays immediate (restart auto-resume self-bounce); later ones back off 0.25s..5s. The delay is slept inside the new drain task before its processing try/finally, so cancelling during the back-off cannot reach _finish_session_task late-arrival respawn (no concurrent handler, no untracked task on shutdown). Fixes #123229 Co-authored-by: ahisblessed --- gateway/platforms/base.py | 46 ++++++++++++++++++--- tests/gateway/test_busy_requeue_hot_loop.py | 14 ++++++- 2 files changed, 53 insertions(+), 7 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 6c544cf9e6..76c85ce623 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -1925,6 +1925,9 @@ class BasePlatformAdapter(ABC): # could drop a newer guard. self._active_sessions: Dict[str, asyncio.Event] = {} self._pending_messages: Dict[str, MessageEvent] = {} + # Consecutive in-band drains per session whose handler returned nothing while an event sat + # pending (the runner's busy-demotion re-queue); drives the drain back-off (#123229). + self._requeue_counts: Dict[str, int] = {} self._pending_text_batches: Dict[str, MessageEvent] = {} self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {} self._session_tasks: Dict[str, asyncio.Task] = {} @@ -4473,6 +4476,8 @@ class BasePlatformAdapter(ABC): await self._run_processing_hook("on_processing_start", event) event._turn_marker_handoff = self.gateway_runner is not None # it can release the marker response = await self._message_handler(event) + if response is not None: + self._requeue_counts.pop(session_key, None) # the handler consumed its event # A muted diagnostic wake ran for the session; its reply is not presented. The # policy read binds the routed profile; delivery itself stays in the launch scope. with self._media_delivery_scope(event.source): @@ -4548,7 +4553,8 @@ class BasePlatformAdapter(ABC): logger.debug("[%s] Processing queued follow-up message", self.name) self._clear_session_guard(session_key) await self._stop_typing_refresh(event.source.chat_id, typing_task, metadata=_thread_metadata) - self._spawn_drain_task(pending_event, session_key) + self._spawn_drain_task(pending_event, session_key, + delay=self._requeue_backoff_delay(session_key, response)) return # Drain task owns the session now. except asyncio.CancelledError: expected = asyncio.current_task() in self._expected_cancelled_tasks @@ -4577,14 +4583,44 @@ class BasePlatformAdapter(ABC): await self._flush_text_debounce_now(session_key) self._finish_session_task(session_key, interrupt_event) - def _spawn_drain_task(self, pending_event: MessageEvent, session_key: str) -> None: + _REQUEUE_BACKOFF_INITIAL_SECONDS = 0.25 + _REQUEUE_BACKOFF_MAX_SECONDS = 5.0 + + def _requeue_backoff_delay(self, session_key: str, response: Any) -> float: + """Delay before re-dispatching a follow-up the handler left pending without answering. + + The runner's busy-demotion puts the event back (possibly a rewritten copy, so this counts + per session, not by event identity) and returns None; re-dispatching at once hot-loops for + the whole busy window (#123229). The first re-queue stays immediate — restart auto-resume + relies on one self-bounce — then back off exponentially to a cap. Defers, never drops.""" + if response is not None: + return 0.0 + attempts = self._requeue_counts.get(session_key, 0) + self._requeue_counts[session_key] = attempts + 1 + if attempts == 0: + return 0.0 + delay = min(self._REQUEUE_BACKOFF_MAX_SECONDS, + self._REQUEUE_BACKOFF_INITIAL_SECONDS * 2 ** min(attempts - 1, 16)) + (logger.info if attempts == 1 else logger.debug)( + "[%s] Handler re-queued a pending event for %s again (session busy elsewhere); " + "backing off %.2fs", self.name, session_key, delay) + return delay + + def _spawn_drain_task(self, pending_event: MessageEvent, session_key: str, + delay: float = 0.0) -> None: """Hand the session to a fresh task for a queued follow-up — never recurse (chained follow-ups grew the C stack to SIGSEGV). Clearing (not deleting) the Event keeps the guard - live for concurrent inbound; ownership moves so stale-lock detection works.""" + live for concurrent inbound; ownership moves so stale-lock detection works. ``delay`` is + slept INSIDE the new owner task, before its processing try/finally, so a cancel during the + back-off just ends it — it can't hit ``_finish_session_task``'s late-arrival respawn.""" self._clear_session_guard(session_key) self._track_session_task( - session_key, - asyncio.create_task(self._process_message_background(pending_event, session_key))) + session_key, asyncio.create_task(self._drain_after(pending_event, session_key, delay))) + + async def _drain_after(self, pending_event: MessageEvent, session_key: str, delay: float) -> None: + if delay > 0: + await asyncio.sleep(delay) + await self._process_message_background(pending_event, session_key) def _clear_session_guard(self, session_key: str) -> None: """Clear (not delete) the session's interrupt Event so the guard stays live for inbound.""" diff --git a/tests/gateway/test_busy_requeue_hot_loop.py b/tests/gateway/test_busy_requeue_hot_loop.py index b25e050ba3..d4803f502d 100644 --- a/tests/gateway/test_busy_requeue_hot_loop.py +++ b/tests/gateway/test_busy_requeue_hot_loop.py @@ -17,6 +17,7 @@ router's IPS classified the burst as a DDoS from this host and blocked discord.c """ import asyncio +import dataclasses import time from datetime import datetime from types import SimpleNamespace @@ -111,10 +112,19 @@ def _runner_with_running_agent(adapter, *, compression_in_flight): @pytest.mark.asyncio -async def test_requeued_busy_event_does_not_hot_loop(): +@pytest.mark.parametrize("rewrite_hook", [False, True]) +async def test_requeued_busy_event_does_not_hot_loop(rewrite_hook): adapter = _Adapter() runner, agent, sk = _runner_with_running_agent(adapter, compression_in_flight=True) - adapter.set_message_handler(runner._handle_message) + + async def handler(event): + # A pre_gateway_dispatch "rewrite" hands the runner a dataclasses.replace copy, so the + # event it re-queues is a different object than the one the adapter dispatched. + if rewrite_hook: + event = dataclasses.replace(event, text=event.text) + return await runner._handle_message(event) + + adapter.set_message_handler(handler) # The new adapter holds no guard for the session (the reconnect replaced it mid-turn). assert sk not in adapter._active_sessions