From caf8e2f214dd1d0742a3c30e9ba2f63374a04777 Mon Sep 17 00:00:00 2001 From: 87 <87degrees@87ui-Macmini.local> Date: Fri, 26 Jun 2026 15:36:34 +0900 Subject: [PATCH] fix(slack): heal wedged Socket Mode via ping/pong staleness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Socket Mode watchdog only reconnects when is_connected() returns False or the receiver task dies. When the underlying aiohttp ClientSession is closed (e.g. after a network blip), slack_sdk gets stuck retrying "Session is closed" while is_connected() can still report healthy and the receiver task stays alive — so the watchdog never fires and the process is alive but deaf to Slack indefinitely. Add a ping/pong staleness probe: Slack sends a ping roughly every ping_interval seconds even on an idle socket, so a stale/missing last_ping_pong_time (past a first-ping grace window) is a reliable signal the transport is wedged. The watchdog now also reconnects on staleness, which rebuilds the handler with a fresh session. Guards non-numeric attributes so a mocked/partial client never triggers a spurious reconnect. 7 new tests; full test_slack.py (216) green. --- plugins/platforms/slack/adapter.py | 47 ++++++++++++++++++ tests/gateway/test_slack.py | 80 ++++++++++++++++++++++++++++++ 2 files changed, 127 insertions(+) diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py index 61365835ca52..54dd79bb134d 100644 --- a/plugins/platforms/slack/adapter.py +++ b/plugins/platforms/slack/adapter.py @@ -693,6 +693,16 @@ class SlackAdapter(BasePlatformAdapter): self._socket_watchdog_task: Optional[asyncio.Task] = None self._socket_reconnect_lock = asyncio.Lock() self._socket_watchdog_interval_s = 15.0 + # Monotonic timestamp of the most recent Socket Mode handler (re)start, + # used to grant a grace window for the first ping/pong after connect. + self._socket_handler_started_monotonic: Optional[float] = None + # Reconnect when no ping/pong has arrived for this many multiples of the + # client's ping_interval. Slack pings roughly every ping_interval seconds + # even on an idle socket, so prolonged silence means a wedged transport. + self._socket_ping_stale_factor = 4 + # Allow at least this long after (re)connect before treating a missing + # first ping/pong as evidence of a wedged transport. + self._socket_first_ping_grace_s = 60.0 def _start_socket_mode_handler(self) -> None: """Start the Slack Socket Mode background task.""" @@ -706,6 +716,7 @@ class SlackAdapter(BasePlatformAdapter): task = asyncio.create_task(self._handler.start_async()) self._socket_mode_task = task + self._socket_handler_started_monotonic = time.monotonic() task.add_done_callback(self._on_socket_mode_task_done) async def _stop_socket_mode_handler(self) -> None: @@ -766,6 +777,37 @@ class SlackAdapter(BasePlatformAdapter): ) return None + def _socket_ping_pong_stale(self) -> bool: + """True when the Socket Mode transport shows no recent ping/pong. + + slack_sdk's Socket Mode client records ``last_ping_pong_time`` whenever + Slack's periodic ping arrives (roughly every ``ping_interval`` seconds, + even on an otherwise idle connection). When the underlying aiohttp + session is closed, the client gets stuck retrying ("Session is closed") + while ``is_connected()`` can still report healthy — so ping/pong + staleness is the reliable signal that the socket is wedged and the + handler must be rebuilt. Guards against non-numeric attributes so a + mocked/partial client never triggers a spurious reconnect. + """ + client = getattr(self._handler, "client", None) + if client is None: + return False + ping_interval = getattr(client, "ping_interval", None) + if not isinstance(ping_interval, (int, float)) or ping_interval <= 0: + return False + last = getattr(client, "last_ping_pong_time", None) + if last is None: + # No ping yet. Healthy right after (re)connect; only suspicious once + # the grace window elapses without ever seeing the first ping/pong. + started = self._socket_handler_started_monotonic + if started is None: + return False + grace = max(self._socket_first_ping_grace_s, ping_interval * 2) + return (time.monotonic() - started) > grace + if not isinstance(last, (int, float)): + return False + return (time.time() - last) > (ping_interval * self._socket_ping_stale_factor) + async def _restart_socket_mode(self, reason: str) -> None: """Reconnect Socket Mode without rebuilding adapter state.""" if not self._running: @@ -810,6 +852,11 @@ class SlackAdapter(BasePlatformAdapter): connected = await self._socket_transport_connected() if connected is False: await self._restart_socket_mode("transport disconnected") + elif self._socket_ping_pong_stale(): + # is_connected() can lie when the aiohttp session is closed + # but the client keeps retrying; ping/pong staleness catches + # that wedged-zombie case that the bool check above misses. + await self._restart_socket_mode("ping/pong stale") except asyncio.CancelledError: raise except Exception: # pragma: no cover - defensive logging diff --git a/tests/gateway/test_slack.py b/tests/gateway/test_slack.py index d8c25a4a9d04..61fc14512fa9 100644 --- a/tests/gateway/test_slack.py +++ b/tests/gateway/test_slack.py @@ -12,6 +12,7 @@ import asyncio import contextlib import os import sys +import time from unittest.mock import AsyncMock, MagicMock, patch, call import pytest @@ -802,6 +803,85 @@ class TestSlackSocketWatchdog: finally: await adapter.disconnect() + # -- ping/pong staleness: heals the wedged transport that is_connected() misses -- + + def _adapter_with_fake_client(self, **client_attrs): + adapter = SlackAdapter(PlatformConfig(enabled=True, token="xoxb-fake")) + client = MagicMock() + for key, value in client_attrs.items(): + setattr(client, key, value) + adapter._handler = MagicMock(client=client) + return adapter + + def test_ping_pong_stale_when_last_ping_old(self): + adapter = self._adapter_with_fake_client( + ping_interval=30, last_ping_pong_time=time.time() - 1000 + ) + assert adapter._socket_ping_pong_stale() is True + + def test_ping_pong_fresh_when_last_ping_recent(self): + adapter = self._adapter_with_fake_client( + ping_interval=30, last_ping_pong_time=time.time() - 5 + ) + assert adapter._socket_ping_pong_stale() is False + + def test_ping_pong_none_within_grace_not_stale(self): + adapter = self._adapter_with_fake_client( + ping_interval=30, last_ping_pong_time=None + ) + adapter._socket_handler_started_monotonic = time.monotonic() + assert adapter._socket_ping_pong_stale() is False + + def test_ping_pong_none_beyond_grace_is_stale(self): + adapter = self._adapter_with_fake_client( + ping_interval=30, last_ping_pong_time=None + ) + adapter._socket_first_ping_grace_s = 0.0 + adapter._socket_handler_started_monotonic = time.monotonic() - 200 + assert adapter._socket_ping_pong_stale() is True + + def test_ping_pong_no_handler_not_stale(self): + adapter = SlackAdapter(PlatformConfig(enabled=True, token="xoxb-fake")) + adapter._handler = None + assert adapter._socket_ping_pong_stale() is False + + def test_ping_pong_nonnumeric_attrs_not_stale(self): + # A mocked/partial client (MagicMock attrs) must never trigger reconnect. + adapter = SlackAdapter(PlatformConfig(enabled=True, token="xoxb-fake")) + adapter._handler = MagicMock() + assert adapter._socket_ping_pong_stale() is False + + @pytest.mark.asyncio + async def test_watchdog_reconnects_when_ping_pong_stale_despite_is_connected_true(self): + adapter = SlackAdapter(PlatformConfig(enabled=True, token="xoxb-fake")) + adapter._socket_watchdog_interval_s = 0.01 + factory, instances = self._make_fake_handler_factory() + + with contextlib.ExitStack() as stack: + for p in self._patch_stack(factory): + stack.enter_context(p) + + try: + assert await adapter.connect() is True + assert len(instances) == 1 + + # Transport lies: is_connected() stays True while ping/pong has + # gone stale (the wedged "Session is closed" zombie). + instances[0].client.is_connected = lambda: True + instances[0].client.ping_interval = 30 + instances[0].client.last_ping_pong_time = time.time() - 1000 + + for _ in range(40): + if len(instances) >= 2: + break + await asyncio.sleep(0.01) + + assert len(instances) >= 2, "watchdog did not heal wedged (lying) transport" + assert instances[0].closed is True + assert adapter._handler is instances[-1] + finally: + await adapter.disconnect() + # --------------------------------------------------------------------------- # TestSlackProxyBehavior