From 5b36d64583bf9ed8c30b127d24b671f229f8abcb Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sun, 2 Aug 2026 23:52:05 +0530 Subject: [PATCH] test: raise blocking-probe timeouts for loaded CI runners CI slices failed the offload tests with 0.5s witness timeouts: on a loaded shared runner the event loop thread can take >0.5s to get scheduled even when NOT blocked, making the probe report a false positive. A genuinely blocked loop can never set the progress event at any timeout (the witness coroutine can't run at all), so 5s only absorbs scheduler flake without weakening the invariant. Mutation re-verified: reverting the offload still fails all 4 tests. --- gateway/platforms/base.py | 12 +++++++----- gateway/run.py | 7 ++++--- tests/gateway/test_delivery_ledger.py | 7 +++++-- tests/gateway/test_delivery_ledger_producer.py | 7 +++++-- 4 files changed, 21 insertions(+), 12 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 36741ae6d229d..c42b9160737d5 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -6066,13 +6066,14 @@ class BasePlatformAdapter(ABC): record_obligation, ) - if ledger_enabled(): + if await asyncio.to_thread(ledger_enabled): _obligation_id = compute_obligation_id( session_key, str(getattr(event, "message_id", "") or ""), text_content, ) - record_obligation( + await asyncio.to_thread( + record_obligation, obligation_id=_obligation_id, session_key=session_key, platform=str( @@ -6083,7 +6084,7 @@ class BasePlatformAdapter(ABC): thread_id=getattr(event.source, "thread_id", None), content=text_content, ) - mark_attempting(_obligation_id) + await asyncio.to_thread(mark_attempting, _obligation_id) except Exception: logger.debug("delivery ledger record failed", exc_info=True) _obligation_id = None @@ -6102,9 +6103,10 @@ class BasePlatformAdapter(ABC): ) if getattr(result, "success", False): - mark_delivered(_obligation_id) + await asyncio.to_thread(mark_delivered, _obligation_id) else: - mark_failed( + await asyncio.to_thread( + mark_failed, _obligation_id, str(getattr(result, "error", "") or ""), ) diff --git a/gateway/run.py b/gateway/run.py index f170f55ddab00..8c162f7dc9a46 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -10231,7 +10231,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew sweep_recoverable, ) - if not ledger_enabled(): + if not await asyncio.to_thread(ledger_enabled): return 0 # Only claim rows we can actually send this boot: self.adapters # holds a platform only after its connect() succeeded, and each @@ -10283,7 +10283,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew result = None try: if result is not None and getattr(result, "success", False): - mark_delivered(row["obligation_id"]) + await asyncio.to_thread(mark_delivered, row["obligation_id"]) redelivered += 1 logger.info( "Redelivered recovered final response to %s:%s " @@ -10292,7 +10292,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew row["obligation_id"], row["attempts"], ) else: - mark_failed( + await asyncio.to_thread( + mark_failed, row["obligation_id"], str(getattr(result, "error", "") or "send failed"), ) diff --git a/tests/gateway/test_delivery_ledger.py b/tests/gateway/test_delivery_ledger.py index 86584c30ff43f..f9421f7e6bab6 100644 --- a/tests/gateway/test_delivery_ledger.py +++ b/tests/gateway/test_delivery_ledger.py @@ -59,13 +59,16 @@ def _blocking_probe(): def _slow_ledger_call(*args, **kwargs): ledger_started.set() - if not event_loop_progressed.wait(timeout=0.5): + # Generous timeout: a genuinely blocked loop can never set the event + # (the witness coroutine cannot run), so a longer wait only guards + # against loaded-CI scheduling flake, not against missing the bug. + if not event_loop_progressed.wait(timeout=5.0): blocked_event_loop.append(True) async def _event_loop_witness(): import asyncio - deadline = asyncio.get_running_loop().time() + 1 + deadline = asyncio.get_running_loop().time() + 10 while not ledger_started.is_set(): if asyncio.get_running_loop().time() >= deadline: raise AssertionError("ledger call never started") diff --git a/tests/gateway/test_delivery_ledger_producer.py b/tests/gateway/test_delivery_ledger_producer.py index c02f3d345af5e..1071d9f36e072 100644 --- a/tests/gateway/test_delivery_ledger_producer.py +++ b/tests/gateway/test_delivery_ledger_producer.py @@ -74,11 +74,14 @@ def _blocking_probe(): def _slow_ledger_call(*args, **kwargs): ledger_started.set() - if not event_loop_progressed.wait(timeout=0.5): + # Generous timeout: a genuinely blocked loop can never set the event + # (the witness coroutine cannot run), so a longer wait only guards + # against loaded-CI scheduling flake, not against missing the bug. + if not event_loop_progressed.wait(timeout=5.0): blocked_event_loop.append(True) async def _event_loop_witness(): - deadline = asyncio.get_running_loop().time() + 1 + deadline = asyncio.get_running_loop().time() + 10 while not ledger_started.is_set(): if asyncio.get_running_loop().time() >= deadline: raise AssertionError("ledger call never started")