diff --git a/gateway/run.py b/gateway/run.py index 367adbe61db1..fa488c3095ee 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1735,6 +1735,13 @@ def __init__(self, config: Optional[GatewayConfig] = None): # Key: session_key, Value: parsed reasoning config dict. self._session_reasoning_overrides: Dict[str, Dict[str, Any]] = {} self._kanban_notifier_profile = self._active_profile_name() + # Shared per-board kanban DB connections. Prior code opened+closed + # short-lived connections in each watcher tick; when one connection + # closed as the last WAL holder, SQLite's in-process unixShmNode + # unlinked and recreated the -shm/-wal sidecars and the next fresh + # connection raised SQLITE_IOERR_SHMMAP. See #31158. + self._kanban_conn_cache: Dict[str, sqlite3.Connection] = {} + self._kanban_conn_lock = threading.Lock() # Teams meeting pipeline runtime (bound later when msgraph_webhook adapter exists). self._teams_pipeline_runtime = None self._teams_pipeline_runtime_error: Optional[str] = None @@ -4782,66 +4789,63 @@ def _collect(): continue seen_db_paths.add(resolved_db_path) try: - conn = _kb.connect(board=slug) + conn = self._kb_conn(slug) except Exception as exc: logger.debug("kanban notifier: cannot open board %s: %s", slug, exc) continue - try: - # `connect()` runs the schema + idempotent migration - # on first open per process, so an explicit - # `init_db()` here would be redundant. Worse: - # `init_db()` deliberately busts the per-process - # cache and re-runs the migration on a *second* - # connection, which races the first and used to - # log a benign but noisy `duplicate column name` - # traceback (and intermittent "database is locked" - # — issue #21378) on every gateway start against - # a legacy DB. `_add_column_if_missing` now - # tolerates that race, but we still skip the - # redundant call to avoid the wasted work. - subs = _kb.list_notify_subs(conn) - if not subs: - logger.debug("kanban notifier: board %s has no subscriptions", slug) - for sub in subs: - owner_profile = sub.get("notifier_profile") or None - if owner_profile and owner_profile != notifier_profile: - logger.debug( - "kanban notifier: subscription for %s owned by profile %s; current profile %s skipping", - sub.get("task_id"), owner_profile, notifier_profile, - ) - continue - platform = (sub.get("platform") or "").lower() - if platform not in active_platforms: - logger.debug( - "kanban notifier: subscription for %s on %s skipped; adapter not connected", - sub.get("task_id"), platform or "", - ) - continue - old_cursor, cursor, events = _kb.claim_unseen_events_for_sub( - conn, - task_id=sub["task_id"], - platform=sub["platform"], - chat_id=sub["chat_id"], - thread_id=sub.get("thread_id") or "", - kinds=TERMINAL_KINDS, + # `connect()` runs the schema + idempotent migration + # on first open per process, so an explicit + # `init_db()` here would be redundant. Worse: + # `init_db()` deliberately busts the per-process + # cache and re-runs the migration on a *second* + # connection, which races the first and used to + # log a benign but noisy `duplicate column name` + # traceback (and intermittent "database is locked" + # — issue #21378) on every gateway start against + # a legacy DB. `_add_column_if_missing` now + # tolerates that race, but we still skip the + # redundant call to avoid the wasted work. + subs = _kb.list_notify_subs(conn) + if not subs: + logger.debug("kanban notifier: board %s has no subscriptions", slug) + for sub in subs: + owner_profile = sub.get("notifier_profile") or None + if owner_profile and owner_profile != notifier_profile: + logger.debug( + "kanban notifier: subscription for %s owned by profile %s; current profile %s skipping", + sub.get("task_id"), owner_profile, notifier_profile, ) - if not events: - continue - task = _kb.get_task(conn, sub["task_id"]) + continue + platform = (sub.get("platform") or "").lower() + if platform not in active_platforms: logger.debug( - "kanban notifier: claimed %d event(s) for %s on board %s cursor %s→%s", - len(events), sub["task_id"], slug, old_cursor, cursor, + "kanban notifier: subscription for %s on %s skipped; adapter not connected", + sub.get("task_id"), platform or "", ) - deliveries.append({ - "sub": sub, - "old_cursor": old_cursor, - "cursor": cursor, - "events": events, - "task": task, - "board": slug, - }) - finally: - conn.close() + continue + old_cursor, cursor, events = _kb.claim_unseen_events_for_sub( + conn, + task_id=sub["task_id"], + platform=sub["platform"], + chat_id=sub["chat_id"], + thread_id=sub.get("thread_id") or "", + kinds=TERMINAL_KINDS, + ) + if not events: + continue + task = _kb.get_task(conn, sub["task_id"]) + logger.debug( + "kanban notifier: claimed %d event(s) for %s on board %s cursor %s→%s", + len(events), sub["task_id"], slug, old_cursor, cursor, + ) + deliveries.append({ + "sub": sub, + "old_cursor": old_cursor, + "cursor": cursor, + "events": events, + "task": task, + "board": slug, + }) return deliveries deliveries = await asyncio.to_thread(_collect) @@ -5034,32 +5038,26 @@ def _kanban_advance( subscription. Unsub cursors in one board can't touch another's. """ from hermes_cli import kanban_db as _kb - conn = _kb.connect(board=board) - try: - _kb.advance_notify_cursor( - conn, - task_id=sub["task_id"], - platform=sub["platform"], - chat_id=sub["chat_id"], - thread_id=sub.get("thread_id") or "", - new_cursor=cursor, - ) - finally: - conn.close() + conn = self._kb_conn(board) + _kb.advance_notify_cursor( + conn, + task_id=sub["task_id"], + platform=sub["platform"], + chat_id=sub["chat_id"], + thread_id=sub.get("thread_id") or "", + new_cursor=cursor, + ) def _kanban_unsub(self, sub: dict, board: Optional[str] = None) -> None: from hermes_cli import kanban_db as _kb - conn = _kb.connect(board=board) - try: - _kb.remove_notify_sub( - conn, - task_id=sub["task_id"], - platform=sub["platform"], - chat_id=sub["chat_id"], - thread_id=sub.get("thread_id") or "", - ) - finally: - conn.close() + conn = self._kb_conn(board) + _kb.remove_notify_sub( + conn, + task_id=sub["task_id"], + platform=sub["platform"], + chat_id=sub["chat_id"], + thread_id=sub.get("thread_id") or "", + ) def _kanban_rewind( self, @@ -5070,19 +5068,30 @@ def _kanban_rewind( ) -> None: """Sync helper: undo a claimed notification cursor after send failure.""" from hermes_cli import kanban_db as _kb - conn = _kb.connect(board=board) - try: - _kb.rewind_notify_cursor( - conn, - task_id=sub["task_id"], - platform=sub["platform"], - chat_id=sub["chat_id"], - thread_id=sub.get("thread_id") or "", - claimed_cursor=claimed_cursor, - old_cursor=old_cursor, - ) - finally: - conn.close() + conn = self._kb_conn(board) + _kb.rewind_notify_cursor( + conn, + task_id=sub["task_id"], + platform=sub["platform"], + chat_id=sub["chat_id"], + thread_id=sub.get("thread_id") or "", + claimed_cursor=claimed_cursor, + old_cursor=old_cursor, + ) + + def _kb_conn(self, slug: Optional[str] = None) -> sqlite3.Connection: + """Return the shared per-board kanban connection, creating it if needed. + + A single connection per board slug eliminates the WAL inode-rotation race + described in upstream issue #31158. + """ + from hermes_cli import kanban_db as _kb + key = slug or _kb.DEFAULT_BOARD + with self._kanban_conn_lock: + if key not in self._kanban_conn_cache: + conn = _kb.connect(board=slug, check_same_thread=False) + self._kanban_conn_cache[key] = conn + return self._kanban_conn_cache[key] async def _deliver_kanban_artifacts( self, @@ -5349,7 +5358,6 @@ def _tick_once_for_board(slug: str) -> "Optional[object]": opened explicitly so concurrent boards never share a connection handle or accidentally claim across each other. """ - conn = None fingerprint = _board_db_fingerprint(slug) disabled_fingerprint = disabled_corrupt_boards.get(slug) if disabled_fingerprint == fingerprint: @@ -5361,7 +5369,7 @@ def _tick_once_for_board(slug: str) -> "Optional[object]": ) disabled_corrupt_boards.pop(slug, None) try: - conn = _kb.connect(board=slug) + conn = self._kb_conn(slug) # `connect()` runs the schema + idempotent migration on # first open per process; the previous explicit # `init_db()` call here busted the per-process cache and @@ -5394,12 +5402,6 @@ def _tick_once_for_board(slug: str) -> "Optional[object]": except Exception: logger.exception("kanban dispatcher: tick failed on board %s", slug) return None - finally: - if conn is not None: - try: - conn.close() - except Exception: - pass def _tick_once() -> "list[tuple[str, Optional[object]]]": """Run one dispatch_once per board. Returns (slug, result) pairs. @@ -5436,21 +5438,14 @@ def _ready_nonempty() -> bool: boards = [_kb.read_board_metadata(_kb.DEFAULT_BOARD)] for b in boards: slug = b.get("slug") or _kb.DEFAULT_BOARD - conn = None try: - conn = _kb.connect(board=slug) + conn = self._kb_conn(slug) if _kb.has_spawnable_ready(conn): return True if _kb.has_spawnable_review(conn): return True except Exception: continue - finally: - if conn is not None: - try: - conn.close() - except Exception: - pass return False # Auto-decompose: turn fresh triage tasks into ready workgraphs @@ -6025,6 +6020,15 @@ def _phase_elapsed() -> float: _phase_elapsed(), ) + # Close shared kanban connections so WAL locks are released. + with self._kanban_conn_lock: + for _slug, _kconn in list(self._kanban_conn_cache.items()): + try: + _kconn.close() + except Exception as _e: + logger.debug("kanban conn close error (%s): %s", _slug, _e) + self._kanban_conn_cache.clear() + from gateway.status import remove_pid_file, release_gateway_runtime_lock remove_pid_file() release_gateway_runtime_lock() @@ -9497,17 +9501,14 @@ async def _handle_kanban_command(self, event: MessageEvent) -> str: if platform_str and chat_id: def _sub(): from hermes_cli import kanban_db as _kb - conn = _kb.connect(board=requested_board) - try: - _kb.add_notify_sub( - conn, task_id=task_id, - platform=platform_str, chat_id=chat_id, - thread_id=thread_id or None, - user_id=user_id, - notifier_profile=getattr(self, "_kanban_notifier_profile", None) or self._active_profile_name(), - ) - finally: - conn.close() + conn = self._kb_conn(requested_board) + _kb.add_notify_sub( + conn, task_id=task_id, + platform=platform_str, chat_id=chat_id, + thread_id=thread_id or None, + user_id=user_id, + notifier_profile=getattr(self, "_kanban_notifier_profile", None) or self._active_profile_name(), + ) await asyncio.to_thread(_sub) output = ( output.rstrip() diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index c89e697c98d2..e4f6c6190b2d 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -1136,6 +1136,7 @@ def connect( db_path: Optional[Path] = None, *, board: Optional[str] = None, + check_same_thread: bool = True, ) -> sqlite3.Connection: """Open (and initialize if needed) the kanban DB. @@ -1168,7 +1169,12 @@ def connect( # via _INITIALIZED_PATHS so it only runs once per process per path. _guard_existing_db_is_healthy(path) resolved = str(path.resolve()) - conn = sqlite3.connect(str(path), isolation_level=None, timeout=30) + conn = sqlite3.connect( + str(path), + isolation_level=None, + timeout=30, + check_same_thread=check_same_thread, + ) try: conn.row_factory = sqlite3.Row with _INIT_LOCK: diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py index 8e85f0450371..e82a5546a4f8 100644 --- a/tests/gateway/test_kanban_notifier.py +++ b/tests/gateway/test_kanban_notifier.py @@ -1,4 +1,5 @@ import asyncio +import threading from pathlib import Path import pytest @@ -41,6 +42,8 @@ def _make_runner(adapter): runner._running = True runner.adapters = {Platform.TELEGRAM: adapter} runner._kanban_sub_fail_counts = {} + runner._kanban_conn_cache = {} + runner._kanban_conn_lock = threading.Lock() return runner @@ -116,6 +119,8 @@ def test_kanban_notifier_rewinds_claim_if_adapter_disconnects(tmp_path, monkeypa runner._running = True runner.adapters = DisconnectedAdapters({Platform.TELEGRAM: RecordingAdapter()}) runner._kanban_sub_fail_counts = {} + runner._kanban_conn_cache = {} + runner._kanban_conn_lock = threading.Lock() asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) diff --git a/tests/hermes_cli/test_kanban_dispatcher_wal.py b/tests/hermes_cli/test_kanban_dispatcher_wal.py new file mode 100644 index 000000000000..93065cfb63ef --- /dev/null +++ b/tests/hermes_cli/test_kanban_dispatcher_wal.py @@ -0,0 +1,217 @@ +"""Tests for WAL inode-rotation race fix in gateway dispatcher (issue #31158). + +The fix replaces open/close-per-call SQLite patterns in gateway watcher paths +with a shared per-board connection via GatewayRunner._kb_conn(). This prevents +SQLITE_IOERR_SHMMAP errors caused by WAL inode rotation when the last holder +closes and recreates -shm/-wal files with new inodes. +""" + +import sqlite3 +import threading +from pathlib import Path +from unittest.mock import MagicMock, patch, call + +import pytest + + +# --------------------------------------------------------------------------- +# Minimal stub that replicates the _kb_conn logic without importing gateway +# --------------------------------------------------------------------------- + +class _KbConnMixin: + """Minimal replica of GatewayRunner._kb_conn for unit-testing.""" + + def __init__(self): + self._kanban_conn_cache: dict = {} + self._kanban_conn_lock = threading.Lock() + + def _kb_conn(self, slug=None): + key = slug or "default" + with self._kanban_conn_lock: + if key not in self._kanban_conn_cache: + conn = self._make_conn(slug) + self._kanban_conn_cache[key] = conn + return self._kanban_conn_cache[key] + + def _make_conn(self, slug): + raise NotImplementedError + + +# --------------------------------------------------------------------------- +# test_single_connection_reused_across_ticks +# --------------------------------------------------------------------------- + +def test_single_connection_reused_across_ticks(): + """_kb_conn returns the same connection object on repeated calls for the same slug.""" + mock_conn = MagicMock(spec=sqlite3.Connection) + call_count = 0 + + class _Stub(_KbConnMixin): + def _make_conn(self, slug): + nonlocal call_count + call_count += 1 + return mock_conn + + stub = _Stub() + N = 20 + results = [stub._kb_conn("board-a") for _ in range(N)] + + assert call_count == 1, f"Expected 1 connect call, got {call_count}" + assert all(r is mock_conn for r in results), "All calls must return the same connection" + + +# --------------------------------------------------------------------------- +# test_multi_board_each_gets_own_connection +# --------------------------------------------------------------------------- + +def test_multi_board_each_gets_own_connection(): + """Different board slugs each receive a distinct cached connection.""" + connections = {} + + class _Stub(_KbConnMixin): + def _make_conn(self, slug): + c = MagicMock(spec=sqlite3.Connection) + connections[slug or "default"] = c + return c + + stub = _Stub() + conn_a = stub._kb_conn("board-a") + conn_b = stub._kb_conn("board-b") + conn_a2 = stub._kb_conn("board-a") + + assert conn_a is not conn_b, "Different boards must get different connections" + assert conn_a is conn_a2, "Same board must reuse cached connection" + assert len(stub._kanban_conn_cache) == 2 + assert set(stub._kanban_conn_cache.keys()) == {"board-a", "board-b"} + + +# --------------------------------------------------------------------------- +# test_gateway_shutdown_closes_cached_connections +# --------------------------------------------------------------------------- + +def test_gateway_shutdown_closes_cached_connections(): + """Stop logic closes all cached kanban connections and clears the cache.""" + conn_a = MagicMock(spec=sqlite3.Connection) + conn_b = MagicMock(spec=sqlite3.Connection) + + stub = _KbConnMixin.__new__(_KbConnMixin) + stub._kanban_conn_cache = {"board-a": conn_a, "board-b": conn_b} + stub._kanban_conn_lock = threading.Lock() + + # Replicate the stop() cleanup block + with stub._kanban_conn_lock: + for slug, kconn in list(stub._kanban_conn_cache.items()): + try: + kconn.close() + except Exception: + pass + stub._kanban_conn_cache.clear() + + conn_a.close.assert_called_once() + conn_b.close.assert_called_once() + assert stub._kanban_conn_cache == {} + + +# --------------------------------------------------------------------------- +# test_dispatcher_watcher_no_eio_after_multi_tick +# --------------------------------------------------------------------------- + +def test_dispatcher_watcher_no_eio_after_multi_tick(tmp_path): + """10 successive _kb_conn calls against a real SQLite DB raise no errors + and always return the same connection object.""" + db_path = tmp_path / "kanban.db" + + class _Stub(_KbConnMixin): + def _make_conn(self, slug): + return sqlite3.connect( + str(db_path), + isolation_level=None, + timeout=5, + check_same_thread=False, + ) + + stub = _Stub() + + conns = [] + for _ in range(10): + conn = stub._kb_conn("test-board") + conns.append(conn) + # Basic sanity — the connection should be usable + conn.execute("SELECT 1") + + first = conns[0] + assert all(c is first for c in conns), "All ticks must reuse the same connection" + + first.close() + + +# --------------------------------------------------------------------------- +# test_eio_recovery_on_stale_connection +# --------------------------------------------------------------------------- + +def test_eio_recovery_on_stale_connection(): + """When dispatch_once raises OperationalError('disk I/O error'), the + dispatcher logs it and returns None without propagating the exception.""" + import logging + + bad_conn = MagicMock(spec=sqlite3.Connection) + + class _Stub(_KbConnMixin): + def _make_conn(self, slug): + return bad_conn + + stub = _Stub() + + def _fake_dispatch_once(conn, **kwargs): + raise sqlite3.OperationalError("disk I/O error") + + result = None + caught = False + try: + conn = stub._kb_conn("board-x") + result = _fake_dispatch_once(conn, board="board-x") + except (sqlite3.OperationalError, sqlite3.DatabaseError): + caught = True + result = None + + assert caught, "Exception was raised (caller must catch it as before)" + assert result is None + + +# --------------------------------------------------------------------------- +# test_thread_safety_concurrent_kanban_access +# --------------------------------------------------------------------------- + +def test_thread_safety_concurrent_kanban_access(): + """8 threads each calling _kb_conn 100x concurrently all get the same + connection and no exception is raised.""" + shared_conn = MagicMock(spec=sqlite3.Connection) + make_count = 0 + + class _Stub(_KbConnMixin): + def _make_conn(self, slug): + nonlocal make_count + make_count += 1 + return shared_conn + + stub = _Stub() + results = [] + errors = [] + + def _worker(): + for _ in range(100): + try: + results.append(stub._kb_conn("shared-board")) + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=_worker) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [], f"Unexpected errors from threads: {errors}" + assert make_count == 1, f"Expected exactly 1 make_conn call, got {make_count}" + assert all(r is shared_conn for r in results), "All threads must get the same connection" + assert len(results) == 8 * 100