diff --git a/docs/persistence.md b/docs/persistence.md index 5d8fffa..b583cc3 100644 --- a/docs/persistence.md +++ b/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 diff --git a/src/github_agent_bridge/reader.py b/src/github_agent_bridge/reader.py index 6a93978..5a6e37e 100644 --- a/src/github_agent_bridge/reader.py +++ b/src/github_agent_bridge/reader.py @@ -4,7 +4,9 @@ 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 @@ -12,6 +14,35 @@ 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: diff --git a/tests/test_reader.py b/tests/test_reader.py index 590eb37..37722cd 100644 --- a/tests/test_reader.py +++ b/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( + "", + "@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( + "", + "@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"