fix: keep IMAP reader alive during SQLite contention

docs/persistence.md

@@ -29,18 +29,21 @@ Backup and restore are deliberately excluded. They use SQLite's backup API in
 `autoupdate.py` and remain specialized database operations rather than
 repository methods.
 
-Executor heartbeat writes, acknowledgement claims and streamed session-event
-writes treat `SQLITE_BUSY` and `SQLITE_LOCKED` as transient contention after the
-connection timeout: they wait and retry instead of terminating the background
-thread. Stream readers enqueue activity before persistence so a blocked SQLite
-writer cannot stop draining the OpenClaw subprocess pipes. Once the main CLI
-process exits, readers get a bounded drain window before the bridge stops them
-and closes its pipe ends; a descendant that inherited stdout or stderr cannot
-hold the worker indefinitely. Acknowledgement retries stop before the external
-GitHub reaction, so ambiguous post-side-effect failures still surface. Other
-`OperationalError` failures also propagate so schema or storage faults are not
-hidden. Recovered heartbeat or session-event contention increments the worker's
-persisted `recent_error_count` on the next successful heartbeat.
+Executor heartbeat writes, acknowledgement claims, streamed session-event
+writes, and individual IMAP reader storage operations treat `SQLITE_BUSY` and
+`SQLITE_LOCKED` as transient contention after the connection timeout: they wait
+and retry instead of terminating the worker, reader pass, or background thread.
+Reader retries stay inside the failed SQLite operation, so they do not fetch the
+message again or repeat the IMAP `\Seen` side effect. Stream readers enqueue
+activity before persistence so a blocked SQLite writer cannot stop draining the
+OpenClaw subprocess pipes. Once the main CLI process exits, readers get a
+bounded drain window before the bridge stops them and closes its pipe ends; a
+descendant that inherited stdout or stderr cannot hold the worker indefinitely.
+Acknowledgement retries stop before the external GitHub reaction, so ambiguous
+post-side-effect failures still surface. Other `OperationalError` failures
+also propagate so schema or storage faults are not hidden. Recovered heartbeat
+or session-event contention increments the worker's persisted
+`recent_error_count` on the next successful heartbeat.
 
 ## Queue transaction boundaries
 

src/github_agent_bridge/reader.py

@@ -4,14 +4,45 @@
 import imaplib
 import sqlite3
 import time
+from collections.abc import Callable
 from dataclasses import dataclass
+from typing import TypeVar
 
 from .models import Notification
 from .parser import decode_header_value, extract_body_text, is_github_notification_message, parse_auth_results
 from .policy import Policy
 from .queue import JobQueue
 
 
+T = TypeVar("T")
+SQLITE_CONTENTION_RETRY_DELAYS = (1.0, 2.0)
+
+
+def _is_sqlite_contention_error(exc: sqlite3.OperationalError) -> bool:
+    error_code = getattr(exc, "sqlite_errorcode", None)
+    if isinstance(error_code, int) and error_code & 0xFF in {
+        sqlite3.SQLITE_BUSY,
+        sqlite3.SQLITE_LOCKED,
+    }:
+        return True
+    message = str(exc).lower()
+    return "database" in message and "locked" in message
+
+
+def _retry_sqlite_contention(operation: Callable[[], T]) -> T:
+    for attempt in range(len(SQLITE_CONTENTION_RETRY_DELAYS) + 1):
+        try:
+            return operation()
+        except sqlite3.OperationalError as exc:
+            if (
+                not _is_sqlite_contention_error(exc)
+                or attempt == len(SQLITE_CONTENTION_RETRY_DELAYS)
+            ):
+                raise
+            time.sleep(SQLITE_CONTENTION_RETRY_DELAYS[attempt])
+    raise AssertionError("unreachable")
+
+
 def imap_mailbox_arg(value: str) -> str:
     """Quote mailbox names with spaces for imaplib.select."""
     if " " in value and not value.startswith('"'):
@@ -52,7 +83,12 @@ def fetch_once(self) -> int:
         raise AssertionError("unreachable")
 
     def _fetch_once(self) -> int:
-        last_uid = int(self.queue.get_state("last_uid", "0") or 0)
+        last_uid = int(
+            _retry_sqlite_contention(
+                lambda: self.queue.get_state("last_uid", "0")
+            )
+            or 0
+        )
         count = 0
         imap = imaplib.IMAP4_SSL(self.config.host, self.config.port)
         try:
@@ -73,27 +109,36 @@ def _fetch_once(self) -> int:
                 if is_github_notification_message(msg, from_addr):
                     n = Notification(uid=uid, message_id=message_id, subject=subject, from_addr=from_addr, body=extract_body_text(msg), auth=parse_auth_results(msg))
                     try:
-                        self.queue.ingest(
-                            n,
-                            self.policy,
-                            source="email",
-                            source_key=n.message_id,
+                        _retry_sqlite_contention(
+                            lambda: self.queue.ingest(
+                                n,
+                                self.policy,
+                                source="email",
+                                source_key=n.message_id,
+                            )
                         )
                     except sqlite3.Error:
                         raise
                     except Exception as exc:
-                        self.queue.quarantine_notification(
-                            n,
-                            reason="ingestion_error",
-                            error=f"{type(exc).__name__}: {exc}",
-                            metadata={"uid": uid, "mailbox": self.config.mailbox},
+                        _retry_sqlite_contention(
+                            lambda: self.queue.quarantine_notification(
+                                n,
+                                reason="ingestion_error",
+                                error=f"{type(exc).__name__}: {exc}",
+                                metadata={
+                                    "uid": uid,
+                                    "mailbox": self.config.mailbox,
+                                },
+                            )
                         )
                     # Only GitHub notifications belong to this bounded context.
                     # Generic/non-GitHub mail must remain untouched for the generic inbox worker.
                     if self.mark_seen:
                         imap.uid("store", str(uid), "+FLAGS", "(\\Seen)")
                     count += 1
-                self.queue.set_state("last_uid", str(uid))
+                _retry_sqlite_contention(
+                    lambda: self.queue.set_state("last_uid", str(uid))
+                )
             return count
         finally:
             try:

tests/test_reader.py

@@ -168,34 +168,114 @@ def test_fetch_once_retries_transient_enqueue_storage_failure(monkeypatch, tmp_p
             "@pilipilisbot https://github.com/gisce/erp/issues/42#issuecomment-99",
         ),
     })
