Keep authored actions responsive during ledger contention (#249)

Closes #248

README.md

@@ -30,8 +30,14 @@ idempotency keys, so retrying after a timeout, reload, process restart, or hando
 another worker replays a confirmed result instead of posting duplicate content. Results
 are coordinated through a bounded SQLite ledger. Set `STACKCHAIN_STATE_DIR` to a
 persistent, writable service directory (or set `STACKCHAIN_IDEMPOTENCY_DB` to an explicit
-SQLite path); the local default is `.stackchain-state/idempotency.sqlite3`. Direct API callers should preserve the `Idempotency-Key`
+SQLite path); the local default is `.stackchain-state/idempotency.sqlite3`. Ledger reads and
-header with the unchanged route and payload until a `201` response is confirmed. Closing an assigned issue, native Comment,
+writes run outside the request event loop, and lock admission is bounded to 100 ms by
+default. Tune it with `STACKCHAIN_IDEMPOTENCY_LOCK_TIMEOUT_SECONDS`; keep the value below
+route deadlines. Reservation contention returns retryable HTTP 503 with `Retry-After: 1`.
+If contention occurs after the upstream mutation, the dashboard fails closed with
+`Retry-After: 5` and asks the caller to verify the result before retrying. Direct API callers
+should preserve the `Idempotency-Key` header with the unchanged route and payload until a
+`201` response is confirmed. Closing an assigned issue, native Comment,
 Approve, and Request changes reviews, and assigned-PR merge require repository
 write permission. Native Comment, Approve, and Request changes reviews support
 head-scoped draft comments anchored to changed lines; the dashboard validates each

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 = sqlite3.connect(self.path, timeout=self.lock_timeout_seconds)
-        connection.execute("PRAGMA busy_timeout = 5000")
+        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:
+        try:
-            connection.execute("BEGIN IMMEDIATE")
+            with self._connect() as connection:
-            connection.execute(
+                connection.execute("BEGIN IMMEDIATE")
-                "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")
                 connection.execute(
-                    "DELETE FROM idempotency_operations WHERE key = ?",
+                    "DELETE FROM idempotency_operations "
-                    (completed[0],),
+                    "WHERE status = 'completed' AND completed_at <= ?",
+                    (now - self.ttl_seconds,),
                 )
