From f6ccfa6bd37961463af35da6b217713d571c2ae0 Mon Sep 17 00:00:00 2001 From: AlexFucuson9 Date: Mon, 6 Jul 2026 08:10:26 +0700 Subject: [PATCH] fix(gateway): isolate per-subscription failures in kanban notifier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The kanban notifier _collect() loop iterates subscriptions without per-subscription error handling. When claim_unseen_events_for_sub raises for one subscription (e.g. DB corruption, lock contention), the entire tick aborts ā€” silently blocking delivery for ALL other subscriptions. Wrap the per-subscription logic in try/except so one bad subscription logs a warning and continues to the next, instead of jamming the entire notifier. Closes #59269 --- gateway/kanban_watchers.py | 79 +++++++++++++++------------ tests/gateway/test_kanban_notifier.py | 43 +++++++++++++++ 2 files changed, 87 insertions(+), 35 deletions(-) diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index eef928650ef6..baa59fd9661a 100644 --- a/gateway/kanban_watchers.py +++ b/gateway/kanban_watchers.py @@ -252,45 +252,54 @@ class GatewayKanbanWatchersMixin: 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: - _owner_adapters = getattr(self, "_profile_adapters", {}).get(owner_profile) - if not _owner_adapters: + try: + owner_profile = sub.get("notifier_profile") or None + if owner_profile and owner_profile != notifier_profile: + _owner_adapters = getattr(self, "_profile_adapters", {}).get(owner_profile) + if not _owner_adapters: + logger.debug( + "kanban notifier: subscription for %s owned by profile %s; current profile %s has no adapter for it, 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 owned by profile %s; current profile %s has no adapter for it, skipping", - sub.get("task_id"), owner_profile, notifier_profile, + "kanban notifier: subscription for %s on %s skipped; adapter not connected", + sub.get("task_id"), platform or "", ) 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 "", + 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, + }) + except Exception as sub_exc: + # Isolate per-subscription failures so one + # bad subscription cannot block delivery for + # all other subscriptions in this tick. + logger.warning( + "kanban notifier: subscription for %s on board %s failed: %s", + sub.get("task_id"), slug, sub_exc, ) - 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, - }) finally: conn.close() return deliveries diff --git a/tests/gateway/test_kanban_notifier.py b/tests/gateway/test_kanban_notifier.py index b5a51d113a6c..45a22cb78a1a 100644 --- a/tests/gateway/test_kanban_notifier.py +++ b/tests/gateway/test_kanban_notifier.py @@ -480,3 +480,46 @@ def _unseen_terminal_events_for(tid, chat_id): return events finally: conn.close() + + +def test_kanban_notifier_isolates_per_subscription_failure(tmp_path, monkeypatch): + """One bad subscription must not block delivery for all others. + + Regression for #59269: when claim_unseen_events_for_sub raises for one + subscription, the entire notifier tick used to abort ā€” silently blocking + delivery for every other subscription. + """ + db_path = tmp_path / "isolation.db" + monkeypatch.setenv("HERMES_KANBAN_DB", str(db_path)) + kb.init_db() + + # Create two tasks with subscriptions and complete both. + conn = kb.connect() + try: + tid_good = kb.create_task(conn, title="good task", assignee="worker") + kb.add_notify_sub(conn, task_id=tid_good, platform="telegram", chat_id="chat-good") + kb.complete_task(conn, tid_good, summary="done") + + tid_bad = kb.create_task(conn, title="bad task", assignee="worker") + kb.add_notify_sub(conn, task_id=tid_bad, platform="telegram", chat_id="chat-bad") + kb.complete_task(conn, tid_bad, summary="done") + finally: + conn.close() + + original_claim = kb.claim_unseen_events_for_sub + + def selective_claim(conn, task_id, **kwargs): + if task_id == tid_bad: + raise RuntimeError("simulated DB corruption for bad task") + return original_claim(conn, task_id=task_id, **kwargs) + + monkeypatch.setattr(kb, "claim_unseen_events_for_sub", selective_claim) + + adapter = RecordingAdapter() + runner = _make_runner(adapter) + + asyncio.run(_run_one_notifier_tick(monkeypatch, runner)) + + # The good task must still be delivered despite the bad task failing. + assert len(adapter.sent) == 1 + assert tid_good in adapter.sent[0]["text"]