Blocking SQLite calls stalling Discord gateway heartbeat

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #51890 fix(gateway): offload handoff-watcher SQLite calls to avoid blocking the async heartbeat (#40695)

  merged to main

  files: gateway/run.py, tests/gateway/test_handoff_watcher_async_db.py

  Show the diff

  • gateway/run.py

    @@ -5974,23 +5974,23 @@ async def _handoff_watcher(self, interval: float = 2.0) -> None:
                     if self._session_db is None:
                         await asyncio.sleep(interval)
                         continue
    -                pending = self._session_db.list_pending_handoffs()
    +                pending = await asyncio.to_thread(self._session_db.list_pending_handoffs)
                     for row in pending:
                         session_id = row.get("id")
                         if not session_id:
                             continue
    -                    if not self._session_db.claim_handoff(session_id):
    +                    if not await asyncio.to_thread(self._session_db.claim_handoff, session_id):
                             # Another tick or another gateway already claimed it.
                             continue
                         try:
                             await self._process_handoff(row)
    -                        self._session_db.complete_handoff(session_id)
    +                        await asyncio.to_thread(self._session_db.complete_handoff, session_id)
                         except Exception as exc:
                             logger.warning(
                                 "Handoff for session %s failed: %s",
                                 session_id, exc, exc_info=True,
                             )
    -                        self._session_db.fail_handoff(session_id, str(exc))
    +                        await asyncio.to_thread(self._session_db.fail_handoff, session_id, str(exc))
                 except asyncio.CancelledError:
                     raise
                 except Exception as exc:

  • tests/gateway/test_handoff_watcher_async_db.py

    @@ -0,0 +1,164 @@
    +"""Regression test for #40695 (salvage of keystone PR #40782).
    +
    +The Discord gateway heartbeat was stalling because the handoff watcher
    +(``GatewayRunner._handoff_watcher``) polled the synchronous, blocking
    +SQLite-backed ``SessionDB`` directly on the asyncio event loop every 2s
    +('Shard ID None heartbeat blocked for more than N seconds').
    +
    +The fix (mirroring PR #40782) wraps every blocking ``SessionDB`` call inside
    +the watcher loop in ``asyncio.to_thread(...)`` so the SQLite I/O runs on a
    +worker thread and never blocks the event loop / Discord heartbeat.
    +
    +These tests assert that behaviour contract. They are mutation-survivable:
    +reverting any ``asyncio.to_thread(self._session_db.<call>)`` wrap back to a
    +direct synchronous call on the loop makes the relevant assertion fail.
    +"""
    +
    +import asyncio
    +import types
    +
    +import pytest
    +
    +import gateway.run as run
    +
    +
    +class _RecordingSessionDB:
    +    """SessionDB stand-in that records the thread each method runs on.
    +
    +    If the watcher calls these methods directly on the event loop (the bug),
    +    they run on the loop thread. If they are wrapped in ``asyncio.to_thread``
    +    (the fix), they run on a *different* worker thread.
    +    """
    +
    +    def __init__(self, loop_thread_ident):
    +        self._loop_thread_ident = loop_thread_ident
    +        self.threads = {}
    +        self.calls = []
    +
    +    def _record(self, name):
    +        import threading
    +
    +        self.threads.setdefault(name, []).append(threading.get_ident())
    +        self.calls.append(name)
    +
    +    def ran_off_loop(self, name):
    +        """True iff every call to ``name`` ran on a non-loop thread."""
    +        idents = self.threads.get(name, [])
    +        return bool(idents) and all(i != self._loop_thread_ident for i in idents)
    +
    +    def list_pending_handoffs(self):
    +        self._record("list_pending_handoffs")
    +        return [{"id": "sess-1"}]
    +
    +    def claim_handoff(self, session_id):
    +        self._record("claim_handoff")
    +        return True
    +
    +    def complete_handoff(self, session_id):
    +        self._record("complete_handoff")
    +
    +    def fail_handoff(self, session_id, error):
    +        self._record("fail_handoff")
    +
    +
    +def _make_fake_runner(session_db, *, fail_process=False):
    +    """Build a minimal object that exposes exactly what the loop body touches."""
    +    fake = types.SimpleNamespace()
    +    fake._session_db = session_db
    +    # _running yields True for the first loop check, then False so the loop
    +    # exits after a single tick.
    +    states = iter([True, False])
    +
    +    class _Running:
    +        def __bool__(_self):
    +            try:
    +                return next(states)
    +            except StopIteration:
    +                return False
    +
    +    fake._running = _Running()
    +
    +    async def _process_handoff(row):
    +        if fail_process:
    +            raise RuntimeError("boom")
    +
    +    fake._process_handoff = _process_handoff
    +    return fake
    +
    +
    +async def _run_one_tick(fake, monkeypatch):
    +    """Run the watcher for a single tick with sleeps neutralised."""
    +
    +    async def _no_sleep(_seconds):
    +        return None
    +
    +    monkeypatch.setattr(run.asyncio, "sleep", _no_sleep)
    +    # Bind the real (patched) method onto our minimal stand-in.
    +    coro = run.GatewayRunner._handoff_watcher(fake, interval=0.0)
    +    await asyncio.wait_for(coro, timeout=5)
    +
    +
    [redacted:email]
    +async def test_watcher_offloads_db_calls_to_threads(monkeypatch):
    +    """The success path must run list_pending/claim/complete off the loop."""
    +    import threading
    +
    +    loop_ident = threading.get_ident()
    +    db = _RecordingSessionDB(loop_ident)
    +    fake = _make_fake_runner(db, fail_process=False)
    +
    +    await _run_one_tick(fake, monkeypatch)
    +
    +    # Sanity: the watcher actually exercised the calls this tick.
    +    assert "list_pending_handoffs" in db.calls
    +    assert "claim_handoff" in db.calls
    +    assert "complete_handoff" in db.calls
    +
    +    # Contract: each blocking SessionDB call ran on a worker thread, NOT the
    +    # asyncio event-loop thread. Reverting a to_thread wrap makes the
    +    # corresponding call run on the loop thread and this fails.
    +    assert db.ran_off_loop("list_pending_handoffs")
    +    assert db.ran_off_loop("claim_handoff")
    +    assert db.ran_off_loop("complete_handoff")
    +
    +
    [redacted:email]
    +async def test_watcher_offloads_fail_handoff_to_thread(monkeypatch):
    +    """The error path must run fail_handoff off the loop too."""
    +    import threading
    +
    +    loop_ident = threading.get_ident()
    +    db = _RecordingSessionDB(loop_ident)
    +    fake = _make_fake_runner(db, fail_process=True)
    +
    +    await _run_one_tick(fake, monkeypatch)
    +
    +    assert "fail_handoff" in db.calls
    +    assert db.ran_off_loop("fail_handoff")
    +
    +
    [redacted:email]
    +async def test_watcher_wraps_calls_via_asyncio_to_thread(monkeypatch):
    +    """Explicitly assert the offload goes through asyncio.to_thread.
    +
    +    Patches ``run.asyncio.to_thread`` and records which SessionDB callables
    +    were handed to it. Mutation-survivable: dropping any wrap removes its
    +    callable from the recorded set.
    +    """
    +    db = _RecordingSessionDB(loop_thread_ident=-1)
    +    fake = _make_fake_runner(db, fail_process=False)
    +
    +    wrapped = []
    +    real_to_thread = run.asyncio.to_thread
    +
    +    async def _spy_to_thread(func, *args, **kwargs):
    +        wrapped.append(getattr(func, "__name__", repr(func)))
    +        return await real_to_thread(func, *args, **kwargs)
    +
    +    monkeypatch.setattr(run.asyncio, "to_thread", _spy_to_thread)
    +
    +    await _run_one_tick(fake, monkeypatch)
    +
    +    assert "list_pending_handoffs" in wrapped
    +    assert "claim_handoff" in wrapped
    +    assert "complete_handoff" in wrapped

