diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py index 1c48309d20..76bb572ce4 100644 --- a/plugins/platforms/buzz/adapter.py +++ b/plugins/platforms/buzz/adapter.py @@ -559,6 +559,8 @@ class BuzzAdapter(BasePlatformAdapter): self._channel_state: Dict[str, dict] = {} # Cursors read from disk at connect(), consumed by each channel's first seed. self._restored_cursors: Dict[str, dict] = {} + # Orders off-loop cursor writes: each snapshot is taken under it, so an older one never lands last. + self._cursor_write_lock = asyncio.Lock() self._channel_names: Dict[str, str] = {} # channel_id -> raw ``channels list`` entry; drives DM-vs-channel classification. self._channel_meta: Dict[str, dict] = {} @@ -1255,8 +1257,8 @@ class BuzzAdapter(BasePlatformAdapter): seen = [str(event_id) for event_id in raw_seen][-_SEEN_CAP:] if isinstance(raw_seen, list) else [] self._restored_cursors[str(channel_id)] = {"chat_type": str(entry.get("chat_type") or ""), "last_ts": last_ts, "seen": seen} - def _save_cursors(self) -> None: - """Persist every watched channel's cursor. Never raises.""" + def _cursor_payload(self) -> dict: + """Snapshot of every watched channel's cursor (taken on the loop: ``_channel_state`` is loop-owned).""" channels = { channel_id: { "chat_type": state.get("chat_type") or "group", "last_ts": int(state.get("last_ts") or 0), @@ -1264,10 +1266,17 @@ class BuzzAdapter(BasePlatformAdapter): } for channel_id, state in self._channel_state.items() } - payload = {"identity": self._self_pubkey, "relay": self.relay_url, "channels": channels} + return {"identity": self._self_pubkey, "relay": self.relay_url, "channels": channels} + + def _save_cursors(self) -> None: + """Persist every watched channel's cursor. Never raises.""" + self._write_cursors(self._cursor_path(), self._cursor_payload()) + + @staticmethod + def _write_cursors(path: Path, payload: dict) -> None: try: from utils import atomic_json_write - atomic_json_write(self._cursor_path(), payload, indent=None) + atomic_json_write(path, payload, indent=None) except Exception: logger.debug("Buzz: could not persist channel cursors", exc_info=True) @@ -1382,7 +1391,9 @@ class BuzzAdapter(BasePlatformAdapter): await self._handle_event(channel_id, state, event) self._trim_seen(state) if self._cursor_mark(state) != before: - self._save_cursors() + # The write fsyncs + renames, and on the WebSocket transport this runs once per inbound event. + async with self._cursor_write_lock: + await asyncio.to_thread(self._write_cursors, self._cursor_path(), self._cursor_payload()) @staticmethod def _parse_imeta_attachments(event: dict) -> Tuple[List[dict], int]: diff --git a/tests/gateway/test_buzz_adapter.py b/tests/gateway/test_buzz_adapter.py index e62f7175f1..dc105edb1f 100644 --- a/tests/gateway/test_buzz_adapter.py +++ b/tests/gateway/test_buzz_adapter.py @@ -3555,3 +3555,27 @@ class TestChannelCursorPersistence: assert f"e{cap * 2 - 1}" in seen assert "e0" not in seen + @pytest.mark.asyncio + async def test_cursor_write_runs_off_the_event_loop(self, adapter, tmp_path, monkeypatch): + """The write fsyncs + renames once per inbound event on the WebSocket transport.""" + import threading + + import utils + + cli = await self._seed(adapter, _event("e1", created_at=100)) + real_write = utils.atomic_json_write + threads = [] + + def _record(*args, **kwargs): + threads.append(threading.get_ident()) + return real_write(*args, **kwargs) + + monkeypatch.setattr(utils, "atomic_json_write", _record) + cli.responses.clear() + cli.script("messages", "get", [_event("e1", created_at=100), _event("e2", created_at=200)]) + await adapter._poll_channel(CHANNEL) + + assert threads and threading.get_ident() not in threads + saved = json.loads(self._cursor_file(tmp_path).read_text(encoding="utf-8")) + assert saved["channels"][CHANNEL]["last_ts"] == 200 + assert saved["channels"][CHANNEL]["seen"] == ["e1", "e2"]