mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-27 13:40:37 +00:00
fe22e3899a
* fix(channels): keep blocking SDK teardown off the event loop in stop()
SlackChannel.stop() called SocketModeClient.close() inline. close() joins
the SDK's message-processor thread (~0.7s every call, measured) and waits
for in-flight listeners, whose blocking Web API calls can run up to the
WebClient timeout, so every run, stream and channel on the Gateway stalled
during shutdown and POST /api/channels/{name}/restart.
Feishu and DingTalk had the same shape: they joined their SDK threads
inline with a 5s timeout, and since those threads only return on a fatal
error, the join normally froze the loop for the full 5s.
The teardown now runs through asyncio.to_thread. Slack detaches the client
first so a still-queued connect skips it, and shields and tracks the close
task so a cancelled stop() leaves it running and a retried stop() awaits
the same close instead of closing twice.
tests/blocking_io/test_channel_stop_offloop.py anchors all three stop()
paths to the strict Blockbuster gate, driving a real slack-sdk client.
* docs(changelog): reference #6134
* fix(channels): join the Discord client thread off the event loop in stop()
DiscordChannel.stop() still joined its client thread inline with a 10s
timeout. The thread usually exits right after the cross-loop client close,
but a timed-out close or a slow _run_client() drain keeps it alive for the
whole timeout, stalling the shared Gateway loop the same way the Slack,
Feishu and DingTalk teardown did. Run the join through asyncio.to_thread so
stop() matches the channels/AGENTS.md teardown rule, and add Discord to the
strict Blockbuster anchor.
64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
"""Regression anchor: channel ``stop()`` must not block the event loop on SDK teardown.
|
|
|
|
``stop()`` runs on the Gateway event loop, both at shutdown and from
|
|
``POST /api/channels/{name}/restart``, so a blocking teardown call there stalls
|
|
every run, stream and channel the Gateway serves:
|
|
|
|
- Slack: ``SocketModeClient.close()`` joins the SDK's message-processor thread
|
|
and waits for in-flight listeners.
|
|
- Feishu / DingTalk: ``stop()`` joins the SDK thread with a 5s timeout, and
|
|
that thread only returns on a fatal error.
|
|
- Discord: ``stop()`` joins the client thread with a 10s timeout, which a
|
|
timed-out client close or a slow ``_run_client()`` drain can use up.
|
|
|
|
Each test drives the real ``stop()`` (and, for Slack, a real slack-sdk client).
|
|
If a join or close regresses back onto the event loop, the strict Blockbuster
|
|
gate raises ``BlockingError`` from the thread-join lock wait.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from app.channels.dingtalk import DingTalkChannel
|
|
from app.channels.discord import DiscordChannel
|
|
from app.channels.feishu import FeishuChannel
|
|
from app.channels.message_bus import MessageBus
|
|
from app.channels.slack import SlackChannel
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
async def test_slack_stop_closes_real_socket_mode_client_off_loop() -> None:
|
|
from slack_sdk.socket_mode import SocketModeClient
|
|
|
|
# Never connected: construction alone starts the SDK's worker threads,
|
|
# which is what close() has to join.
|
|
socket_client = SocketModeClient(app_token="xapp-test")
|
|
channel = SlackChannel(bus=MessageBus(), config={})
|
|
channel._socket_client = socket_client
|
|
channel._running = True
|
|
|
|
await channel.stop()
|
|
|
|
assert socket_client.closed is True
|
|
assert channel._socket_client is None
|
|
|
|
|
|
@pytest.mark.parametrize("channel_cls", [FeishuChannel, DingTalkChannel, DiscordChannel])
|
|
async def test_stop_joins_sdk_thread_off_loop(channel_cls) -> None:
|
|
release = threading.Event()
|
|
sdk_thread = threading.Thread(target=release.wait, daemon=True)
|
|
sdk_thread.start()
|
|
channel = channel_cls(MessageBus(), {})
|
|
channel._thread = sdk_thread
|
|
channel._running = True
|
|
release.set()
|
|
|
|
await channel.stop()
|
|
|
|
assert not sdk_thread.is_alive()
|
|
assert channel._thread is None
|