From fe22e3899a1cbe689cd7e5e3f60201602b45179f Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Thu, 1 Oct 2026 11:00:53 +0900 Subject: [PATCH] fix(channels): keep blocking SDK teardown off the event loop in stop() (#6134) * 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. --- CHANGELOG.md | 14 +++ CHANGELOG_zh.md | 10 +++ backend/app/channels/AGENTS.md | 2 +- backend/app/channels/dingtalk.py | 4 +- backend/app/channels/discord.py | 5 +- backend/app/channels/feishu.py | 4 +- backend/app/channels/slack.py | 20 ++++- .../blocking_io/test_channel_stop_offloop.py | 63 +++++++++++++ backend/tests/test_dingtalk_channel.py | 26 ++++++ backend/tests/test_discord_channel.py | 28 ++++++ backend/tests/test_feishu_parser.py | 25 ++++++ .../tests/test_slack_channel_connections.py | 90 +++++++++++++++++++ 12 files changed, 284 insertions(+), 7 deletions(-) create mode 100644 backend/tests/blocking_io/test_channel_stop_offloop.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1dd5c178af..6bb07fd6a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -433,6 +433,19 @@ This release closes that milestone with **301 merged pull requests**. ### Fixed +- **channels:** Stopping or restarting the Slack, Feishu, DingTalk or Discord + channel no longer freezes the Gateway event loop. `SlackChannel.stop()` called + `SocketModeClient.close()` inline, which joins the SDK's message-processor + thread (about 0.7s every time) and waits for in-flight event listeners, whose + blocking Slack Web API calls can run up to the client timeout; Feishu and + DingTalk joined their SDK threads inline with a 5s timeout, and because those + threads only exit on a fatal error the join normally waited the full 5s; + Discord joined its client thread inline with a 10s timeout, which a timed-out + client close or a slow drain could use up. Every run, stream and channel on + the Gateway stalled meanwhile, including on + `POST /api/channels/{name}/restart`. The teardown now runs in a worker thread; + Slack's close is shielded and tracked, so a cancelled shutdown leaves it + running and a retried `stop()` awaits it instead of closing twice. ([#6134]) - **middleware:** `_externalize_to_sandbox` now validates the full byte count of externalized tool outputs instead of only testing non-emptiness with `test -s`. When a remote sandbox write truncated the file part-way (e.g. disk @@ -7489,4 +7502,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6091]: https://github.com/bytedance/deer-flow/pull/6091 [#6093]: https://github.com/bytedance/deer-flow/pull/6093 [#6112]: https://github.com/bytedance/deer-flow/pull/6112 +[#6134]: https://github.com/bytedance/deer-flow/pull/6134 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 69ab13233c..b6658ec7e2 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -412,6 +412,15 @@ ### 修复 +- **渠道：** 停止或重启 Slack、飞书、钉钉、Discord 渠道不再冻结 Gateway 事件循环。 + `SlackChannel.stop()` 直接调用 `SocketModeClient.close()`，它会 join SDK 的消息处理线程 + （每次约 0.7 秒），并等待正在执行的事件监听器，而监听器中阻塞的 Slack Web API 调用 + 最长可持续到客户端超时；飞书与钉钉则直接以 5 秒超时 join SDK 线程，由于这些线程 + 只在致命错误时退出，这次 join 通常会等满 5 秒；Discord 直接以 10 秒超时 join 客户端线程， + 客户端关闭超时或清理缓慢时会用满这段时间。期间 Gateway 上的所有运行、流和渠道 + 都会停顿，`POST /api/channels/{name}/restart` 也不例外。现在这些清理都在工作线程中执行； + Slack 的 close 会被 shield 并跟踪，因此被取消的关闭流程不会中断它，重试的 `stop()` + 会等待同一个 close，而不会再关闭一次。([#6134]) - **中间件：** `_externalize_to_sandbox` 现在校验外部化工具输出的完整字节大小， 而不再仅使用 `test -s` 检查文件是否非空。此前当远程沙箱写入中途被截断时（如磁盘满 或管道故障），残缺的文件也会通过校验并把错误路径交付给模型；现在会严格核对字节数与负载一致， @@ -6289,3 +6298,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6091]: https://github.com/bytedance/deer-flow/pull/6091 [#6093]: https://github.com/bytedance/deer-flow/pull/6093 [#6112]: https://github.com/bytedance/deer-flow/pull/6112 +[#6134]: https://github.com/bytedance/deer-flow/pull/6134 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 99fba291c6..35589d6f1d 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -10,7 +10,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk - `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout` A swallowed streaming failure publishes its final outbound before releasing the inbound dedupe key, so a provider redelivery can retry without overtaking the terminal reply. **What may be published from the stream is an allowlist, not a denylist** (`_accumulate_stream_text`): only assistant message types — LangChain serializes `AIMessage.type` as `"ai"` and `AIMessageChunk.type` as `"AIMessageChunk"`, plus the OpenAI-style `"assistant"` spelling for foreign runtimes — become displayable text. The previous rule rejected only payloads whose `type` contained `"tool"` and therefore published everything else, which leaked DeerFlow's hidden model context to every streaming IM channel: `DynamicContextMiddleware` injects recalled memory as a hidden `HumanMessage` (`type == "human"`) and rewrites the user's own turn into a new `HumanMessage`, `DurableContextMiddleware` injects a hidden `` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay, which published a `` fact block and, in another run, a verbatim echo of the user's own message as the assistant's reply. Matching is by prefix (`ai` / `assistant`), never substring, because ordinary words contain `"ai"` (`chain`, `domain`). The message type is resolved by `_stream_payload_type`, which handles both the `model_dump()` shape DeerFlow's own gateway emits and LangChain's `to_json()` constructor shape (whose top-level `type` is the literal `"constructor"`, with the class name at the tail of the `id` path). A bare `str` payload is no longer accepted at all: it carries no type information, so it cannot be attributed to the assistant, and nothing in DeerFlow produces one (`runtime/serialization.py::serialize_messages_tuple` always emits `[message_dict, metadata]`). -- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()`: it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. +- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()`: it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`, since every run and channel on the Gateway shares that loop and `stop()` also serves `POST /api/channels/{name}/restart`; Slack's `SocketModeClient.close()` is a shielded, tracked task so a cancelled stop leaves it running and a retried stop awaits the same close. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. - `service.py` - Manages lifecycle of all configured channels from `config.yaml`. Shutdown closes manager admission first and keeps transports alive until every manager worker/follow-up watcher has exited. A successful manager stop therefore owns no live handler; if the Gateway's outer timeout cancels shutdown, the service retains its channel objects and global singleton so unfinished resources are not detached and cleanup can be retried. - `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, edits the "Working on it..." stream target in place via `editMessageText`, and can optionally send final Markdown replies as Rich Messages through `channels.telegram.rich_messages` — the Rich path only fires when `rich_messages` is on **and** the text contains a rich construct (fenced code, emphasis, a table, a task list, `
`, block math, or a link), so structured command/error replies (none of these) stay plain text and their newlines and `` tokens survive; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; because `stop()` runs on the main loop while typing tasks belong to `_discord_loop`, normal cross-thread cancellation, awaiting, and map cleanup are scheduled there with a bounded wait, while `_run_client()` drains tasks in its `finally` block before an exception/disconnect can make that loop unusable; an already-stopped foreign loop must never have its tasks awaited from the main loop; to that end every outbound cross-loop call (`send` / `send_file` / `_get_channel_or_thread`) goes through `_run_on_discord_loop`, which bounds the await (`DISCORD_OUTBOUND_TIMEOUT_SECONDS` 30s; `DISCORD_UPLOAD_TIMEOUT_SECONDS` 120s for file uploads, whose unbounded-size payload needs room for a slow uplink plus 429 retry-after) and fails fast with a `RuntimeError` when the client loop is missing or not running — a dead client becomes a logged send failure instead of a permanently hung `ChannelManager` worker — and `is_running` reports client-thread aliveness (like `feishu.py`) so `ensure_channel_ready` can restart the channel after `_run_client()` exits on a fatal error; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`) - `buzz.py` - Buzz (Nostr relay) implementation: one NIP-42-authenticated websocket, pubkey-allowlist + mention/DM/thread-follow gating, streaming replies via in-place kind-40003 edits; requires the `buzz` dependency extra. Its durable seen-event replay guard uses async `aseen()` / `arecord()` / `aflush()` boundaries: initial JSON reads and coalesced atomic writes run off the Gateway event loop, and `stop()` awaits the final flush before returning. **Subscription model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model)): the relay fans kind-9 chat events out **only** to channel-scoped subscriptions, proved against a live relay — `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event (the connector authenticated and then sat silent forever), `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel (the same shape as Buzz's own `buzz-acp` harness). Every connection therefore rebuilds three kinds of subscription after NIP-42 auth: `buzz-discovery` (`{"kinds":[39000]}`, a historical query returning exactly the channels this identity belongs to, one stored event each then `EOSE` — adding `#p` returns zero, do not "narrow" it), `buzz-membership` (`{"kinds":[44100,44101],"#p":[us],"since":}`, the relay-signed member-added/member-removed notifications whose `p` tag names the affected member and `h` tag the channel), and one `buzz-chat-` per discovered channel. Chat subscriptions open as each kind-39000 arrives; the discovery `EOSE` is the completeness barrier that retries any that failed and warns when discovery found nothing. A kind-44100 for our pubkey subscribes to the new channel **live** (then re-issues discovery so its name/type reach the DM-detection cache, but only when that channel's metadata is actually missing — an unconditional refresh is how a burst of 44100s multiplied into one discovery pass each); a kind-44101 issues `buzz_nostr.close_frame` for exactly that channel's subscription and drops its metadata. **The membership subscription is scoped to LIVE events** and this is load-bearing: buzz-relay *stores* 44100/44101 and serves history newest-first (default limit 2000), so an unscoped filter replayed the whole membership history on every connect — every stored add read as live, re-running discovery once each (M+1 discovery passes × N stored kind-39000 events per connect, observed live as two `channel discovery complete` lines and one channel logged `` because the 44100 path subscribed before its metadata arrived), re-subscribing channels we have since been removed from, and letting a stored 44101 transiently unsubscribe a channel we are still in. `since` is anchored at the moment the socket opened (`_session_started_at`) minus `MEMBERSHIP_LOOKBACK_SECONDS` (60) of slack, which covers both relay clock skew and a membership change published *during* the connect/auth handshake; the slack can only cost an idempotent replay of the last minute. **A relay `CLOSED` frame is recovered, not merely forgotten** — every subscription on the socket fails silently when dropped, so `_handle_closed` re-issues it, bounded by `MAX_RESUBSCRIBE_ATTEMPTS` (3) per subscription id per connection *and per auth epoch* (`_resubscribe_attempts` is reset on session start, on re-auth, and by `stop()`, so pre-auth rejections — which the auth branch already recovers wholesale — never spend the authenticated session's budget). The first retry is immediate (the common case is a one-off hiccup); later ones back off 1s then 2s, awaited inline in the read loop rather than in a background task that could outlive its own socket. **`auth-required:` before this socket has completed its NIP-42 handshake is the expected bootstrap sequence, not a refusal**, and `_handle_closed` short-circuits it ahead of every recovery path: the connector opens its control REQs immediately in case the relay serves unauthenticated reads, a closed relay answers `auth-required:` plus an `AUTH` challenge, and the auth branch re-opens everything. That case is logged at DEBUG and consumes neither the permanent-refusal branch nor the retry budget (a chat subscription is still dropped from `_chat_subscriptions`, since it genuinely is not subscribed and discovery is what re-opens it). Treating it as permanent produced an operator-facing warning claiming discovery/membership tracking was DOWN in the same run where discovery then completed, every channel was subscribed, and a brand-new channel's kind-44100 was picked up one second later. The boundary is the per-socket `_auth_completed` flag, set once the signed AUTH event has been sent and cleared on session entry, session exit, and `stop()`; the same reason *after* that stays loud, and says the subscription is down until the relay's next AUTH challenge or the next reconnect rather than borrowing the non-auth wording. Then `_is_transient_close` decides whether to retry at all: NIP-01/NIP-42 `auth-required:`/`restricted:`/`blocked:`/`mute:`/`invalid:`/`pow:` prefixes and buzz-relay's own removal/revocation prose are permanent (do not fight the relay over a channel that is no longer ours; a post-auth `auth-required:` is recovered by the AUTH branch, not by re-issuing), `rate-limited:`/`error:`/no-reason-at-all and anything unrecognized are transient — the default resolves toward "keep listening" because going silently deaf is the failure this exists to remove, and the attempt budget bounds a wrong guess. A chat `CLOSED` is only ever recovered for a channel already in `_chat_subscriptions`: a `CLOSED` is relay-supplied, so acting on an unknown one would let a relay induce a subscription just by naming a channel. Every subscription that goes unlistened is logged at WARNING, never INFO. Subscription ids are deterministic per channel precisely so one can be replaced or closed without disturbing the others on the socket, and `_chat_subscriptions` is per-socket state cleared on session end, on re-auth (a pre-auth REQ may have been rejected), and by `stop()`. Three bounds on remote-fed state: `MAX_CACHED_CHANNELS` (512) caps the kind-39000 metadata cache, the watermark map, and the resubscribe-attempt map; `MAX_CHANNEL_SUBSCRIPTIONS` (256, well under buzz-relay's own 1024-per-connection ceiling) caps live chat subscriptions — at the cap new channels are refused and named in a warning rather than evicting a working subscription. **Known bound (documented, not fixed):** the relay caps historical delivery at 2000 events per subscription, newest-first, even with a `since`, so >2000 unread messages in a *single* channel across a disconnect loses the oldest — the relay never sends them and the watermark advances past them. That is the one remaining path that can skip; everything else fails toward replay. **Trust model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-trust-model)): every inbound `EVENT` is authenticated at the single `handle_relay_frame` choke point — the NIP-01 id is recomputed from the delivered payload and the BIP-340 Schnorr signature verified against the claimed `pubkey` (`buzz_nostr.verify_event`, pure and total: malformed input returns `False`, never raises) — so `ev["pubkey"]`, the authorization principal for both the allowlist and the `/connect` bind, cannot be forged by a relay the DeerFlow operator does not run. What remains trusted is the *authorship* of kind-39000 channel metadata: any member can sign one, and because per-channel subscriptions are now driven by discovery, a forged kind-39000 has two effects rather than one — it can mark a channel `type: "dm"` (relaxing `require_mention` for that channel) **and** it can induce a chat subscription for a channel of the forger's choosing, since the channels we listen to are exactly the channels we hold metadata for. Neither makes anything be *acted on*: `allowed_users` and per-event signature verification are independent gates, so an induced subscription only means the relay reads its own traffic back to a subscriber that drops it, bounded by `MAX_CHANNEL_SUBSCRIPTIONS` (which refuses rather than evicts, so it cannot displace a real channel). Same for a forged kind-44100, except its `p` tag is re-checked locally so it must at least name us. Closing this needs a configured trusted relay pubkey, which `relay_url` is not. `allowed_users` is deny-by-default (empty = nobody, unlike siblings' empty = everyone), so `start()` logs a WARNING when it is empty and each drop logs at DEBUG. The resubscribe cursor (`since`) is **per channel**, advances only for events that were actually processed, and never past `now + MAX_FUTURE_SKEW_SECONDS`, because it is peer-supplied (`created_at`) and a single future-dated event otherwise made the connector permanently deaf. Per channel rather than global is the safety-critical half: subscriptions are per channel, so one shared cursor is the newest event seen in *any* channel and a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect — measured on a live relay, three channels of one identity sat ~28h apart. Per-channel cursors can only ever cost duplicate delivery (absorbed by the manager's `event_id` dedupe), and an evicted cursor degrades to "no `since`", i.e. the relay's default backlog — both fail toward replay, never toward a miss. Streaming tracks every oversize chunk index (`_stream_targets` for chunk 0, `_stream_tails` for the rest), since the manager republishes cumulative text and reposting `chunks[1:]` per update flooded the channel; all of it is per-connection state cleared by `stop()`, and the remote-fed kind-39000 cache is capped at `MAX_CACHED_CHANNELS`. **`send()` refuses outright to publish text carrying a hidden model-context wrapper** (``, ``, `` — `_HIDDEN_CONTEXT_MARKERS`), logging at ERROR and clearing the stream bookkeeping on a blocked `is_final`. This is defense in depth behind the manager's allowlist, and it lives here rather than in a sibling connector because on Buzz a leak is permanent: every streaming update is an immutable public Nostr event, so a corrective edit only changes what clients render while the original leaked event stays on the relay. Matching is on the literal opening tag, so a reply that merely talks about memory is still published diff --git a/backend/app/channels/dingtalk.py b/backend/app/channels/dingtalk.py index 362ed73c41..f7425c7eaa 100644 --- a/backend/app/channels/dingtalk.py +++ b/backend/app/channels/dingtalk.py @@ -228,7 +228,9 @@ class DingTalkChannel(Channel): self._card_repliers.clear() self._card_track_ids.clear() if self._thread: - self._thread.join(timeout=5) + # The SDK thread only returns on a fatal error, so this join + # normally waits out its full timeout; keep it off the event loop. + await asyncio.to_thread(self._thread.join, timeout=5) self._thread = None logger.info("DingTalk channel stopped") diff --git a/backend/app/channels/discord.py b/backend/app/channels/discord.py index 32699216fb..784dc5cb6a 100644 --- a/backend/app/channels/discord.py +++ b/backend/app/channels/discord.py @@ -264,7 +264,10 @@ class DiscordChannel(Channel): logger.exception("[Discord] error while closing client") if self._thread: - self._thread.join(timeout=10) + # The client thread normally exits right after the close above, but + # a timed-out close or a slow ``_run_client()`` drain can keep it + # alive for the whole join timeout; keep that wait off the loop. + await asyncio.to_thread(self._thread.join, timeout=10) self._thread = None # _run_client() normally drains these tasks in its finally block. If diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index b97611ad69..df1a1a5ae1 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -283,7 +283,9 @@ class FeishuChannel(Channel): self._background_tasks.clear() self._running_card_tasks.clear() if self._thread: - self._thread.join(timeout=5) + # The SDK thread only returns on a fatal error, so this join + # normally waits out its full timeout; keep it off the event loop. + await asyncio.to_thread(self._thread.join, timeout=5) self._thread = None logger.info("Feishu channel stopped") diff --git a/backend/app/channels/slack.py b/backend/app/channels/slack.py index 9d562b7582..0cf6e1cac4 100644 --- a/backend/app/channels/slack.py +++ b/backend/app/channels/slack.py @@ -103,6 +103,7 @@ class SlackChannel(Channel): def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None: super().__init__(name="slack", bus=bus, config=config) self._socket_client = None + self._socket_close_task: asyncio.Task[None] | None = None self._web_client = None self._loop: asyncio.AbstractEventLoop | None = None self._allowed_users = _normalize_allowed_users(config.get("allowed_users", [])) @@ -176,9 +177,22 @@ class SlackChannel(Channel): self._running = False self.bus.unsubscribe_outbound(self._on_outbound) await self._close_and_drain_threadsafe_futures() - if self._socket_client: - self._socket_client.close() - self._socket_client = None + # ``SocketModeClient.close()`` joins the SDK's message-processor thread + # (up to ~1s) and waits for in-flight listeners, whose blocking Web API + # calls run up to the WebClient timeout, so it runs off the event loop. + # The client is detached first so a still-queued connect skips it. The + # close task is shielded and tracked: a cancelled stop() leaves it + # running, and a retried stop() awaits it instead of closing twice. + socket_client, self._socket_client = self._socket_client, None + if socket_client is not None: + self._socket_close_task = asyncio.create_task(asyncio.to_thread(socket_client.close)) + close_task = self._socket_close_task + if close_task is not None: + try: + await asyncio.shield(close_task) + finally: + if close_task.done(): + self._socket_close_task = None logger.info("Slack channel stopped") async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None: diff --git a/backend/tests/blocking_io/test_channel_stop_offloop.py b/backend/tests/blocking_io/test_channel_stop_offloop.py new file mode 100644 index 0000000000..b4baffbd30 --- /dev/null +++ b/backend/tests/blocking_io/test_channel_stop_offloop.py @@ -0,0 +1,63 @@ +"""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 diff --git a/backend/tests/test_dingtalk_channel.py b/backend/tests/test_dingtalk_channel.py index 6f41208798..84b18727e7 100644 --- a/backend/tests/test_dingtalk_channel.py +++ b/backend/tests/test_dingtalk_channel.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import logging +import threading from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -1770,6 +1771,31 @@ class TestCardMode: _run(go()) +class TestStop: + def test_stop_joins_stream_thread_off_the_event_loop(self): + async def go(): + channel = DingTalkChannel(MessageBus(), config={}) + release = threading.Event() + # dingtalk-stream's start_forever() never returns, so the join in + # stop() waits out its timeout; this thread blocks the same way, + # bounded so a join run on the event loop fails the assertion + # below instead of hanging the test. + stream_thread = threading.Thread(target=release.wait, args=(2,), daemon=True) + stream_thread.start() + channel._thread = stream_thread + channel._running = True + + stop_task = asyncio.create_task(channel.stop()) + await asyncio.sleep(0.05) + assert not stop_task.done() + + release.set() + await stop_task + assert channel._thread is None + + _run(go()) + + # --------------------------------------------------------------------------- # Inbound file support # --------------------------------------------------------------------------- diff --git a/backend/tests/test_discord_channel.py b/backend/tests/test_discord_channel.py index 1f999ebae6..5304f11b6c 100644 --- a/backend/tests/test_discord_channel.py +++ b/backend/tests/test_discord_channel.py @@ -115,6 +115,34 @@ async def test_discord_stop_does_not_clobber_store_before_load(tmp_path) -> None assert json.loads(store_path.read_text()) == {"chan-9": "thread-9"} +@pytest.mark.asyncio +async def test_discord_stop_joins_client_thread_off_the_event_loop() -> None: + """stop() must not join the client thread on the shared Gateway loop. + + 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 up to + the 10s join timeout, and the Gateway loop must keep running meanwhile. + """ + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + release = threading.Event() + # Bounded so a join run on the event loop fails the assertion below + # instead of hanging the test. + client_thread = threading.Thread(target=release.wait, args=(2,), daemon=True) + client_thread.start() + channel._discord_loop = None + channel._client = None + channel._thread = client_thread + channel._cancel_ephemeral_tasks = AsyncMock() + + stop_task = asyncio.create_task(channel.stop()) + await asyncio.sleep(0.05) + assert not stop_task.done() + + release.set() + await stop_task + assert channel._thread is None + + def _make_discord_message(text: str): return SimpleNamespace( id=111, diff --git a/backend/tests/test_feishu_parser.py b/backend/tests/test_feishu_parser.py index 1f814f53f8..bafc7c6dda 100644 --- a/backend/tests/test_feishu_parser.py +++ b/backend/tests/test_feishu_parser.py @@ -1,6 +1,7 @@ import asyncio import json import tempfile +import threading from io import BytesIO from pathlib import Path from unittest.mock import AsyncMock, MagicMock @@ -165,6 +166,30 @@ def test_feishu_is_not_running_when_ws_thread_exits(): assert channel.is_running is False +def test_feishu_stop_joins_ws_thread_off_the_event_loop(): + async def go(): + channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"}) + release = threading.Event() + # lark's ws client never returns from start(), so the join in stop() + # waits out its timeout; this thread blocks the same way, bounded so a + # join run on the event loop fails the assertion below instead of + # hanging the test. + ws_thread = threading.Thread(target=release.wait, args=(2,), daemon=True) + ws_thread.start() + channel._thread = ws_thread + channel._running = True + + stop_task = asyncio.create_task(channel.stop()) + await asyncio.sleep(0.05) + assert not stop_task.done() + + release.set() + await stop_task + assert channel._thread is None + + asyncio.run(go()) + + def test_feishu_event_handler_ignores_non_content_message_events(): import lark_oapi as lark diff --git a/backend/tests/test_slack_channel_connections.py b/backend/tests/test_slack_channel_connections.py index f7aea73a20..cec537d22f 100644 --- a/backend/tests/test_slack_channel_connections.py +++ b/backend/tests/test_slack_channel_connections.py @@ -4,10 +4,13 @@ from __future__ import annotations import asyncio import sys +import threading from datetime import UTC, datetime, timedelta from types import ModuleType from unittest.mock import AsyncMock, MagicMock +import pytest + from app.channels.message_bus import MessageBus, OutboundMessage @@ -242,7 +245,13 @@ def test_slack_socket_mode_queued_connect_does_not_target_replacement_client(mon queued_callbacks = [] loop = asyncio.get_running_loop() + real_run_in_executor = loop.run_in_executor + def queue_run_in_executor(executor, func, *args): + # Hold back only the connect; stop() offloads close() through the + # same executor path and must still run. + if func != channel._connect_socket_mode: + return real_run_in_executor(executor, func, *args) queued_callbacks.append((func, args)) return loop.create_future() @@ -268,3 +277,84 @@ def test_slack_socket_mode_queued_connect_does_not_target_replacement_client(mon await channel.stop() anyio.run(go) + + +class _BlockingSocketModeClient: + """Socket Mode stand-in whose ``close()`` blocks like the SDK's thread joins.""" + + def __init__(self): + self.close_started = threading.Event() + self.release = threading.Event() + self.close_calls = 0 + self.connect_calls = 0 + + def connect(self): + self.connect_calls += 1 + + def close(self): + self.close_calls += 1 + self.close_started.set() + # Bounded so a close() run on the event loop fails the test's + # assertions instead of deadlocking it. + self.release.wait(timeout=2) + + +def test_slack_stop_closes_socket_client_off_the_event_loop(): + import anyio + + from app.channels.slack import SlackChannel + + async def go(): + channel = SlackChannel(bus=MessageBus(), config={}) + socket_client = _BlockingSocketModeClient() + channel._socket_client = socket_client + channel._running = True + + stop_task = asyncio.create_task(channel.stop()) + assert await asyncio.to_thread(socket_client.close_started.wait, 2) + # close() is blocked in a worker thread, so the loop keeps running and + # stop() is still waiting for it. + assert not stop_task.done() + # A connect still queued in the executor must not reopen the client + # that is being closed. + channel._connect_socket_mode(socket_client) + assert socket_client.connect_calls == 0 + + socket_client.release.set() + await stop_task + assert socket_client.close_calls == 1 + assert channel._socket_client is None + + anyio.run(go) + + +def test_slack_stop_retried_after_cancellation_awaits_the_same_close(): + import anyio + + from app.channels.slack import SlackChannel + + async def go(): + channel = SlackChannel(bus=MessageBus(), config={}) + socket_client = _BlockingSocketModeClient() + channel._socket_client = socket_client + channel._running = True + + first_stop = asyncio.create_task(channel.stop()) + assert await asyncio.to_thread(socket_client.close_started.wait, 2) + first_stop.cancel() + with pytest.raises(asyncio.CancelledError): + await first_stop + + # The Gateway retries a cancelled shutdown on the same channel. The + # retry must wait for the close still running in the worker rather + # than return early or start a second close. + retry = asyncio.create_task(channel.stop()) + await asyncio.sleep(0.05) + assert not retry.done() + + socket_client.release.set() + await retry + assert socket_client.close_calls == 1 + assert channel._socket_client is None + + anyio.run(go)