From 713a861e75e037e3ff7b4bb61f03d3d0739785cb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=93=88=E5=9F=BA=E7=B1=B3?= <140241684+BlueX888@users.noreply.github.com> Date: Thu, 5 Nov 2026 12:07:31 +0800 Subject: [PATCH] fix(channels): ignore non-finite WeChat server longpoll timeouts (#7898) * fix(channels): ignore non-finite WeChat server longpoll timeouts The getupdates response's longpolling_timeout_ms was passed through float() with only TypeError/ValueError handled. A huge JSON integer raised OverflowError before the message loop and cursor write, so the same batch was re-polled forever, while Infinity/NaN were stored and turned the next request timeout into inf/nan. Apply the same positive-finite rule the channel already uses for configured timing values: catch OverflowError and reject non-finite results, keeping the previous valid timeout. Co-Authored-By: Claude Code * fix(channels): bound WeChat server long-poll hints --------- Co-authored-by: Claude Code Co-authored-by: Willem Jiang --- README.md | 1 + backend/app/channels/AGENTS.md | 2 +- backend/app/channels/wechat.py | 13 +-- backend/tests/test_wechat_channel.py | 119 +++++++++++++++++++++++++++ 4 files changed, 124 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index d813df2d2e..b3cfc5aa54 100644 --- a/README.md +++ b/README.md @@ -1506,6 +1506,7 @@ Scheduled-task result and lifecycle notices can also be pushed to a bound Feishu 7. `allowed_users` takes iLink user IDs. Unset, `null`, `[]`, or a blank string allows everyone. A single ID is one entry, not a sequence of characters, and an unquoted integer-valued number is stored as that integer's text. A scalar string containing commas or interior whitespace logs a warning but remains one literal ID; use a YAML list for multiple IDs. Invalid entries are ignored with a warning; any other configured value that yields no valid ID denies every user and logs an error. `/connect` is still accepted before that check, and a denied sender is dropped before inbound media is downloaded. 8. Shutdown waits for in-flight cursor writes. On token expiry, DeerFlow persists the cursor reset and removes the saved token before completing poller cancellation. 9. Inbound attachments keep separate staged copies even when filenames match within one message or across messages. Their original filenames are preserved for upload ingestion. Private staged copies are removed after ingestion or when a message is dropped or fails. +10. WeChat server long-poll hints are accepted from 1 through 300 seconds. Invalid hints leave the last accepted value in use, or `polling_timeout` if no hint has been accepted. Set `respect_server_longpoll_timeout: false` to always use the configured value. **WeCom Setup** diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index c725e159d1..467cbf79f7 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -73,7 +73,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ - Disabled by default. It is a user-binding layer on top of the existing `channels.*` runtime config, not a replacement for provider bot credentials. - No public IP, OAuth callback URL, or provider webhook route is required by the current implementation. - Telegram binds via deep-link `/start ` over long polling; Slack, Discord, Feishu/Lark, DingTalk, WeChat, and WeCom use `/connect ` through their outbound workers. -- WeChat `polling_timeout`, `polling_retry_delay`, `qrcode_poll_interval`, and `qrcode_poll_timeout` require positive finite seconds; invalid values use defaults to prevent hot loops or endless sleeps. +- WeChat `polling_timeout`, `polling_retry_delay`, `qrcode_poll_interval`, and `qrcode_poll_timeout` use positive finite seconds or defaults. Ignore server long-poll hints outside 1–300 s. - `allowed_users.py`: QQ/WeChat/Telegram/Slack/DingTalk empty allows all; invalid warns; all-invalid denies. WeChat: int IDs, gate downloads. Buzz/Discord `parse_channel_ids`: trim; no splitting/mapping keys; empty = no exemption. Separator warnings: Buzz/QQ/WeChat/Slack/DingTalk. - WeCom serializes `start()` and `stop()` for each channel instance. The SDK `connect()` task covers connection setup only; after the handshake, the SDK owns a separate receive task. Shutdown cancels an in-progress connection attempt and awaits the SDK's actual asynchronous receive-task/socket cleanup before releasing lifecycle state or allowing a restart. Cancellation of `stop()` still propagates, but only after owned cleanup finishes and lifecycle references are cleared; real connection failures remain reported by `_on_ws_task_done`. - WeCom outbound content is capped at the protocol's 20480 UTF-8 bytes: stream replies clip on a character boundary with a truncation marker (one stream carries the whole reply and cannot split mid-way), while proactive pushes split into at most 10 sequential markdown messages per push, with the remaining tail clipped and marker-terminated. Both paths measure bytes, not characters. A per-chat send lock serializes each split batch end to end, because manager workers run concurrently and two long pushes to the same chat would otherwise interleave chunks; locks are reference-counted and reclaimed once no sender holds or waits on them, so the registry does not grow for the life of the Gateway. diff --git a/backend/app/channels/wechat.py b/backend/app/channels/wechat.py index 09643fa748..548b30cf86 100644 --- a/backend/app/channels/wechat.py +++ b/backend/app/channels/wechat.py @@ -1084,16 +1084,9 @@ class WechatChannel(Channel): def _update_longpoll_timeout(self, data: Mapping[str, Any]) -> None: if not self._respect_server_longpoll_timeout: return - raw_timeout = data.get("longpolling_timeout_ms") - if raw_timeout is None: - return - try: - timeout_ms = float(raw_timeout) - except (TypeError, ValueError): - return - if timeout_ms <= 0: - return - self._server_longpoll_timeout_seconds = timeout_ms / 1000.0 + timeout_ms = self._coerce_float(data.get("longpolling_timeout_ms"), 0.0) + if 1_000 <= timeout_ms <= 300_000: + self._server_longpoll_timeout_seconds = timeout_ms / 1000.0 def _base_info(self) -> dict[str, str]: return {"channel_version": self._channel_version} diff --git a/backend/tests/test_wechat_channel.py b/backend/tests/test_wechat_channel.py index 662b986a9c..b512b90812 100644 --- a/backend/tests/test_wechat_channel.py +++ b/backend/tests/test_wechat_channel.py @@ -1683,6 +1683,125 @@ def test_poll_loop_updates_server_timeout(monkeypatch): _run(go()) +@pytest.mark.parametrize( + "value", + [ + 10**400, + float("inf"), + float("-inf"), + float("nan"), + "1e400", + "Infinity", + "-Infinity", + "NaN", + -1.0, + 0.0, + 1, + 999.999, + 300_000.001, + 10**300, + 1e308, + ], +) +@pytest.mark.parametrize("previous_timeout", [None, 42.0]) +def test_update_longpoll_timeout_ignores_invalid_server_value(value, previous_timeout): + from app.channels.wechat import WechatChannel + + channel = WechatChannel(bus=MessageBus(), config={"bot_token": "bot-token", "polling_timeout": 37}) + channel._server_longpoll_timeout_seconds = previous_timeout + + channel._update_longpoll_timeout({"longpolling_timeout_ms": value}) + + assert channel._server_longpoll_timeout_seconds == previous_timeout + assert channel._current_longpoll_timeout_seconds() == (37.0 if previous_timeout is None else previous_timeout) + + +@pytest.mark.parametrize(("value", "expected"), [(1_000, 1.0), ("42000", 42.0), (300_000, 300.0)]) +def test_update_longpoll_timeout_accepts_values_in_range(value, expected): + from app.channels.wechat import WechatChannel + + channel = WechatChannel(bus=MessageBus(), config={"bot_token": "bot-token"}) + + channel._update_longpoll_timeout({"longpolling_timeout_ms": value}) + + assert channel._current_longpoll_timeout_seconds() == expected + + +def test_update_longpoll_timeout_respects_opt_out(): + from app.channels.wechat import WechatChannel + + channel = WechatChannel( + bus=MessageBus(), + config={"bot_token": "bot-token", "polling_timeout": 37, "respect_server_longpoll_timeout": False}, + ) + + channel._update_longpoll_timeout({"longpolling_timeout_ms": 42_000}) + + assert channel._server_longpoll_timeout_seconds is None + assert channel._current_longpoll_timeout_seconds() == 37.0 + + +@pytest.mark.parametrize("invalid_timeout", [10**400, 1e308, 1], ids=["overflow", "huge-finite", "too-short"]) +def test_poll_loop_ignores_invalid_server_timeout(monkeypatch, invalid_timeout): + """Process the batch and reuse the last valid timeout on the next request.""" + from app.channels.wechat import WechatChannel + + async def go(): + post_calls: list[dict[str, Any]] = [] + handled: list[dict[str, Any]] = [] + + def _client_factory(*args, **kwargs): + return _MockAsyncClient( + post_calls=post_calls, + post_responses=[ + { + "ret": 0, + "msgs": [], + "get_updates_buf": "cursor-1", + "longpolling_timeout_ms": 42000, + }, + { + "ret": 0, + "msgs": [ + { + "message_type": 1, + "message_id": "msg-1", + "from_user_id": "wx-user-1", + "context_token": "ctx-1", + "item_list": [{"type": 1, "text_item": {"text": "hello"}}], + } + ], + "get_updates_buf": "cursor-2", + "longpolling_timeout_ms": invalid_timeout, + }, + {"ret": 0, "msgs": [{"message_id": "stop"}]}, + ], + **kwargs, + ) + + monkeypatch.setattr("app.channels.wechat.httpx.AsyncClient", _client_factory) + + channel = WechatChannel(bus=MessageBus(), config={"bot_token": "bot-token"}) + channel._running = True + + async def _fake_handle_update(raw): + handled.append(raw) + if raw["message_id"] == "stop": + channel._running = False + + channel._handle_update = _fake_handle_update # type: ignore[method-assign] + + await asyncio.wait_for(channel._poll_loop(), 3.0) + + assert [message["message_id"] for message in handled] == ["msg-1", "stop"] + assert channel._get_updates_buf == "cursor-2" + assert channel._server_longpoll_timeout_seconds == 42.0 + assert [call["json"]["get_updates_buf"] for call in post_calls] == ["", "cursor-1", "cursor-2"] + assert [call["timeout"] for call in post_calls] == [40.0, 47.0, 47.0] + + _run(go()) + + def test_poll_loop_one_bad_message_does_not_permanently_lose_its_siblings(monkeypatch, tmp_path: Path, caplog): """A single message that fails to process must not sink the rest of its batch.