Skip to content

fix(gateway): cache kanban DB connections per OS thread in GatewayRunner #32322

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
steveonjava wants to merge 3 commits into from
Closed

fix(gateway): cache kanban DB connections per OS thread in GatewayRunner #32322

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
de2dd4e
fix(gateway): cache kanban DB connections per OS thread in GatewayRunner
steveonjava May 25, 2026
8634043
fix(gateway): evict stale closed connections from TLS cache
steveonjava May 25, 2026
1feea4b
chore(release): add steveonjava to AUTHOR_MAP
steveonjava May 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
183 changes: 94 additions & 89 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1861,6 +1861,39 @@ def __init__(self, config: Optional[GatewayConfig] = None):
# Track background tasks to prevent garbage collection mid-execution
self._background_tasks: set = set()

# Per-thread SQLite connection cache for kanban DB access.
# Each asyncio.to_thread worker gets its own long-lived connection via
# threading.local() — avoids b-tree corruption from shared pager state.
self._kb_tls: threading.local = threading.local()

def _kb_conn(self, slug: Optional[str] = None) -> "sqlite3.Connection":
"""Return a per-thread cached SQLite connection for the given board slug.

Each OS thread (including asyncio.to_thread workers) gets its own
connection. Connections are created on first access per thread and
kept alive for the thread's lifetime, so the SQLite WAL pager state
is never shared across threads (which causes b-tree corruption).
If a cached connection was closed externally, it is evicted and a fresh
one is opened — callers must not close the returned connection.
"""
import sqlite3
from hermes_cli import kanban_db as _kb
key = slug or "default"
cache = getattr(self._kb_tls, "cache", None)
if cache is None:
cache = {}
self._kb_tls.cache = cache
conn = cache.get(key)
if conn is not None:
# Detect a closed connection and evict it so we open a fresh one.
try:
conn.execute("SELECT 1")
except sqlite3.ProgrammingError:
del cache[key]
conn = None
if conn is None:
cache[key] = _kb.connect(board=slug)
return cache[key]

def _wire_teams_pipeline_runtime(self) -> None:
"""Bind the Teams meeting pipeline runtime to Graph webhook ingress.
Expand Down Expand Up @@ -4868,66 +4901,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 "<missing>",
)
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 "<missing>",
)
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)
Expand Down Expand Up @@ -5120,32 +5150,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 = 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,
)
finally:
conn.close()

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 = 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 "",
)
finally:
conn.close()

def _kanban_rewind(
self,
Expand All @@ -5156,9 +5180,8 @@ 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 = self._kb_conn(board)
_kb.rewind_notify_cursor(
conn,
task_id=sub["task_id"],
platform=sub["platform"],
Expand All @@ -5167,8 +5190,6 @@ def _kanban_rewind(
claimed_cursor=claimed_cursor,
old_cursor=old_cursor,
)
finally:
conn.close()

async def _deliver_kanban_artifacts(
self,
Expand Down Expand Up @@ -5447,7 +5468,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
Expand Down Expand Up @@ -5480,12 +5501,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.
Expand Down Expand Up @@ -5522,21 +5537,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
Expand Down Expand Up @@ -9600,17 +9608,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 = 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(),
)
finally:
conn.close()
await asyncio.to_thread(_sub)
output = (
output.rstrip()
Expand Down
1 change: 1 addition & 0 deletions scripts/release.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1027,6 +1027,7 @@
"lisanhu2014@hotmail.com": "lisanhu",
"0668001438@zte.com.cn": "chenyunbo411",
"steven_chanin@alum.mit.edu": "stevenchanin",
"steveonjava@gmail.com": "steveonjava",
"fiver@example.com": "halmisen",
"mayq0422@gmail.com": "yuqianma",
"yuqian@zmetasoft.com": "yuqianma",
Expand Down
Loading