mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(slack): log Socket Mode connect failures instead of dropping them (#5640)
* fix(slack): log Socket Mode connect failures instead of dropping them SlackChannel.start() ran SocketModeClient.connect() through loop.run_in_executor() and discarded the returned future. When the connect raised (rejected app token, network error, TLS failure) the exception was stored on a future nothing awaited, so the only trace was asyncio's "Future exception was never retrieved" at garbage collection. Meanwhile the channel had already logged "Slack channel started" and reported itself running. Route the executor call through _connect_socket_mode(), which logs the failure with logger.exception() the way the Telegram and Discord channel threads already do. The regression test injects a failing connect and asserts the error reaches the channel logger. * fix(slack): bind queued connect to its client --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
29dbce45a1
commit
bc262743d7
@ -141,10 +141,27 @@ class SlackChannel(Channel):
|
||||
self._running = True
|
||||
self.bus.subscribe_outbound(self._on_outbound)
|
||||
|
||||
# Start socket mode in background thread
|
||||
asyncio.get_event_loop().run_in_executor(None, self._socket_client.connect)
|
||||
# Start socket mode in background thread. Go through
|
||||
# ``_connect_socket_mode`` so a connect failure gets logged: nothing
|
||||
# awaits the future ``run_in_executor`` returns, so an exception stored
|
||||
# on it would never surface.
|
||||
socket_client = self._socket_client
|
||||
asyncio.get_event_loop().run_in_executor(None, self._connect_socket_mode, socket_client)
|
||||
logger.info("Slack channel started")
|
||||
|
||||
def _connect_socket_mode(self, socket_client: Any) -> None:
|
||||
"""Connect one start attempt's Socket Mode client if it is still current.
|
||||
|
||||
Mirrors the Telegram and Discord channels, whose background threads log
|
||||
their errors instead of letting them vanish with the thread.
|
||||
"""
|
||||
if socket_client is not self._socket_client:
|
||||
return
|
||||
try:
|
||||
socket_client.connect()
|
||||
except Exception:
|
||||
logger.exception("[Slack] Socket Mode connection failed; the channel will not receive events")
|
||||
|
||||
async def stop(self) -> None:
|
||||
self._running = False
|
||||
self.bus.unsubscribe_outbound(self._on_outbound)
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from types import ModuleType
|
||||
@ -135,3 +136,135 @@ def test_slack_http_events_mode_is_rejected(monkeypatch, caplog):
|
||||
assert "Slack HTTP Events mode is not supported" in caplog.text
|
||||
|
||||
anyio.run(go)
|
||||
|
||||
|
||||
def test_slack_socket_mode_connect_failure_is_logged(monkeypatch, caplog):
|
||||
import anyio
|
||||
|
||||
from app.channels.slack import SlackChannel
|
||||
|
||||
class FailingSocketModeClient:
|
||||
def __init__(self, *, app_token, web_client):
|
||||
self.socket_mode_request_listeners = []
|
||||
|
||||
def connect(self):
|
||||
raise RuntimeError("apps.connections.open failed: invalid_auth")
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
slack_sdk = ModuleType("slack_sdk")
|
||||
slack_sdk.WebClient = MagicMock
|
||||
socket_mode = ModuleType("slack_sdk.socket_mode")
|
||||
socket_mode.SocketModeClient = FailingSocketModeClient
|
||||
response = ModuleType("slack_sdk.socket_mode.response")
|
||||
response.SocketModeResponse = object
|
||||
monkeypatch.setitem(sys.modules, "slack_sdk", slack_sdk)
|
||||
monkeypatch.setitem(sys.modules, "slack_sdk.socket_mode", socket_mode)
|
||||
monkeypatch.setitem(sys.modules, "slack_sdk.socket_mode.response", response)
|
||||
|
||||
async def go():
|
||||
channel = SlackChannel(
|
||||
bus=MessageBus(),
|
||||
config={
|
||||
"bot_token": "xoxb-operator",
|
||||
"app_token": "xapp-token",
|
||||
# A configured bot user id skips the auth_test lookup, so the
|
||||
# only background work start() schedules is the connect.
|
||||
"bot_user_id": "UBOT",
|
||||
"connection_repo": MagicMock(),
|
||||
},
|
||||
)
|
||||
|
||||
# start() drops the future for the executor-backed connect; capture it
|
||||
# here so the test can wait for the connect attempt to finish.
|
||||
loop = asyncio.get_running_loop()
|
||||
real_run_in_executor = loop.run_in_executor
|
||||
connect_futures = []
|
||||
|
||||
def capture_run_in_executor(executor, func, *args):
|
||||
future = real_run_in_executor(executor, func, *args)
|
||||
connect_futures.append(future)
|
||||
return future
|
||||
|
||||
monkeypatch.setattr(loop, "run_in_executor", capture_run_in_executor)
|
||||
|
||||
with caplog.at_level("ERROR", logger="app.channels.slack"):
|
||||
await channel.start()
|
||||
assert len(connect_futures) == 1
|
||||
await asyncio.gather(*connect_futures, return_exceptions=True)
|
||||
await channel.stop()
|
||||
|
||||
assert "Socket Mode connection failed" in caplog.text
|
||||
|
||||
anyio.run(go)
|
||||
|
||||
|
||||
def test_slack_socket_mode_queued_connect_does_not_target_replacement_client(monkeypatch):
|
||||
import anyio
|
||||
|
||||
from app.channels.slack import SlackChannel
|
||||
|
||||
clients = []
|
||||
|
||||
class RecordingSocketModeClient:
|
||||
def __init__(self, *, app_token, web_client):
|
||||
self.socket_mode_request_listeners = []
|
||||
self.connect_calls = 0
|
||||
clients.append(self)
|
||||
|
||||
def connect(self):
|
||||
self.connect_calls += 1
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
slack_sdk = ModuleType("slack_sdk")
|
||||
slack_sdk.WebClient = MagicMock
|
||||
socket_mode = ModuleType("slack_sdk.socket_mode")
|
||||
socket_mode.SocketModeClient = RecordingSocketModeClient
|
||||
response = ModuleType("slack_sdk.socket_mode.response")
|
||||
response.SocketModeResponse = object
|
||||
monkeypatch.setitem(sys.modules, "slack_sdk", slack_sdk)
|
||||
monkeypatch.setitem(sys.modules, "slack_sdk.socket_mode", socket_mode)
|
||||
monkeypatch.setitem(sys.modules, "slack_sdk.socket_mode.response", response)
|
||||
|
||||
async def go():
|
||||
channel = SlackChannel(
|
||||
bus=MessageBus(),
|
||||
config={
|
||||
"bot_token": "xoxb-operator",
|
||||
"app_token": "xapp-token",
|
||||
"bot_user_id": "UBOT",
|
||||
"connection_repo": MagicMock(),
|
||||
},
|
||||
)
|
||||
queued_callbacks = []
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
def queue_run_in_executor(executor, func, *args):
|
||||
queued_callbacks.append((func, args))
|
||||
return loop.create_future()
|
||||
|
||||
monkeypatch.setattr(loop, "run_in_executor", queue_run_in_executor)
|
||||
|
||||
await channel.start()
|
||||
await channel.stop()
|
||||
await channel.start()
|
||||
|
||||
assert len(clients) == 2
|
||||
assert len(queued_callbacks) == 2
|
||||
|
||||
stale_func, stale_args = queued_callbacks[0]
|
||||
stale_func(*stale_args)
|
||||
assert clients[0].connect_calls == 0
|
||||
assert clients[1].connect_calls == 0
|
||||
|
||||
current_func, current_args = queued_callbacks[1]
|
||||
current_func(*current_args)
|
||||
assert clients[0].connect_calls == 0
|
||||
assert clients[1].connect_calls == 1
|
||||
|
||||
await channel.stop()
|
||||
|
||||
anyio.run(go)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user