diff --git a/gateway/platforms/weixin.py b/gateway/platforms/weixin.py index 7c5ef23ca2..dfdda2e1f3 100644 --- a/gateway/platforms/weixin.py +++ b/gateway/platforms/weixin.py @@ -823,11 +823,15 @@ class WeixinAdapter(OwnAccessPolicyMixin, BasePlatformAdapter): consecutive_failures = await backoff() continue consecutive_failures = 0 - if response.get("get_updates_buf"): - sync_buf = str(response["get_updates_buf"]) - _save_sync_buf(self._hermes_home, self._account_id, sync_buf) + # Dispatch before persisting: the off-loop write is an await, and a disconnect that + # cancels it must not leave the advanced cursor on disk with this batch undelivered. for message in response.get("msgs") or []: asyncio.create_task(self._process_message_safe(message)) + # atomic_json_write fsyncs + renames: persist off the loop, and only when the cursor + # moved (an empty long-poll echoes the same buffer back every cycle). + if response.get("get_updates_buf") and str(response["get_updates_buf"]) != sync_buf: + sync_buf = str(response["get_updates_buf"]) + await asyncio.to_thread(_save_sync_buf, self._hermes_home, self._account_id, sync_buf) except asyncio.CancelledError: break except Exception as exc: diff --git a/tests/gateway/test_weixin.py b/tests/gateway/test_weixin.py index a63dc89b08..b8a8c459c9 100644 --- a/tests/gateway/test_weixin.py +++ b/tests/gateway/test_weixin.py @@ -690,6 +690,49 @@ class TestWeixinApiTimeout: assert result == {"ret": 0, "msgs": [], "get_updates_buf": "buf-123"} +class TestWeixinPollLoopSyncBuf: + """The long-poll cursor write (fsync + rename) must not run on the event loop.""" + + def _run_polls(self, monkeypatch, buffers): + import threading + + adapter = _make_adapter() + adapter._running = True + adapter._poll_session = Mock() + responses = iter(buffers) + saves = [] + + async def _get_updates(*args, **kwargs): + try: + return {"ret": 0, "msgs": [], "get_updates_buf": next(responses)} + except StopIteration: + adapter._running = False + return {"ret": 0, "msgs": []} + + def _save(hermes_home, account_id, sync_buf): + saves.append((sync_buf, threading.get_ident())) + + monkeypatch.setattr(weixin, "_get_updates", _get_updates) + monkeypatch.setattr(weixin, "_load_sync_buf", lambda *a: "buf-0") + monkeypatch.setattr(weixin, "_save_sync_buf", _save) + + async def scenario(): + await adapter._poll_loop() + return threading.get_ident() + + return saves, asyncio.run(scenario()) + + def test_cursor_write_runs_off_the_loop_thread(self, monkeypatch): + saves, loop_thread = self._run_polls(monkeypatch, ["buf-1"]) + assert [buf for buf, _ in saves] == ["buf-1"] + assert all(thread != loop_thread for _, thread in saves) + + def test_unchanged_cursor_is_not_rewritten(self, monkeypatch): + # Empty long-polls (and the timeout sentinel) echo the current buffer back. + saves, _ = self._run_polls(monkeypatch, ["buf-0", "buf-1", "buf-1", "buf-2"]) + assert [buf for buf, _ in saves] == ["buf-1", "buf-2"] + + class TestWeixinVoiceAlwaysDownloaded: """Regression tests for #27300: when WeChat (Weixin) returns a ``voice_item.text`` (Tencent Cloud's STT) we must still download