fix(weixin): persist the long-poll cursor off the event loop, only when it moves
_poll_loop called _save_sync_buf inline after every getUpdates response. It ends in atomic_json_write (mkstemp + fsync + os.replace), so each poll cycle blocked the gateway's event loop for the duration of an fsync, and it did so even when nothing changed: an empty long-poll, and the timeout sentinel from _get_updates, echo the current buffer back, so the same value was rewritten every cycle. Write only when the buffer differs from the one in memory, and dispatch the write with asyncio.to_thread. The loop awaits it before the next poll, so writes stay serialized. (cherry picked from commit 316b775941ffcc7e1bdd17c044e879cbca1eb14f)
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user