fix(gateway): recover pending event from delivery adapter
(cherry picked from commit fcf6973148b84b69ca65f23c3d60a1c369765106)
This commit is contained in:
@@ -4138,15 +4138,18 @@ class BasePlatformAdapter(ABC):
|
||||
# base path instead (returned False, or raised before storing it) and nothing is
|
||||
# queued, start this event.
|
||||
if session_key not in self._active_sessions:
|
||||
orphan = self._pending_messages.pop(session_key, None)
|
||||
if orphan is None:
|
||||
# The runner may have accepted the event into the delivery adapter's
|
||||
# pending slot while the turn released its guard. That slot is normally
|
||||
# drained by the finishing task, but the task can finish between the
|
||||
# handler's await and this ownership check. Start it here rather than
|
||||
# acknowledging an event with no owner (notably background wakes and
|
||||
# steer fallbacks).
|
||||
orphan = self.get_pending_message(session_key)
|
||||
# Busy admission may queue through the runner's delivery adapter. That can be a
|
||||
# replacement instance after reconnect, so recover from the adapter that owns the
|
||||
# event rather than reading this intake adapter's slot (and do it exactly once).
|
||||
delivery_adapter = self
|
||||
runner = getattr(self, "gateway_runner", None)
|
||||
if runner is not None:
|
||||
try:
|
||||
delivery_adapter = runner._delivery_adapter_for(event.source) or self
|
||||
except Exception:
|
||||
logger.debug("[%s] Delivery-adapter lookup failed during pending recovery",
|
||||
self.name, exc_info=True)
|
||||
orphan = delivery_adapter.get_pending_message(session_key)
|
||||
if orphan is not None:
|
||||
self._start_session_processing(orphan, session_key)
|
||||
elif not handled:
|
||||
|
||||
@@ -263,3 +263,52 @@ async def test_followup_stored_before_the_busy_handler_raised_runs_once(
|
||||
adapter.release_reply.set()
|
||||
held.release()
|
||||
await adapter.cancel_background_tasks()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_guard_release_recovers_pending_event_from_replacement_delivery_adapter():
|
||||
"""A reconnect handoff must drain the delivery adapter's slot, not the intake adapter twice."""
|
||||
intake = _Adapter()
|
||||
replacement = _Adapter()
|
||||
pending = _event("M2")
|
||||
incoming = _event("M1")
|
||||
session_key = intake._event_session_key(incoming)
|
||||
started = []
|
||||
get_calls = []
|
||||
|
||||
class _Runner:
|
||||
def _delivery_adapter_for(self, source):
|
||||
return replacement
|
||||
|
||||
intake.gateway_runner = _Runner()
|
||||
original_get = replacement.get_pending_message
|
||||
|
||||
def _get_pending(key):
|
||||
get_calls.append(key)
|
||||
return original_get(key)
|
||||
|
||||
replacement.get_pending_message = _get_pending
|
||||
|
||||
def _start(event, key):
|
||||
started.append((event, key))
|
||||
return True
|
||||
|
||||
intake._start_session_processing = _start
|
||||
intake._active_sessions[session_key] = asyncio.Event()
|
||||
|
||||
async def _busy_handler(event, key):
|
||||
replacement._pending_messages[key] = pending
|
||||
intake._active_sessions.pop(key, None)
|
||||
return True
|
||||
|
||||
async def _message_handler(event):
|
||||
return ""
|
||||
|
||||
intake.set_busy_session_handler(_busy_handler)
|
||||
intake.set_message_handler(_message_handler)
|
||||
await intake.handle_message(incoming)
|
||||
|
||||
assert started == [(pending, session_key)]
|
||||
assert get_calls == [session_key]
|
||||
assert session_key not in replacement._pending_messages
|
||||
assert session_key not in intake._pending_messages
|
||||
|
||||
Reference in New Issue
Block a user