fix(buzz): persist channel cursors off the event loop

_handle_events called _save_cursors inline whenever a batch moved the
cursor. It ends in atomic_json_write (mkstemp + fsync + os.replace), and
on the WebSocket transport _handle_events runs once per inbound EVENT
frame, so every message paid an fsync on the gateway's event loop,
stalling every other adapter and in-flight turn for its duration.

Split the snapshot from the write: the payload is still built on the
loop (_channel_state is loop-owned), and the write goes through
asyncio.to_thread. The snapshot is taken under an asyncio.Lock so a
slower, older write can never land after a newer one and regress the
durable cursor. connect() keeps the synchronous _save_cursors.

(cherry picked from commit 299929741429262f497b86a67940ffb37faa1694)

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]:

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"]