-            connection.execute(
+                row = connection.execute(
-                "INSERT INTO idempotency_operations "
+                    "SELECT fingerprint, status, response_json FROM idempotency_operations "
-                "(key, fingerprint, status, created_at) VALUES (?, ?, 'pending', ?)",
+                    "WHERE key = ?",
-                (key, encoded, now),
+                    (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:
+        try:
-            connection.execute(
+            with self._connect() as connection:
-                "UPDATE idempotency_operations SET status = 'completed', "
+                connection.execute(
-                "response_json = ?, completed_at = ? WHERE key = ?",
+                    "UPDATE idempotency_operations SET status = 'completed', "
-                (encoded, self.clock(), key),
+                    "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:

tests/test_authored_idempotency.py

@@ -1,4 +1,7 @@
 import asyncio
+import sqlite3
+import threading
+import time
 
 import httpx
 import pytest
@@ -228,3 +231,170 @@ async def test_orphaned_pending_comment_fails_closed_without_upstream_retry(
         "This action may still be completing; verify its result before retrying"
     )
     assert calls == 0
+
+
+@pytest.mark.anyio
+async def test_slow_ledger_reservation_does_not_block_health_requests(monkeypatch):
+    async def assigned(_repository, _number):
+        return True
+
+    async def comment(_repository, _number, body):
+        return {"id": 248, "body": body}
+
+    started = threading.Event()
+    original_reserve = main._idempotency_ledger.reserve
+
+    def slow_reserve(key, fingerprint):
+        started.set()
+        time.sleep(0.2)
+        return original_reserve(key, fingerprint)
+
+    monkeypatch.setattr(main._idempotency_ledger, "reserve", slow_reserve)
+    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
+    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        started_at = asyncio.get_running_loop().time()
+        authored = asyncio.create_task(client.post(
+            "/api/v1/repos/stackchain/api/issues/248/comments",
+            json={"body": "Keep the dashboard responsive"},
+            headers={"Idempotency-Key": "contention-248"},
+        ))
+        assert await asyncio.to_thread(started.wait, 1)
+        health = await client.get("/healthz")
+        health_elapsed = asyncio.get_running_loop().time() - started_at
+        authored_response = await authored
+
+    assert health.status_code == 200
+    assert health_elapsed < 0.15
+    assert authored_response.status_code == 201
+
+
+@pytest.mark.anyio
+async def test_locked_ledger_returns_bounded_retryable_response(monkeypatch, tmp_path):
+    upstream_calls = 0
+
+    async def assigned(_repository, _number):
+        return True
+
+    async def comment(_repository, _number, body):
+        nonlocal upstream_calls
+        upstream_calls += 1
+        return {"id": 248, "body": body}
+
+    database = tmp_path / "contended.sqlite3"
+    ledger = IdempotencyLedger(
+        database,
+        ttl_seconds=600,
+        max_entries=256,
+        lock_timeout_seconds=0.05,
+    )
+    monkeypatch.setattr(main, "_idempotency_ledger", ledger)
+    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
+    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
+    blocker = sqlite3.connect(database)
+    blocker.execute("BEGIN IMMEDIATE")
+    transport = httpx.ASGITransport(app=main.app)
+
+    try:
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+            started_at = asyncio.get_running_loop().time()
+            response = await client.post(
+                "/api/v1/repos/stackchain/api/issues/248/comments",
+                json={"body": "Retry without freezing"},
+                headers={"Idempotency-Key": "locked-ledger-248"},
+            )
+            elapsed = asyncio.get_running_loop().time() - started_at
+    finally:
+        blocker.rollback()
+        blocker.close()
+
+    assert response.status_code == 503
+    assert response.headers["Retry-After"] == "1"
+    assert response.json()["detail"] == "Authored action queue is busy; please retry"
+    assert elapsed < 0.5
+    assert upstream_calls == 0
+
+
+@pytest.mark.anyio
+async def test_slow_ledger_completion_does_not_block_health_requests(monkeypatch):
+    async def assigned(_repository, _number):
+        return True
+
+    async def comment(_repository, _number, body):
+        return {"id": 249, "body": body}
+
+    started = threading.Event()
+    original_complete = main._idempotency_ledger.complete
+
+    def slow_complete(key, response):
+        started.set()
+        time.sleep(0.2)
+        return original_complete(key, response)
+
+    monkeypatch.setattr(main._idempotency_ledger, "complete", slow_complete)
+    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
+    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        started_at = asyncio.get_running_loop().time()
+        authored = asyncio.create_task(client.post(
+            "/api/v1/repos/stackchain/api/issues/248/comments",
+            json={"body": "Persist without freezing"},
+            headers={"Idempotency-Key": "slow-completion-248"},
+        ))
+        assert await asyncio.to_thread(started.wait, 1)
+        health = await client.get("/healthz")
+        health_elapsed = asyncio.get_running_loop().time() - started_at
+        authored_response = await authored
+
+    assert health.status_code == 200
+    assert health_elapsed < 0.15
+    assert authored_response.status_code == 201
+
+
+@pytest.mark.anyio
+async def test_locked_completion_fails_closed_with_retry_guidance(monkeypatch, tmp_path):
+    database = tmp_path / "completion-contention.sqlite3"
+    ledger = IdempotencyLedger(
+        database, ttl_seconds=600, max_entries=256, lock_timeout_seconds=0.05
+    )
+    blocker = None
+
+    async def assigned(_repository, _number):
+        return True
+
+    async def comment(_repository, _number, body):
+        nonlocal blocker
+        blocker = sqlite3.connect(database)
+        blocker.execute("BEGIN IMMEDIATE")
+        return {"id": 250, "body": body}
+
+    monkeypatch.setattr(main, "_idempotency_ledger", ledger)
+    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
+    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
+    transport = httpx.ASGITransport(app=main.app)
+
+    try:
+        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+            response = await client.post(
+                "/api/v1/repos/stackchain/api/issues/248/comments",
+                json={"body": "The upstream mutation may already exist"},
+                headers={"Idempotency-Key": "locked-completion-248"},
+            )
+    finally:
+        if blocker is not None:
+            blocker.rollback()
+            blocker.close()
+
+    assert response.status_code == 503
+    assert response.headers["Retry-After"] == "5"
+    assert response.json()["detail"] == (
+        "This action may have completed; verify its result before retrying"
+    )
+    assert ledger.reserve(
+        "locked-completion-248",
+        ("issue-comment", "stackchain/api", 248, "The upstream mutation may already exist"),
+    ).state == "pending"