From b3c76dcd697790a61ef130308855d94d6ebb8fd4 Mon Sep 17 00:00:00 2001 From: timmy Date: Sat, 8 Aug 2026 01:01:31 +0000 Subject: [PATCH] fix: keep authored actions responsive under ledger contention (#248) --- README.md | 10 +- src/idempotency.py | 104 ++++++++++-------- src/main.py | 21 +++- tests/test_authored_idempotency.py | 170 +++++++++++++++++++++++++++++ 4 files changed, 256 insertions(+), 49 deletions(-) diff --git a/README.md b/README.md index ab6dc9c..72e97a6 100644 --- a/README.md +++ b/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` -header with the unchanged route and payload until a `201` response is confirmed. Closing an assigned issue, native Comment, +SQLite path); the local default is `.stackchain-state/idempotency.sqlite3`. Ledger reads and +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 diff --git a/src/idempotency.py b/src/idempotency.py index 97e10a0..7ee5946 100644 --- a/src/idempotency.py +++ b/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 diff --git a/src/main.py b/src/main.py index 158de3f..061e09c 100644 --- a/src/main.py +++ b/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: diff --git a/tests/test_authored_idempotency.py b/tests/test_authored_idempotency.py index e84e225..8d6ebd3 100644 --- a/tests/test_authored_idempotency.py +++ b/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" -- 2.43.0