-    original_ingest = queue.ingest
+    monkeypatch.setattr(imaplib, "IMAP4_SSL", lambda *args: mailbox)
+    monkeypatch.setattr("github_agent_bridge.actors.github_actor_details_for_context", lambda ctx, *, gh_bin="gh": None)
+
+    queue.database.timeout_seconds = 0.01
+    queue.database.busy_timeout_ms = 10
+    locker = sqlite3.connect(db, isolation_level=None)
+    locker.execute("BEGIN IMMEDIATE")
+    released = {"value": False}
+    sleeps = []
+
+    def release_lock(delay):
+        sleeps.append(delay)
+        locker.commit()
+        locker.close()
+        released["value"] = True
+
+    monkeypatch.setattr(reader_module.time, "sleep", release_lock)
+
+    reader = ImapReader(
+        ImapConfig("imap.example.com", 993, "bot@example.com", "secret"),
+        queue,
+        Policy(trusted_orgs={"gisce"}, bot_logins={"pilipilisbot"}),
+    )
+
+    try:
+        assert reader.fetch_once() == 1
+    finally:
+        if not released["value"]:
+            locker.rollback()
+            locker.close()
+
+    assert sleeps == [1.0]
+    assert queue.get_state("last_uid") == "1"
+    assert queue.stats()["pending"] == 1
+    with sqlite3.connect(db) as con:
+        quarantined_count = con.execute("SELECT COUNT(*) FROM quarantined_notifications").fetchone()[0]
+
+    assert quarantined_count == 0
+
+
+def test_fetch_once_retries_state_advance_without_repeating_imap_effects(
+    monkeypatch, tmp_path
+):
+    db = tmp_path / "bridge.sqlite3"
+    queue = JobQueue(db)
+    mailbox = MailboxWithMessages({
+        1: github_message(
+            "<retry-state@github.com>",
+            "@pilipilisbot https://github.com/gisce/erp/issues/42#issuecomment-99",
+        ),
+    })
+    original_set_state = queue.set_state
     attempts = {"count": 0}
+    sleeps = []
 
-    def fail_once_then_ingest(notification, policy, **kwargs):
+    def fail_once_then_set_state(key, value):
         attempts["count"] += 1
         if attempts["count"] == 1:
             raise sqlite3.OperationalError("database is locked")
-        return original_ingest(notification, policy, **kwargs)
+        return original_set_state(key, value)
 
     monkeypatch.setattr(imaplib, "IMAP4_SSL", lambda *args: mailbox)
-    monkeypatch.setattr(queue, "ingest", fail_once_then_ingest)
+    monkeypatch.setattr(queue, "set_state", fail_once_then_set_state)
+    monkeypatch.setattr(reader_module.time, "sleep", sleeps.append)
     monkeypatch.setattr("github_agent_bridge.actors.github_actor_details_for_context", lambda ctx, *, gh_bin="gh": None)
 
     reader = ImapReader(
         ImapConfig("imap.example.com", 993, "bot@example.com", "secret"),
         queue,
         Policy(trusted_orgs={"gisce"}, bot_logins={"pilipilisbot"}),
+        mark_seen=True,
     )
 
-    with pytest.raises(sqlite3.OperationalError, match="database is locked"):
-        reader.fetch_once()
-
-    assert queue.get_state("last_uid", "0") == "0"
     assert reader.fetch_once() == 1
     assert attempts["count"] == 2
+    assert sleeps == [1.0]
+    assert len(mailbox.stores) == 1
     assert queue.get_state("last_uid") == "1"
     assert queue.stats()["pending"] == 1
-    with sqlite3.connect(db) as con:
-        quarantined_count = con.execute("SELECT COUNT(*) FROM quarantined_notifications").fetchone()[0]
 
-    assert quarantined_count == 0
+
+def test_fetch_once_does_not_retry_non_contention_storage_failure(
+    monkeypatch, tmp_path
+):
+    queue = JobQueue(tmp_path / "bridge.sqlite3")
+    mailbox = MailboxWithMessages({
+        1: github_message(
+            "<broken-storage@github.com>",
+            "@pilipilisbot https://github.com/gisce/erp/issues/42#issuecomment-99",
+        ),
+    })
+    sleeps = []
+
+    def fail_ingest(*args, **kwargs):
+        raise sqlite3.OperationalError("no such table: jobs")
+
+    monkeypatch.setattr(imaplib, "IMAP4_SSL", lambda *args: mailbox)
+    monkeypatch.setattr(queue, "ingest", fail_ingest)
+    monkeypatch.setattr(reader_module.time, "sleep", sleeps.append)
+
+    reader = ImapReader(
+        ImapConfig("imap.example.com", 993, "bot@example.com", "secret"),
+        queue,
+        Policy(trusted_orgs={"gisce"}, bot_logins={"pilipilisbot"}),
+    )
+
+    with pytest.raises(sqlite3.OperationalError, match="no such table"):
+        reader.fetch_once()
+
+    assert sleeps == []
+    assert queue.get_state("last_uid", "0") == "0"