• #54752 perf(state): merge FTS5 segments + add handoff_state index to curb write-lock contention

  merged to main

  files: hermes_state.py, scripts/release.py, tests/test_hermes_state.py

• #60984 fix(gateway): offload channel directory session scans off the event loop

  merged to main

  files: gateway/channel_directory.py, scripts/release.py, tests/gateway/test_channel_directory.py

  Show the diff

  • gateway/channel_directory.py

    @@ -6,6 +6,7 @@
     action="list" and for resolving human-friendly channel names to numeric IDs.
     """
     
    +import asyncio
     import json
     import logging
     from datetime import datetime
    @@ -121,7 +122,7 @@ async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]:
         for platform, adapter in adapters.items():
             try:
                 if platform == Platform.DISCORD:
    -                platforms["discord"] = _build_discord(adapter)
    +                platforms["discord"] = await asyncio.to_thread(_build_discord, adapter)
                 elif platform == Platform.SLACK:
                     platforms["slack"] = await _build_slack(adapter)
             except Exception as e:
    @@ -142,7 +143,7 @@ async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]:
                 or plat_name not in adapter_platform_names
             ):
                 continue
    -        platforms[plat_name] = _build_from_sessions(plat_name)
    +        platforms[plat_name] = await asyncio.to_thread(_build_from_sessions, plat_name)
     
         # Include plugin-registered platforms (dynamic enum members aren't in
         # Platform.__members__, so the loop above misses them). Same
    @@ -156,7 +157,7 @@ async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]:
                     and entry.name not in platforms
                     and entry.name in adapter_platform_names
                 ):
    -                platforms[entry.name] = _build_from_sessions(entry.name)
    +                platforms[entry.name] = await asyncio.to_thread(_build_from_sessions, entry.name)
         except Exception:
             pass
     
    @@ -223,7 +224,7 @@ async def _build_slack(adapter) -> List[Dict[str, Any]]:
         """
         team_clients = getattr(adapter, "_team_clients", None) or {}
         if not team_clients:
    -        return _build_from_sessions("slack")
    +        return await asyncio.to_thread(_build_from_sessions, "slack")
     
         channels: List[Dict[str, Any]] = []
         seen_ids: set = set()
    @@ -267,7 +268,7 @@ async def _build_slack(adapter) -> List[Dict[str, Any]]:
                 continue
     
         # Merge in DM/group entries discovered from session history.
    -    for entry in _build_from_sessions("slack"):
    +    for entry in await asyncio.to_thread(_build_from_sessions, "slack"):
             if entry.get("id") not in seen_ids:
                 channels.append(entry)
                 seen_ids.add(entry.get("id"))

  • scripts/release.py

    @@ -300,6 +300,7 @@
         "[redacted:email]": "itsflownium",
         "[redacted:email]": "dirtyren",
         "[redacted:email]": "nullptr0807",  # PR #60956 salvage (gateway hygiene in-place compaction; #60947)
    +    "[redacted:email]": "embwl0x",  # PR #60810 salvage (channel directory offload)
         "[redacted:email]": "doncazper",
         "[redacted:email]": "alex107ivanov",
         "[redacted:email]": "rungmc357",

  • tests/gateway/test_channel_directory.py

    @@ -3,6 +3,7 @@
     import asyncio
     import json
     import os
    +import threading
     from types import SimpleNamespace
     from unittest.mock import AsyncMock, MagicMock, patch
     
    @@ -84,6 +85,85 @@ def broken_dump(data, fp, *args, **kwargs):
             assert result == previous
     
     
    +class TestBuildChannelDirectoryOffload:
    +    def test_discord_builder_runs_off_event_loop_thread(self, tmp_path):
    +        from gateway.config import Platform
    +
    +        cache_file = tmp_path / "channel_directory.json"
    +        loop_thread = threading.get_ident()
    +        builder_threads = []
    +
    +        def fake_build_discord(_adapter):
    +            builder_threads.append(threading.get_ident())
    +            return []
    +
    +        with patch("gateway.channel_directory._build_discord", side_effect=fake_build_discord), \
    +             patch("gateway.channel_directory.DIRECTORY_PATH", cache_file):
    +            asyncio.run(build_channel_directory({Platform.DISCORD: object()}))
    +
    +        assert builder_threads
    +        assert all(tid != loop_thread for tid in builder_threads)
    +
    +    def test_session_discovery_runs_off_event_loop_thread(self, tmp_path):
    +        from gateway.config import Platform
    +
    +        cache_file = tmp_path / "channel_directory.json"
    +        loop_thread = threading.get_ident()
    +        calls = []
    +
    +        def fake_build_from_sessions(platform_name):
    +            calls.append((platform_name, threading.get_ident()))
    +            return []
    +
    +        with patch("gateway.channel_directory._build_from_sessions", side_effect=fake_build_from_sessions), \
    +             patch("gateway.channel_directory.DIRECTORY_PATH", cache_file):
    +            asyncio.run(build_channel_directory({Platform.TELEGRAM: object()}))
    +
    +        assert [name for name, _ in calls] == ["telegram"]
    +        assert calls[0][1] != loop_thread
    +
    +    def test_plugin_session_discovery_runs_off_event_loop_thread(self, tmp_path):
    +        cache_file = tmp_path / "channel_directory.json"
    +        loop_thread = threading.get_ident()
    +        calls = []
    +        plugin_entry = SimpleNamespace(name="irc")
    +
    +        def fake_build_from_sessions(platform_name):
    +            calls.append((platform_name, threading.get_ident()))
    +            return []
    +
    +        with patch("gateway.channel_directory._build_from_sessions", side_effect=fake_build_from_sessions), \
    +             patch("gateway.channel_directory.DIRECTORY_PATH", cache_file), \
    +             patch(
    +                 "gateway.platform_registry.platform_registry.plugin_entries",
    +                 return_value=[plugin_entry],
    +             ):
    +            asyncio.run(build_channel_directory({"irc": object()}))
    +
    +        assert [name for name, _ in calls] == ["irc"]
    +        assert calls[0][1] != loop_thread
    +
    +    def test_slack_session_merge_runs_off_event_loop_thread(self):
    +        loop_thread = threading.get_ident()
    +        calls = []
    +
    +        class FakeSlackClient:
    +            async def users_conversations(self, **_kwargs):
    +                return {"ok": True, "channels": []}
    +
    +        def fake_build_from_sessions(platform_name):
    +            calls.append((platform_name, threading.get_ident()))
    +            return [{"id": "D1", "name": "Alice", "type": "dm"}]
    +
    +        adapter = SimpleNamespace(_team_clients={"T1": FakeSlackClient()})
    +        with patch("gateway.channel_directory._build_from_sessions", side_effect=fake_build_from_sessions):
    +            channels = asyncio.run(_build_slack(adapter))
    +
    +        assert channels == [{"id": "D1", "name": "Alice", "type": "dm"}]
    +        assert [name for name, _ in calls] == ["slack"]
    +        assert calls[0][1] != loop_thread
    +
    +
     class TestResolveChannelName:
         def _setup(self, tmp_path, platforms):
             cache_file = _write_directory(tmp_path, platforms)