feat: make authored actions idempotent (#185)

src/main.py

@@ -1,7 +1,7 @@
 import asyncio
 import math
 import time
-from collections.abc import Awaitable
+from collections.abc import Awaitable, Coroutine
 from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import Any, Literal
@@ -61,6 +61,8 @@ REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
 ISSUE_ACTION_TIMEOUT_SECONDS = 5.0
 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
 NOTIFICATION_MUTATION_TIMEOUT_SECONDS = 5.0
 NOTIFICATION_PAGE_TIMEOUT_SECONDS = 5.0
 WORK_PAGE_TIMEOUT_SECONDS = 5.0
@@ -90,6 +92,9 @@ _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]
+] = {}
 _available_issue_snapshot_task: asyncio.Task | None = None
 _available_issue_snapshot_value: list[dict] | None = None
 _available_issue_snapshot_created_at: float | None = None
@@ -193,6 +198,52 @@ class PullMergeSubmission(BaseModel):
     expected_head_sha: str = Field(min_length=1, max_length=128)
 
 
+async def _run_idempotent_authored_action(
+    operation: Coroutine[Any, Any, Any],
+    *,
+    idempotency_key: str | None,
+    fingerprint: tuple[Any, ...],
+    timeout: float,
+):
+    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)
+
+    existing = _authored_action_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(_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)
+    return await asyncio.wait_for(asyncio.shield(task), timeout=timeout)
+
+
 def _context_payload(user_data, repo_data, issues_data, prs_data) -> dict:
     user_model = User(
         id=user_data["id"],
@@ -908,13 +959,19 @@ async def read_notification(thread_id: int = PathParam(gt=0)) -> JSONResponse:
 
 @app.post("/api/v1/notifications/{thread_id}/reply", status_code=201)
 async def reply_to_notification(
-    reply: NotificationReply, thread_id: int = PathParam(gt=0)
+    reply: NotificationReply,
+    thread_id: int = PathParam(gt=0),
+    idempotency_key: str | None = Header(default=None, max_length=128),
 ) -> JSONResponse:
     try:
-        result = await asyncio.wait_for(
+        result = await _run_idempotent_authored_action(
             gitea_proxy.reply_to_notification(thread_id, reply.body),
+            idempotency_key=idempotency_key,
+            fingerprint=("notification-reply", thread_id, reply.body),
             timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
         )
+    except HTTPException:
+        raise
     except Exception:
         return JSONResponse(
             {
@@ -1129,7 +1186,11 @@ async def create_assigned_issue(
 
 @app.post("/api/v1/repos/{owner}/{repo}/issues/{number}/comments", status_code=201)
 async def comment_on_assigned_issue(
-    comment: IssueComment, owner: str, repo: str, number: int = PathParam(gt=0)
+    comment: IssueComment,
+    owner: str,
+    repo: str,
+    number: int = PathParam(gt=0),
+    idempotency_key: str | None = Header(default=None, max_length=128),
 ):
     repository = f"{owner}/{repo}"
 
@@ -1139,7 +1200,12 @@ async def comment_on_assigned_issue(
         return await gitea_proxy.comment_on_issue(repository, number, comment.body)
 
     try:
-        result = await asyncio.wait_for(post_comment(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS)
+        result = await _run_idempotent_authored_action(
+            post_comment(),
+            idempotency_key=idempotency_key,
+            fingerprint=("issue-comment", repository, number, comment.body),
+            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
+        )
     except HTTPException:
         raise
     except Exception:
@@ -1292,7 +1358,11 @@ async def assigned_pull_detail(owner: str, repo: str, number: int = PathParam(gt
 
 @app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/comments", status_code=201)
 async def comment_on_assigned_pull(
-    comment: IssueComment, owner: str, repo: str, number: int = PathParam(gt=0)
+    comment: IssueComment,
+    owner: str,
+    repo: str,
+    number: int = PathParam(gt=0),
+    idempotency_key: str | None = Header(default=None, max_length=128),
 ):
     repository = f"{owner}/{repo}"
 
@@ -1302,8 +1372,11 @@ async def comment_on_assigned_pull(
         return await gitea_proxy.comment_on_issue(repository, number, comment.body)
 
     try:
-        result = await asyncio.wait_for(
-            post_comment(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
+        result = await _run_idempotent_authored_action(
+            post_comment(),
+            idempotency_key=idempotency_key,
+            fingerprint=("pull-comment", repository, number, comment.body),
+            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
         )
     except HTTPException:
         raise
@@ -1358,7 +1431,11 @@ async def merge_assigned_pull(
 
 @app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/review", status_code=201)
 async def submit_review(
-    submission: PullReviewSubmission, owner: str, repo: str, number: int
+    submission: PullReviewSubmission,
+    owner: str,
+    repo: str,
+    number: int,
+    idempotency_key: str | None = Header(default=None, max_length=128),
 ):
     repository = f"{owner}/{repo}"
 
@@ -1380,8 +1457,18 @@ async def submit_review(
         return await gitea_proxy.submit_pull_review(*args)
 
     try:
-        result = await asyncio.wait_for(
-            submit_requested_review(), timeout=REVIEW_DETAIL_TIMEOUT_SECONDS
+        comment_fingerprint = tuple(
+            (comment.path, comment.body, comment.new_position, comment.old_position)
+            for comment in submission.comments
+        )
+        result = await _run_idempotent_authored_action(
+            submit_requested_review(),
+            idempotency_key=idempotency_key,
+            fingerprint=(
+                "pull-review", repository, number, submission.expected_head_sha,
+                submission.decision, submission.body, comment_fingerprint,
+            ),
+            timeout=REVIEW_DETAIL_TIMEOUT_SECONDS,
         )
     except HTTPException:
         raise