fix: keep authored actions responsive under ledger contention (#248)

src/idempotency.py

@@ -12,6 +12,10 @@ class Reservation:
     response: Any = None
 
 
+class IdempotencyLedgerBusy(RuntimeError):
+    """Raised when a durable completion cannot acquire the SQLite write lock."""
+
+
 class IdempotencyLedger:
     """Small durable ledger for replaying successful Gitea mutations."""
 
@@ -21,18 +25,22 @@ class IdempotencyLedger:
         *,
         ttl_seconds: float,
         max_entries: int,
+        lock_timeout_seconds: float = 0.1,
         clock: Callable[[], float] = time.time,
     ) -> None:
         self.path = Path(path)
         self.ttl_seconds = ttl_seconds
         self.max_entries = max_entries
+        self.lock_timeout_seconds = lock_timeout_seconds
         self.clock = clock
         self.path.parent.mkdir(parents=True, exist_ok=True)
         self._initialize()
 
     def _connect(self) -> sqlite3.Connection:
-        connection = sqlite3.connect(self.path, timeout=5)
-        connection.execute("PRAGMA busy_timeout = 5000")
+        connection = sqlite3.connect(self.path, timeout=self.lock_timeout_seconds)
+        connection.execute(
+            f"PRAGMA busy_timeout = {max(1, int(self.lock_timeout_seconds * 1000))}"
+        )
         return connection
 
     def _initialize(self) -> None:
@@ -61,43 +69,48 @@ class IdempotencyLedger:
     def reserve(self, key: str, fingerprint: tuple[Any, ...]) -> Reservation:
         encoded = self._fingerprint(fingerprint)
         now = self.clock()
-        with self._connect() as connection:
-            connection.execute("BEGIN IMMEDIATE")
-            connection.execute(
-                "DELETE FROM idempotency_operations "
-                "WHERE status = 'completed' AND completed_at <= ?",
-                (now - self.ttl_seconds,),
-            )
-            row = connection.execute(
-                "SELECT fingerprint, status, response_json FROM idempotency_operations "
-                "WHERE key = ?",
-                (key,),
-            ).fetchone()
-            if row is not None:
-                if row[0] != encoded:
-                    return Reservation("conflict")
-                if row[1] == "completed":
-                    return Reservation("completed", json.loads(row[2]))
-                return Reservation("pending")
-            count = connection.execute(
-                "SELECT COUNT(*) FROM idempotency_operations"
-            ).fetchone()[0]
-            if count >= self.max_entries:
-                completed = connection.execute(
-                    "SELECT key FROM idempotency_operations "
-                    "WHERE status = 'completed' ORDER BY completed_at, rowid LIMIT 1"
-                ).fetchone()
-                if completed is None:
-                    return Reservation("busy")
+        try:
+            with self._connect() as connection:
+                connection.execute("BEGIN IMMEDIATE")
                 connection.execute(
-                    "DELETE FROM idempotency_operations WHERE key = ?",
-                    (completed[0],),
+                    "DELETE FROM idempotency_operations "
+                    "WHERE status = 'completed' AND completed_at <= ?",
+                    (now - self.ttl_seconds,),
                 )
-            connection.execute(
-                "INSERT INTO idempotency_operations "
-                "(key, fingerprint, status, created_at) VALUES (?, ?, 'pending', ?)",
-                (key, encoded, now),
-            )
+                row = connection.execute(
+                    "SELECT fingerprint, status, response_json FROM idempotency_operations "
+                    "WHERE key = ?",
+                    (key,),
+                ).fetchone()
+                if row is not None:
+                    if row[0] != encoded:
+                        return Reservation("conflict")
+                    if row[1] == "completed":
+                        return Reservation("completed", json.loads(row[2]))
+                    return Reservation("pending")
+                count = connection.execute(
+                    "SELECT COUNT(*) FROM idempotency_operations"
+                ).fetchone()[0]
+                if count >= self.max_entries:
+                    completed = connection.execute(
+                        "SELECT key FROM idempotency_operations "
+                        "WHERE status = 'completed' ORDER BY completed_at, rowid LIMIT 1"
+                    ).fetchone()
+                    if completed is None:
+                        return Reservation("busy")
+                    connection.execute(
+                        "DELETE FROM idempotency_operations WHERE key = ?",
+                        (completed[0],),
+                    )
+                connection.execute(
+                    "INSERT INTO idempotency_operations "
+                    "(key, fingerprint, status, created_at) VALUES (?, ?, 'pending', ?)",
+                    (key, encoded, now),
+                )
+        except sqlite3.OperationalError as exc:
+            if "locked" in str(exc).lower() or "busy" in str(exc).lower():
+                return Reservation("busy")
+            raise
         return Reservation("reserved")
 
     def clear(self) -> None:
