From a5f63a6fa65bce4d7eb6eb5c3b6bfaa6bb6fdd5f Mon Sep 17 00:00:00 2001 From: timmy Date: Fri, 7 Aug 2026 14:37:10 +0000 Subject: [PATCH] feat: persist authored action idempotency (#201) --- .gitignore | 1 + README.md | 7 +- src/idempotency.py | 114 ++++++++++++++++++++++++ src/main.py | 136 ++++++++++++++--------------- tests/test_authored_idempotency.py | 88 +++++++++++++++++++ tests/test_idempotency_ledger.py | 73 ++++++++++++++++ tests/test_issue_api.py | 78 +++++++++++++++-- 7 files changed, 417 insertions(+), 80 deletions(-) create mode 100644 src/idempotency.py create mode 100644 tests/test_idempotency_ledger.py diff --git a/.gitignore b/.gitignore index 5194c30..e878ebb 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ __pycache__/ *.py[cod] .pytest_cache/ .release-engine/ +.stackchain-state/ diff --git a/README.md b/README.md index e922b6a..ccfc3f7 100644 --- a/README.md +++ b/README.md @@ -23,8 +23,11 @@ Pull-request replies and mobile My Work issue and PR comments use Gitea's issue-comment API; mobile issue capture requires issue creation and assignment permission. Issue capture and authored mobile actions (issue comments, pull-request comments, notification replies, and reviews) persist per-draft -idempotency keys, so retrying after a timeout or reload replays a confirmed result instead -of posting duplicate content. Direct API callers should preserve the `Idempotency-Key` +idempotency keys, so retrying after a timeout, reload, process restart, or handoff to +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, Approve, and Request changes reviews, and assigned-PR merge require repository write permission. Native Comment, Approve, and Request changes reviews support diff --git a/src/idempotency.py b/src/idempotency.py new file mode 100644 index 0000000..97e10a0 --- /dev/null +++ b/src/idempotency.py @@ -0,0 +1,114 @@ +import json +import sqlite3 +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable + + +@dataclass(frozen=True) +class Reservation: + state: str + response: Any = None + + +class IdempotencyLedger: + """Small durable ledger for replaying successful Gitea mutations.""" + + def __init__( + self, + path: str | Path, + *, + ttl_seconds: float, + max_entries: int, + clock: Callable[[], float] = time.time, + ) -> None: + self.path = Path(path) + self.ttl_seconds = ttl_seconds + self.max_entries = max_entries + 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") + return connection + + def _initialize(self) -> None: + with self._connect() as connection: + connection.execute( + """ + CREATE TABLE IF NOT EXISTS idempotency_operations ( + key TEXT PRIMARY KEY, + fingerprint TEXT NOT NULL, + status TEXT NOT NULL CHECK(status IN ('pending', 'completed')), + response_json TEXT, + created_at REAL NOT NULL, + completed_at REAL + ) + """ + ) + connection.execute( + "CREATE INDEX IF NOT EXISTS idempotency_completed_at_idx " + "ON idempotency_operations(completed_at)" + ) + + @staticmethod + def _fingerprint(value: tuple[Any, ...]) -> str: + return json.dumps(value, separators=(",", ":"), sort_keys=True) + + 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") + 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), + ) + return Reservation("reserved") + + def clear(self) -> None: + with self._connect() as connection: + connection.execute("DELETE FROM idempotency_operations") + + 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), + ) diff --git a/src/main.py b/src/main.py index 2eaf9d9..1ade543 100644 --- a/src/main.py +++ b/src/main.py @@ -1,5 +1,6 @@ import asyncio import math +import os import time from collections.abc import Awaitable, Coroutine from contextlib import asynccontextmanager @@ -24,6 +25,7 @@ from src.gitea_proxy import ( pull_review_detail, repos, ) +from src.idempotency import IdempotencyLedger from src.models import Issue, PullRequest, Repo, User from src.suggestion_engine import compute from src.views import router as frontend_router @@ -90,12 +92,18 @@ _live_section_retry_at: dict[str, float | None] = { } _live_snapshot_refreshing_sections: set[str] = set() _read_notification_ids: set[int] = set() -_issue_creation_operations: dict[ - str, tuple[tuple[Any, ...], asyncio.Task, float] -] = {} _authored_action_operations: dict[ str, tuple[tuple[Any, ...], asyncio.Task, float] ] = {} +_state_dir = Path(os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state")) +_idempotency_ledger = IdempotencyLedger( + os.getenv("STACKCHAIN_IDEMPOTENCY_DB", str(_state_dir / "idempotency.sqlite3")), + ttl_seconds=AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS, + max_entries=( + AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES + + ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES + ), +) _available_issue_snapshot_task: asyncio.Task | None = None _available_issue_snapshot_value: list[dict] | None = None _available_issue_snapshot_created_at: float | None = None @@ -236,39 +244,52 @@ async def _run_idempotent_authored_action( if not idempotency_key: return await asyncio.wait_for(operation, timeout=timeout) - now = time.monotonic() - expired = [ - key for key, (_, task, created_at) in _authored_action_operations.items() - if task.done() and now - created_at >= AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS - ] - for key in expired: - _authored_action_operations.pop(key, None) + reservation = _idempotency_ledger.reserve(idempotency_key, fingerprint) + if reservation.state == "conflict": + operation.close() + raise HTTPException(status_code=409, detail="Idempotency key already used") + if reservation.state == "completed": + operation.close() + return reservation.response + if reservation.state == "busy": + operation.close() + raise HTTPException( + status_code=503, + detail="Authored action queue is busy; please retry", + headers={"Retry-After": "1"}, + ) existing = _authored_action_operations.get(idempotency_key) - if existing is not None: + if reservation.state == "pending": operation.close() - if existing[0] != fingerprint: - raise HTTPException(status_code=409, detail="Idempotency key already used") + if existing is None or existing[0] != fingerprint: + raise HTTPException( + status_code=503, + detail=( + "This action may still be completing; verify its result before retrying" + ), + headers={"Retry-After": "5"}, + ) task = existing[1] else: - while len(_authored_action_operations) >= AUTHORED_ACTION_IDEMPOTENCY_MAX_ENTRIES: - completed = [ - key for key, (_, task, _) in _authored_action_operations.items() - if task.done() - ] - if not completed: - operation.close() - raise HTTPException( - status_code=503, - detail="Authored action queue is busy; please retry", - headers={"Retry-After": "1"}, - ) - oldest = min( - completed, key=lambda key: _authored_action_operations[key][2] - ) - _authored_action_operations.pop(oldest) - task = asyncio.create_task(operation) - _authored_action_operations[idempotency_key] = (fingerprint, task, now) + async def persist_result(): + result = await operation + _idempotency_ledger.complete(idempotency_key, result) + return result + + task = asyncio.create_task(persist_result()) + _authored_action_operations[idempotency_key] = ( + fingerprint, + task, + time.monotonic(), + ) + + def discard_completed(completed: asyncio.Task) -> None: + existing_operation = _authored_action_operations.get(idempotency_key) + if existing_operation is not None and existing_operation[1] is completed: + _authored_action_operations.pop(idempotency_key, None) + + task.add_done_callback(discard_completed) return await asyncio.wait_for(asyncio.shield(task), timeout=timeout) @@ -1269,48 +1290,19 @@ async def create_assigned_issue( repository, creation.title, creation.body, login, creation.label_ids ) - operation = create_issue() - if idempotency_key: - now = time.monotonic() - expired = [ - key for key, (_, task, created_at) in _issue_creation_operations.items() - if task.done() - and now - created_at >= ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS - ] - for key in expired: - _issue_creation_operations.pop(key, None) - fingerprint = ( - repository, creation.title, creation.body, tuple(creation.label_ids) - ) - existing = _issue_creation_operations.get(idempotency_key) - if existing is not None: - operation.close() - if existing[0] != fingerprint: - raise HTTPException(status_code=409, detail="Idempotency key already used") - task = existing[1] - else: - while len(_issue_creation_operations) >= ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES: - completed = [ - key for key, (_, task, _) in _issue_creation_operations.items() - if task.done() - ] - if not completed: - operation.close() - raise HTTPException( - status_code=503, - detail="Issue creation is busy; please retry", - headers={"Retry-After": "1"}, - ) - oldest = min( - completed, key=lambda key: _issue_creation_operations[key][2] - ) - _issue_creation_operations.pop(oldest) - task = asyncio.create_task(operation) - _issue_creation_operations[idempotency_key] = (fingerprint, task, now) - operation = asyncio.shield(task) - try: - result = await asyncio.wait_for(operation, timeout=ISSUE_ACTION_TIMEOUT_SECONDS) + result = await _run_idempotent_authored_action( + create_issue(), + idempotency_key=idempotency_key, + fingerprint=( + "issue-create", + repository, + creation.title, + creation.body, + tuple(creation.label_ids), + ), + timeout=ISSUE_ACTION_TIMEOUT_SECONDS, + ) except HTTPException: raise except Exception: diff --git a/tests/test_authored_idempotency.py b/tests/test_authored_idempotency.py index 433a37f..e84e225 100644 --- a/tests/test_authored_idempotency.py +++ b/tests/test_authored_idempotency.py @@ -4,13 +4,16 @@ import httpx import pytest from src import main +from src.idempotency import IdempotencyLedger @pytest.fixture(autouse=True) def clear_authored_operations(): main._authored_action_operations.clear() + main._idempotency_ledger.clear() yield main._authored_action_operations.clear() + main._idempotency_ledger.clear() @pytest.mark.anyio @@ -140,3 +143,88 @@ async def test_review_replays_same_key_and_rejects_changed_head(monkeypatch): assert [created.status_code, replayed.status_code, conflict.status_code] == [201, 201, 409] assert calls == [("abc", "approve", "Good")] + + +@pytest.mark.anyio +async def test_completed_comment_replays_after_ledger_reconstruction(monkeypatch, tmp_path): + calls = 0 + + async def assigned(_repository, _number): + return True + + async def comment(_repository, _number, body): + nonlocal calls + calls += 1 + return {"id": 201, "body": body} + + database = tmp_path / "actions.sqlite3" + monkeypatch.setattr( + main, + "_idempotency_ledger", + IdempotencyLedger(database, ttl_seconds=600, max_entries=256), + raising=False, + ) + monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned) + monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment) + transport = httpx.ASGITransport(app=main.app) + headers = {"Idempotency-Key": "restart-comment-201"} + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + created = await client.post( + "/api/v1/repos/stackchain/api/issues/7/comments", + json={"body": "Persist me"}, + headers=headers, + ) + main._authored_action_operations.clear() + monkeypatch.setattr( + main, + "_idempotency_ledger", + IdempotencyLedger(database, ttl_seconds=600, max_entries=256), + ) + replayed = await client.post( + "/api/v1/repos/stackchain/api/issues/7/comments", + json={"body": "Persist me"}, + headers=headers, + ) + + assert [created.status_code, replayed.status_code] == [201, 201] + assert replayed.json() == {"id": 201, "body": "Persist me"} + assert calls == 1 + + +@pytest.mark.anyio +async def test_orphaned_pending_comment_fails_closed_without_upstream_retry( + monkeypatch, tmp_path +): + calls = 0 + + async def assigned(_repository, _number): + return True + + async def comment(_repository, _number, body): + nonlocal calls + calls += 1 + return {"id": 202, "body": body} + + ledger = IdempotencyLedger( + tmp_path / "orphan.sqlite3", ttl_seconds=600, max_entries=256 + ) + fingerprint = ("issue-comment", "stackchain/api", 7, "Verify first") + assert ledger.reserve("orphan-comment-201", fingerprint).state == "reserved" + 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) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/api/v1/repos/stackchain/api/issues/7/comments", + json={"body": "Verify first"}, + headers={"Idempotency-Key": "orphan-comment-201"}, + ) + + assert response.status_code == 503 + assert response.json()["detail"] == ( + "This action may still be completing; verify its result before retrying" + ) + assert calls == 0 diff --git a/tests/test_idempotency_ledger.py b/tests/test_idempotency_ledger.py new file mode 100644 index 0000000..0763e05 --- /dev/null +++ b/tests/test_idempotency_ledger.py @@ -0,0 +1,73 @@ +import sqlite3 +from concurrent.futures import ThreadPoolExecutor +from threading import Barrier + +from src.idempotency import IdempotencyLedger + + +def test_completed_result_replays_after_ledger_reconstruction(tmp_path): + database = tmp_path / "idempotency.sqlite3" + first = IdempotencyLedger(database, ttl_seconds=600, max_entries=256) + + reservation = first.reserve("create-201", ("issue-create", "stackchain/api", "Ship it")) + assert reservation.state == "reserved" + first.complete("create-201", {"number": 42, "title": "Ship it"}) + + reconstructed = IdempotencyLedger(database, ttl_seconds=600, max_entries=256) + replay = reconstructed.reserve( + "create-201", ("issue-create", "stackchain/api", "Ship it") + ) + + assert replay.state == "completed" + assert replay.response == {"number": 42, "title": "Ship it"} + assert sqlite3.connect(database).execute( + "SELECT status FROM idempotency_operations WHERE key = ?", ("create-201",) + ).fetchone() == ("completed",) + + +def test_reservation_expires_completed_rows_and_enforces_capacity(tmp_path): + now = [1_000.0] + ledger = IdempotencyLedger( + tmp_path / "bounded.sqlite3", + ttl_seconds=10, + max_entries=2, + clock=lambda: now[0], + ) + assert ledger.reserve("old", ("comment", "old")).state == "reserved" + ledger.complete("old", {"id": 1}) + now[0] += 11 + + assert ledger.reserve("new-1", ("comment", "new-1")).state == "reserved" + ledger.complete("new-1", {"id": 2}) + assert ledger.reserve("new-2", ("comment", "new-2")).state == "reserved" + ledger.complete("new-2", {"id": 3}) + replacement = ledger.reserve("new-3", ("comment", "new-3")) + + assert replacement.state == "reserved" + with sqlite3.connect(ledger.path) as connection: + keys = connection.execute( + "SELECT key FROM idempotency_operations ORDER BY key" + ).fetchall() + indexes = connection.execute( + "PRAGMA index_list(idempotency_operations)" + ).fetchall() + assert keys == [("new-2",), ("new-3",)] + assert any(index[1] == "idempotency_completed_at_idx" for index in indexes) + + +def test_independent_connections_atomically_reserve_one_operation(tmp_path): + database = tmp_path / "workers.sqlite3" + ledgers = [ + IdempotencyLedger(database, ttl_seconds=600, max_entries=256) + for _ in range(2) + ] + barrier = Barrier(2) + + def reserve(ledger): + barrier.wait() + return ledger.reserve("worker-key-201", ("review", "stackchain/api", 7)).state + + with ThreadPoolExecutor(max_workers=2) as executor: + states = list(executor.map(reserve, ledgers)) + + assert sorted(states) == ["pending", "reserved"] diff --git a/tests/test_issue_api.py b/tests/test_issue_api.py index c3d1f83..17a33b0 100644 --- a/tests/test_issue_api.py +++ b/tests/test_issue_api.py @@ -5,6 +5,7 @@ import httpx import pytest from src import gitea_proxy, main +from src.idempotency import IdempotencyLedger @pytest.mark.anyio @@ -226,9 +227,11 @@ async def test_edit_assigned_issue_reports_revision_conflict_without_mutation(mo @pytest.fixture(autouse=True) def clear_issue_creation_operations(): - main._issue_creation_operations.clear() + main._authored_action_operations.clear() + main._idempotency_ledger.clear() yield - main._issue_creation_operations.clear() + main._authored_action_operations.clear() + main._idempotency_ledger.clear() @pytest.mark.anyio @@ -331,6 +334,61 @@ async def test_create_issue_replays_one_upstream_result_for_concurrent_idempoten assert calls == 1 +@pytest.mark.anyio +async def test_completed_issue_creation_replays_after_ledger_reconstruction(monkeypatch, tmp_path): + calls = 0 + + async def user(): + return {"login": "timmy"} + + async def available_repos(): + return [{"full_name": "stackchain/api"}] + + async def create(repository, title, body, assignee, label_ids): + nonlocal calls + calls += 1 + return { + "id": 201, "number": 201, "title": title, "state": "open", + "repository": repository, "labels": [], "assignees": [assignee], + "updated_at": "2026-08-07T15:00:00Z", + "url": "https://forge.example/stackchain/api/issues/201", + } + + database = tmp_path / "issues.sqlite3" + monkeypatch.setattr( + main, + "_idempotency_ledger", + IdempotencyLedger(database, ttl_seconds=600, max_entries=256), + ) + monkeypatch.setattr(main.gitea_proxy, "current_user", user) + monkeypatch.setattr(main.gitea_proxy, "repos", available_repos) + monkeypatch.setattr(main.gitea_proxy, "create_issue", create) + transport = httpx.ASGITransport(app=main.app) + headers = {"Idempotency-Key": "restart-create-201"} + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + created = await client.post( + "/api/v1/repos/stackchain/api/issues", + json={"title": "Durable capture"}, + headers=headers, + ) + main._authored_action_operations.clear() + monkeypatch.setattr( + main, + "_idempotency_ledger", + IdempotencyLedger(database, ttl_seconds=600, max_entries=256), + ) + replayed = await client.post( + "/api/v1/repos/stackchain/api/issues", + json={"title": "Durable capture"}, + headers=headers, + ) + + assert [created.status_code, replayed.status_code] == [201, 201] + assert replayed.json()["number"] == 201 + assert calls == 1 + + @pytest.mark.anyio async def test_create_issue_rejects_changed_payload_for_an_existing_idempotency_key(monkeypatch): calls = [] @@ -412,7 +470,9 @@ async def test_create_issue_retry_recovers_result_after_the_first_request_times_ @pytest.mark.anyio -async def test_create_issue_idempotency_registry_evicts_oldest_entry_at_size_limit(monkeypatch): +async def test_create_issue_durable_ledger_evicts_completed_entry_at_size_limit( + monkeypatch, tmp_path +): async def user(): return {"login": "timmy"} @@ -427,7 +487,10 @@ async def test_create_issue_idempotency_registry_evicts_oldest_entry_at_size_lim "url": "https://forge.example/stackchain/api/issues/17", } - monkeypatch.setattr(main, "ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES", 2) + ledger = IdempotencyLedger( + tmp_path / "bounded-issues.sqlite3", ttl_seconds=600, max_entries=2 + ) + monkeypatch.setattr(main, "_idempotency_ledger", ledger) monkeypatch.setattr(main.gitea_proxy, "current_user", user) monkeypatch.setattr(main.gitea_proxy, "repos", available_repos) monkeypatch.setattr(main.gitea_proxy, "create_issue", create) @@ -441,8 +504,11 @@ async def test_create_issue_idempotency_registry_evicts_oldest_entry_at_size_lim ) assert response.status_code == 201 - assert len(main._issue_creation_operations) == 2 - assert "capture-177-bounded-0" not in main._issue_creation_operations + replay = ledger.reserve( + "capture-177-bounded-0", + ("issue-create", "stackchain/api", "Capture 0", "", ()), + ) + assert replay.state == "reserved" @pytest.mark.anyio -- 2.43.0