Merge pull request 'Persist authored action idempotency across restarts and workers' (#202) from timmy/201-durable-idempotency into main

.gitignore

@@ -2,3 +2,4 @@ __pycache__/
 *.py[cod]
 .pytest_cache/
 .release-engine/
+.stackchain-state/

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

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),
+            )

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:

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

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"]

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