@@ -106,9 +119,14 @@ class IdempotencyLedger:
 
     def complete(self, key: str, response: Any) -> None:
         encoded = json.dumps(response, separators=(",", ":"), sort_keys=True)
-        with self._connect() as connection:
-            connection.execute(
-                "UPDATE idempotency_operations SET status = 'completed', "
-                "response_json = ?, completed_at = ? WHERE key = ?",
-                (encoded, self.clock(), key),
-            )
+        try:
+            with self._connect() as connection:
+                connection.execute(
+                    "UPDATE idempotency_operations SET status = 'completed', "
+                    "response_json = ?, completed_at = ? WHERE key = ?",
+                    (encoded, self.clock(), key),
+                )
+        except sqlite3.OperationalError as exc:
+            if "locked" in str(exc).lower() or "busy" in str(exc).lower():
+                raise IdempotencyLedgerBusy from exc
+            raise

src/main.py

@@ -26,7 +26,7 @@ from src.gitea_proxy import (
     pull_review_detail,
     repos,
 )
-from src.idempotency import IdempotencyLedger
+from src.idempotency import IdempotencyLedger, IdempotencyLedgerBusy
 from src.models import Issue, Milestone, PullRequest, Repo, User
 from src.suggestion_engine import compute
 from src.views import router as frontend_router
@@ -66,6 +66,9 @@ ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS = 600.0
 ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES = 256
 AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS = 600.0
 AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES = 256
+AUTHORED_ACTION_LEDGER_LOCK_TIMEOUT_SECONDS = float(
+    os.getenv("STACKCHAIN_IDEMPOTENCY_LOCK_TIMEOUT_SECONDS", "0.1")
+)
 NOTIFICATION_MUTATION_TIMEOUT_SECONDS = 5.0
 NOTIFICATION_PAGE_TIMEOUT_SECONDS = 5.0
 WORK_PAGE_TIMEOUT_SECONDS = 5.0
@@ -105,6 +108,7 @@ _idempotency_ledger = IdempotencyLedger(
         AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES
         + ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES
     ),
+    lock_timeout_seconds=AUTHORED_ACTION_LEDGER_LOCK_TIMEOUT_SECONDS,
 )
 _available_issue_snapshot_task: asyncio.Task | None = None
 _available_issue_snapshot_value: list[dict] | None = None
@@ -264,7 +268,9 @@ async def _run_idempotent_authored_action(
     if not idempotency_key:
         return await asyncio.wait_for(operation, timeout=timeout)
 
-    reservation = _idempotency_ledger.reserve(idempotency_key, fingerprint)
+    reservation = await asyncio.to_thread(
+        _idempotency_ledger.reserve, idempotency_key, fingerprint
+    )
     if reservation.state == "conflict":
         operation.close()
         raise HTTPException(status_code=409, detail="Idempotency key already used")
@@ -294,7 +300,7 @@ async def _run_idempotent_authored_action(
     else:
         async def persist_result():
             result = await operation
-            _idempotency_ledger.complete(idempotency_key, result)
+            await asyncio.to_thread(_idempotency_ledger.complete, idempotency_key, result)
             return result
 
         task = asyncio.create_task(persist_result())
@@ -310,7 +316,14 @@ async def _run_idempotent_authored_action(
                 _authored_action_operations.pop(idempotency_key, None)
 
         task.add_done_callback(discard_completed)
-    return await asyncio.wait_for(asyncio.shield(task), timeout=timeout)
+    try:
+        return await asyncio.wait_for(asyncio.shield(task), timeout=timeout)
+    except IdempotencyLedgerBusy:
+        raise HTTPException(
+            status_code=503,
+            detail="This action may have completed; verify its result before retrying",
+            headers={"Retry-After": "5"},
+        )
 
 
 def _context_payload(user_data, repo_data, issues_data, prs_data) -> dict: