Make mobile authored actions idempotent across retries #186

Merged
rockachopa merged 1 commits from timmy/185-idempotent-authored-actions into main 2026-08-07 09:56:15 +00:00
9 changed files with 389 additions and 30 deletions

View File

@ -20,10 +20,11 @@ create issue comments, close assigned issues, inspect/comment on assigned pull
requests, merge assigned pull requests, and submit pull-request reviews. requests, merge assigned pull requests, and submit pull-request reviews.
Pull-request replies and mobile My Work issue and PR comments use Gitea's Pull-request replies and mobile My Work issue and PR comments use Gitea's
issue-comment API; mobile issue capture requires issue issue-comment API; mobile issue capture requires issue
creation and assignment permission. Issue capture persists a per-draft idempotency key, creation and assignment permission. Issue capture and authored mobile actions (issue
so retrying after a timeout or reload replays a confirmed creation instead of posting a comments, pull-request comments, notification replies, and reviews) persist per-draft
duplicate; callers integrating directly should preserve the `Idempotency-Key` header idempotency keys, so retrying after a timeout or reload replays a confirmed result instead
with the unchanged payload until a `201` response is confirmed. Closing an assigned issue, native Comment, of posting duplicate content. 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 Approve, and Request changes reviews, and assigned-PR merge require repository
write permission. Native Comment, Approve, and Request changes reviews support write permission. Native Comment, Approve, and Request changes reviews support
head-scoped draft comments anchored to changed lines; the dashboard validates each head-scoped draft comments anchored to changed lines; the dashboard validates each

View File

@ -598,7 +598,7 @@ textarea { resize: vertical; min-height: 120px; }
if (!response.ok) throw new Error(payload.error || 'Review request failed.'); if (!response.ok) throw new Error(payload.error || 'Review request failed.');
return payload; return payload;
} }
const reviewController = createReviewController({ fetchJson: fetchReviewJson }); const reviewController = createReviewController({ fetchJson: fetchReviewJson, storage: localStorage });
const issueController = createIssueSheet({ fetchJson: fetchReviewJson, storage: localStorage }); const issueController = createIssueSheet({ fetchJson: fetchReviewJson, storage: localStorage });
const issueCapture = createIssueCapture({ fetchJson: fetchReviewJson, storage: localStorage }); const issueCapture = createIssueCapture({ fetchJson: fetchReviewJson, storage: localStorage });
const pullController = createPullSheet({ fetchJson: fetchReviewJson, storage: localStorage }); const pullController = createPullSheet({ fetchJson: fetchReviewJson, storage: localStorage });
@ -670,10 +670,13 @@ textarea { resize: vertical; min-height: 120px; }
return payload; return payload;
} }
async function postNotificationReply(notificationId, body) { async function postNotificationReply(notificationId, body, operationId) {
const response = await fetch('api/v1/notifications/' + encodeURIComponent(notificationId) + '/reply', { const response = await fetch('api/v1/notifications/' + encodeURIComponent(notificationId) + '/reply', {
method: 'POST', method: 'POST',
headers: { Accept: 'application/json', 'Content-Type': 'application/json' }, headers: {
Accept: 'application/json', 'Content-Type': 'application/json',
'Idempotency-Key': operationId,
},
body: JSON.stringify({ body }), body: JSON.stringify({ body }),
}); });
const payload = await response.json().catch(() => ({})); const payload = await response.json().catch(() => ({}));

View File

@ -1,4 +1,4 @@
function createIssueSheet({ fetchJson, storage }) { function createIssueSheet({ fetchJson, storage, createOperationId = () => globalThis.crypto?.randomUUID?.() || String(Date.now()) + '-' + Math.random() }) {
let commentRequest = null; let commentRequest = null;
let closeRequest = null; let closeRequest = null;
let releaseRequest = null; let releaseRequest = null;
@ -6,6 +6,7 @@ function createIssueSheet({ fetchJson, storage }) {
const issuePath = item => 'api/v1/repos/' + item.repository.split('/').map(encodeURIComponent).join('/') + const issuePath = item => 'api/v1/repos/' + item.repository.split('/').map(encodeURIComponent).join('/') +
'/issues/' + encodeURIComponent(item.number); '/issues/' + encodeURIComponent(item.number);
const draftKey = item => 'stackchain.issue-comment.v1:' + item.repository + '#' + item.number; const draftKey = item => 'stackchain.issue-comment.v1:' + item.repository + '#' + item.number;
const operationKey = item => draftKey(item) + ':operation';
const labelDraftKey = item => 'stackchain.issue-labels.v1:' + item.repository + '#' + item.number; const labelDraftKey = item => 'stackchain.issue-labels.v1:' + item.repository + '#' + item.number;
return { return {
@ -24,7 +25,10 @@ function createIssueSheet({ fetchJson, storage }) {
catch (_error) { return ''; } catch (_error) { return ''; }
}, },
saveDraft(item, body) { saveDraft(item, body) {
try { storage?.setItem(draftKey(item), body); } try {
if ((storage?.getItem(draftKey(item)) || '') !== body) storage?.removeItem(operationKey(item));
storage?.setItem(draftKey(item), body);
}
catch (_error) { /* The textarea remains the fallback. */ } catch (_error) { /* The textarea remains the fallback. */ }
}, },
loadLabelDraft(item) { loadLabelDraft(item) {
@ -82,13 +86,20 @@ function createIssueSheet({ fetchJson, storage }) {
comment(item, body) { comment(item, body) {
if (commentRequest) return commentRequest; if (commentRequest) return commentRequest;
this.saveDraft(item, body); this.saveDraft(item, body);
let operationId;
try {
operationId = storage?.getItem(operationKey(item)) || String(createOperationId()).slice(0, 128);
storage?.setItem(operationKey(item), operationId);
} catch (_error) { operationId = String(createOperationId()).slice(0, 128); }
commentRequest = fetchJson(issuePath(item) + '/comments', { commentRequest = fetchJson(issuePath(item) + '/comments', {
method: 'POST', method: 'POST',
headers: { Accept: 'application/json', 'Content-Type': 'application/json' }, headers: { Accept: 'application/json', 'Content-Type': 'application/json', 'Idempotency-Key': operationId },
body: JSON.stringify({ body }), body: JSON.stringify({ body }),
}).then(result => { }).then(result => {
try { storage?.removeItem(draftKey(item)); } try { storage?.removeItem(draftKey(item)); }
catch (_error) { /* The upstream comment is authoritative. */ } catch (_error) { /* The upstream comment is authoritative. */ }
try { storage?.removeItem(operationKey(item)); }
catch (_error) { /* A confirmed result no longer needs replay identity. */ }
return result; return result;
}).finally(() => { commentRequest = null; }); }).finally(() => { commentRequest = null; });
return commentRequest; return commentRequest;

View File

@ -286,27 +286,41 @@ function createNotificationReader({ load, markRead, onOpen, onDetail, onItems, o
}; };
} }
function createNotificationReplier({ post, storage, onStatus }) { function createNotificationReplier({
post, storage, onStatus,
createOperationId = () => globalThis.crypto?.randomUUID?.() || String(Date.now()) + '-' + Math.random(),
}) {
let pending = false; let pending = false;
const keyFor = item => 'stackchain.update-reply.v1.' + item.notification_id; const keyFor = item => 'stackchain.update-reply.v1.' + item.notification_id;
const operationKeyFor = item => keyFor(item) + ':operation';
return { return {
loadDraft(item) { loadDraft(item) {
try { return storage.getItem(keyFor(item)) || ''; } try { return storage.getItem(keyFor(item)) || ''; }
catch (_error) { return ''; } catch (_error) { return ''; }
}, },
saveDraft(item, body) { saveDraft(item, body) {
try { storage.setItem(keyFor(item), body); } try {
if ((storage.getItem(keyFor(item)) || '') !== body) storage.removeItem(operationKeyFor(item));
storage.setItem(keyFor(item), body);
}
catch (_error) { /* Keep the editable textarea as the fallback. */ } catch (_error) { /* Keep the editable textarea as the fallback. */ }
}, },
async submit(item, body) { async submit(item, body) {
if (pending) return false; if (pending) return false;
pending = true; pending = true;
this.saveDraft(item, body); this.saveDraft(item, body);
let operationId;
try {
operationId = storage.getItem(operationKeyFor(item)) || String(createOperationId()).slice(0, 128);
storage.setItem(operationKeyFor(item), operationId);
} catch (_error) { operationId = String(createOperationId()).slice(0, 128); }
onStatus('Sending reply…'); onStatus('Sending reply…');
try { try {
const result = await post(item.notification_id, body); const result = await post(item.notification_id, body, operationId);
try { storage.removeItem(keyFor(item)); } try { storage.removeItem(keyFor(item)); }
catch (_error) { /* The posted reply is still authoritative. */ } catch (_error) { /* The posted reply is still authoritative. */ }
try { storage.removeItem(operationKeyFor(item)); }
catch (_error) { /* A confirmed result no longer needs replay identity. */ }
onStatus('Reply posted. You can mark this update read when ready.'); onStatus('Reply posted. You can mark this update read when ready.');
return result; return result;
} catch (_error) { } catch (_error) {

View File

@ -11,12 +11,13 @@ function mergeEligibility(detail) {
return { allowed: true, reason: 'Ready to merge' }; return { allowed: true, reason: 'Ready to merge' };
} }
function createPullSheet({ fetchJson, storage }) { function createPullSheet({ fetchJson, storage, createOperationId = () => globalThis.crypto?.randomUUID?.() || String(Date.now()) + '-' + Math.random() }) {
let commentRequest = null; let commentRequest = null;
let mergeRequest = null; let mergeRequest = null;
const pathFor = item => 'api/v1/repos/' + String(item.repository || '').split('/') const pathFor = item => 'api/v1/repos/' + String(item.repository || '').split('/')
.map(encodeURIComponent).join('/') + '/pulls/' + encodeURIComponent(item.number); .map(encodeURIComponent).join('/') + '/pulls/' + encodeURIComponent(item.number);
const draftKey = item => 'stackchain.pull-comment.v1:' + item.repository + '#' + item.number; const draftKey = item => 'stackchain.pull-comment.v1:' + item.repository + '#' + item.number;
const operationKey = item => draftKey(item) + ':operation';
return { return {
load(item) { load(item) {
@ -27,19 +28,29 @@ function createPullSheet({ fetchJson, storage }) {
catch (_error) { return ''; } catch (_error) { return ''; }
}, },
saveDraft(item, body) { saveDraft(item, body) {
try { storage?.setItem(draftKey(item), body); } try {
if ((storage?.getItem(draftKey(item)) || '') !== body) storage?.removeItem(operationKey(item));
storage?.setItem(draftKey(item), body);
}
catch (_error) { /* The textarea remains the fallback. */ } catch (_error) { /* The textarea remains the fallback. */ }
}, },
comment(item, body) { comment(item, body) {
if (commentRequest) return commentRequest; if (commentRequest) return commentRequest;
this.saveDraft(item, body); this.saveDraft(item, body);
let operationId;
try {
operationId = storage?.getItem(operationKey(item)) || String(createOperationId()).slice(0, 128);
storage?.setItem(operationKey(item), operationId);
} catch (_error) { operationId = String(createOperationId()).slice(0, 128); }
commentRequest = fetchJson(pathFor(item) + '/comments', { commentRequest = fetchJson(pathFor(item) + '/comments', {
method: 'POST', method: 'POST',
headers: { Accept: 'application/json', 'Content-Type': 'application/json' }, headers: { Accept: 'application/json', 'Content-Type': 'application/json', 'Idempotency-Key': operationId },
body: JSON.stringify({ body }), body: JSON.stringify({ body }),
}).then(result => { }).then(result => {
try { storage?.removeItem(draftKey(item)); } try { storage?.removeItem(draftKey(item)); }
catch (_error) { /* The upstream comment is authoritative. */ } catch (_error) { /* The upstream comment is authoritative. */ }
try { storage?.removeItem(operationKey(item)); }
catch (_error) { /* A confirmed result no longer needs replay identity. */ }
return result; return result;
}).finally(() => { commentRequest = null; }); }).finally(() => { commentRequest = null; });
return commentRequest; return commentRequest;

View File

@ -1,4 +1,8 @@
function createReviewController({ fetchJson }) { function createReviewController({
fetchJson,
storage,
createOperationId = () => globalThis.crypto?.randomUUID?.() || String(Date.now()) + '-' + Math.random(),
}) {
let pendingSubmission = null; let pendingSubmission = null;
function endpoint(item) { function endpoint(item) {
@ -16,10 +20,26 @@ function createReviewController({ fetchJson }) {
function submit(item, payload) { function submit(item, payload) {
if (pendingSubmission) return pendingSubmission; if (pendingSubmission) return pendingSubmission;
const operationKey = 'stackchain.review-submit.v1:' + item.repository + '#' + item.number;
const fingerprint = JSON.stringify(payload);
let operationId;
try {
const saved = JSON.parse(storage?.getItem(operationKey) || 'null');
operationId = saved?.fingerprint === fingerprint && saved?.operationId
? saved.operationId : String(createOperationId()).slice(0, 128);
storage?.setItem(operationKey, JSON.stringify({ fingerprint, operationId }));
} catch (_error) { operationId = String(createOperationId()).slice(0, 128); }
pendingSubmission = fetchJson(endpoint(item), { pendingSubmission = fetchJson(endpoint(item), {
method: 'POST', method: 'POST',
headers: { Accept: 'application/json', 'Content-Type': 'application/json' }, headers: {
Accept: 'application/json', 'Content-Type': 'application/json',
'Idempotency-Key': operationId,
},
body: JSON.stringify(payload), body: JSON.stringify(payload),
}).then(result => {
try { storage?.removeItem(operationKey); }
catch (_error) { /* A confirmed result no longer needs replay identity. */ }
return result;
}).finally(() => { pendingSubmission = null; }); }).finally(() => { pendingSubmission = null; });
return pendingSubmission; return pendingSubmission;
} }

View File

@ -1,7 +1,7 @@
import asyncio import asyncio
import math import math
import time import time
from collections.abc import Awaitable from collections.abc import Awaitable, Coroutine
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from pathlib import Path from pathlib import Path
from typing import Any, Literal from typing import Any, Literal
@ -61,6 +61,8 @@ REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
ISSUE_ACTION_TIMEOUT_SECONDS = 5.0 ISSUE_ACTION_TIMEOUT_SECONDS = 5.0
ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS = 600.0 ISSUE_CREATION_IDEMPOTENCY_TTL_SECONDS = 600.0
ISSUE_CREATION_IDEMPOTENCY_MAX_ENTRIES = 256 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_MUTATION_TIMEOUT_SECONDS = 5.0
NOTIFICATION_PAGE_TIMEOUT_SECONDS = 5.0 NOTIFICATION_PAGE_TIMEOUT_SECONDS = 5.0
WORK_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[ _issue_creation_operations: dict[
str, tuple[tuple[Any, ...], asyncio.Task, float] 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_task: asyncio.Task | None = None
_available_issue_snapshot_value: list[dict] | None = None _available_issue_snapshot_value: list[dict] | None = None
_available_issue_snapshot_created_at: float | 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) 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: def _context_payload(user_data, repo_data, issues_data, prs_data) -> dict:
user_model = User( user_model = User(
id=user_data["id"], 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) @app.post("/api/v1/notifications/{thread_id}/reply", status_code=201)
async def reply_to_notification( 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: ) -> JSONResponse:
try: try:
result = await asyncio.wait_for( result = await _run_idempotent_authored_action(
gitea_proxy.reply_to_notification(thread_id, reply.body), 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, timeout=NOTIFICATION_MUTATION_TIMEOUT_SECONDS,
) )
except HTTPException:
raise
except Exception: except Exception:
return JSONResponse( return JSONResponse(
{ {
@ -1129,7 +1186,11 @@ async def create_assigned_issue(
@app.post("/api/v1/repos/{owner}/{repo}/issues/{number}/comments", status_code=201) @app.post("/api/v1/repos/{owner}/{repo}/issues/{number}/comments", status_code=201)
async def comment_on_assigned_issue( 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}" 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) return await gitea_proxy.comment_on_issue(repository, number, comment.body)
try: 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: except HTTPException:
raise raise
except Exception: 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) @app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/comments", status_code=201)
async def comment_on_assigned_pull( 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}" 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) return await gitea_proxy.comment_on_issue(repository, number, comment.body)
try: try:
result = await asyncio.wait_for( result = await _run_idempotent_authored_action(
post_comment(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS post_comment(),
idempotency_key=idempotency_key,
fingerprint=("pull-comment", repository, number, comment.body),
timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
) )
except HTTPException: except HTTPException:
raise raise
@ -1358,7 +1431,11 @@ async def merge_assigned_pull(
@app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/review", status_code=201) @app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/review", status_code=201)
async def submit_review( 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}" repository = f"{owner}/{repo}"
@ -1380,8 +1457,18 @@ async def submit_review(
return await gitea_proxy.submit_pull_review(*args) return await gitea_proxy.submit_pull_review(*args)
try: try:
result = await asyncio.wait_for( comment_fingerprint = tuple(
submit_requested_review(), timeout=REVIEW_DETAIL_TIMEOUT_SECONDS (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: except HTTPException:
raise raise

View File

@ -0,0 +1,142 @@
import asyncio
import httpx
import pytest
from src import main
@pytest.fixture(autouse=True)
def clear_authored_operations():
main._authored_action_operations.clear()
yield
main._authored_action_operations.clear()
@pytest.mark.anyio
async def test_issue_comment_replays_one_upstream_result_for_concurrent_requests(monkeypatch):
calls = 0
started = asyncio.Event()
release = asyncio.Event()
async def assigned(_repository, _number):
return True
async def comment(_repository, _number, body):
nonlocal calls
calls += 1
started.set()
await release.wait()
return {"id": 82, "body": body}
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": "issue-comment-185"}
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
first = asyncio.create_task(client.post(
"/api/v1/repos/stackchain/api/issues/7/comments",
json={"body": "Ship it"}, headers=headers,
))
await started.wait()
second = asyncio.create_task(client.post(
"/api/v1/repos/stackchain/api/issues/7/comments",
json={"body": "Ship it"}, headers=headers,
))
await asyncio.sleep(0)
release.set()
responses = await asyncio.gather(first, second)
assert [response.status_code for response in responses] == [201, 201]
assert calls == 1
@pytest.mark.anyio
async def test_pull_comment_rejects_changed_payload_for_same_key(monkeypatch):
calls = []
async def assigned(_repository, _number):
return True
async def comment(_repository, _number, body):
calls.append(body)
return {"id": len(calls), "body": body}
monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
transport = httpx.ASGITransport(app=main.app)
headers = {"Idempotency-Key": "pull-comment-185"}
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
created = await client.post(
"/api/v1/repos/stackchain/api/pulls/7/comments",
json={"body": "Ship it"}, headers=headers,
)
conflict = await client.post(
"/api/v1/repos/stackchain/api/pulls/7/comments",
json={"body": "Changed"}, headers=headers,
)
assert created.status_code == 201
assert conflict.status_code == 409
assert calls == ["Ship it"]
@pytest.mark.anyio
async def test_notification_reply_retry_recovers_after_caller_timeout(monkeypatch):
calls = 0
async def reply(_thread_id, body):
nonlocal calls
calls += 1
await asyncio.sleep(0.03)
return {"id": 91, "body": body}
monkeypatch.setattr(main, "NOTIFICATION_MUTATION_TIMEOUT_SECONDS", 0.01)
monkeypatch.setattr(main.gitea_proxy, "reply_to_notification", reply)
transport = httpx.ASGITransport(app=main.app)
headers = {"Idempotency-Key": "notification-reply-185"}
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
timed_out = await client.post(
"/api/v1/notifications/42/reply", json={"body": "Retry"}, headers=headers,
)
await asyncio.sleep(0.03)
recovered = await client.post(
"/api/v1/notifications/42/reply", json={"body": "Retry"}, headers=headers,
)
assert timed_out.status_code == 503
assert recovered.status_code == 201
assert recovered.json()["id"] == 91
assert calls == 1
@pytest.mark.anyio
async def test_review_replays_same_key_and_rejects_changed_head(monkeypatch):
calls = []
async def requested(_repository, _number):
return True
async def submit(_repository, _number, head, decision, body):
calls.append((head, decision, body))
return {"id": 93, "state": "APPROVED"}
monkeypatch.setattr(main, "is_requested_review", requested)
monkeypatch.setattr(main.gitea_proxy, "submit_pull_review", submit)
transport = httpx.ASGITransport(app=main.app)
headers = {"Idempotency-Key": "review-185"}
payload = {"expected_head_sha": "abc", "decision": "approve", "body": "Good"}
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
created = await client.post(
"/api/v1/repos/stackchain/api/pulls/7/review", json=payload, headers=headers,
)
replayed = await client.post(
"/api/v1/repos/stackchain/api/pulls/7/review", json=payload, headers=headers,
)
conflict = await client.post(
"/api/v1/repos/stackchain/api/pulls/7/review",
json={**payload, "expected_head_sha": "def"}, headers=headers,
)
assert [created.status_code, replayed.status_code, conflict.status_code] == [201, 201, 409]
assert calls == [("abc", "approve", "Good")]

View File

@ -1888,7 +1888,7 @@ def test_review_controller_submits_once_while_request_is_in_flight():
const createReviewController = require({json.dumps(str(REVIEW_SHEET))}); const createReviewController = require({json.dumps(str(REVIEW_SHEET))});
let resolveRequest; let resolveRequest;
const calls = []; const calls = [];
const controller = createReviewController({{ fetchJson: (url, options) => {{ const controller = createReviewController({{ createOperationId: () => 'review-op-test', fetchJson: (url, options) => {{
calls.push({{url, options}}); calls.push({{url, options}});
return new Promise(resolve => {{ resolveRequest = resolve; }}); return new Promise(resolve => {{ resolveRequest = resolve; }});
}} }}); }} }});
@ -1914,6 +1914,7 @@ Promise.all([first, second]).then(results => process.stdout.write(JSON.stringify
"headers": { "headers": {
"Accept": "application/json", "Accept": "application/json",
"Content-Type": "application/json", "Content-Type": "application/json",
"Idempotency-Key": "review-op-test",
}, },
"body": json.dumps( "body": json.dumps(
{ {
@ -1975,3 +1976,72 @@ async def test_mobile_review_failure_offers_an_in_place_retry_for_the_same_item(
assert "qs('#retry-review-load').hidden = true" in html assert "qs('#retry-review-load').hidden = true" in html
assert "openReviewSheet(selectedReview, reviewTrigger)" in html assert "openReviewSheet(selectedReview, reviewTrigger)" in html
assert "qs('#retry-review-load').focus()" in html assert "qs('#retry-review-load').focus()" in html
def test_issue_comment_reuses_operation_key_after_reload_until_success():
script = f"""
const createIssueSheet = require({json.dumps(str(ISSUE_SHEET))});
const values = new Map();
const storage = {{getItem:k => values.get(k) || null, setItem:(k,v) => values.set(k,v), removeItem:k => values.delete(k)}};
const calls = [];
const item = {{repository:'stackchain/api', number:7}};
const first = createIssueSheet({{storage, createOperationId:() => 'issue-op-185', fetchJson:(_u,o) => {{calls.push(o.headers['Idempotency-Key']); return Promise.reject(new Error('timeout'));}}}});
first.comment(item, 'Ship it').catch(() => {{
const second = createIssueSheet({{storage, createOperationId:() => 'wrong-key', fetchJson:(_u,o) => {{calls.push(o.headers['Idempotency-Key']); return Promise.resolve({{id:82}});}}}});
second.comment(item, 'Ship it').then(() => process.stdout.write(JSON.stringify({{calls, keys:[...values.keys()]}})));
}});
"""
result = subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True)
assert json.loads(result.stdout) == {"calls": ["issue-op-185", "issue-op-185"], "keys": []}
def test_pull_comment_reuses_operation_key_after_reload_until_success():
script = f"""
const createPullSheet = require({json.dumps(str(PULL_SHEET))});
const values = new Map();
const storage = {{getItem:k => values.get(k) || null, setItem:(k,v) => values.set(k,v), removeItem:k => values.delete(k)}};
const calls = [];
const item = {{repository:'stackchain/api', number:7}};
const first = createPullSheet({{storage, createOperationId:() => 'pull-op-185', fetchJson:(_u,o) => {{calls.push(o.headers['Idempotency-Key']); return Promise.reject(new Error('timeout'));}}}});
first.comment(item, 'Ship it').catch(() => {{
createPullSheet({{storage, createOperationId:() => 'wrong-key', fetchJson:(_u,o) => {{calls.push(o.headers['Idempotency-Key']); return Promise.resolve({{id:91}});}}}}).comment(item, 'Ship it')
.then(() => process.stdout.write(JSON.stringify({{calls, keys:[...values.keys()]}})));
}});
"""
result = subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True)
assert json.loads(result.stdout) == {"calls": ["pull-op-185", "pull-op-185"], "keys": []}
def test_notification_reply_reuses_persisted_operation_key():
script = f"""
const build = require({json.dumps(str(MY_WORK))});
const values = new Map();
const storage = {{getItem:k => values.get(k) || null, setItem:(k,v) => values.set(k,v), removeItem:k => values.delete(k)}};
const calls = [];
const item = {{notification_id:42}};
const first = build.createNotificationReplier({{storage, onStatus:()=>{{}}, createOperationId:() => 'reply-op-185', post:(_id,_body,key) => {{calls.push(key); return Promise.reject(new Error('timeout'));}}}});
first.submit(item, 'Retry').then(() => {{
const second = build.createNotificationReplier({{storage, onStatus:()=>{{}}, createOperationId:() => 'wrong-key', post:(_id,_body,key) => {{calls.push(key); return Promise.resolve({{id:91}});}}}});
second.submit(item, 'Retry').then(() => process.stdout.write(JSON.stringify({{calls, keys:[...values.keys()]}})));
}});
"""
result = subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True)
assert json.loads(result.stdout) == {"calls": ["reply-op-185", "reply-op-185"], "keys": []}
def test_review_submission_reuses_persisted_operation_key():
script = f"""
const createReviewController = require({json.dumps(str(REVIEW_SHEET))});
const values = new Map();
const storage = {{getItem:k => values.get(k) || null, setItem:(k,v) => values.set(k,v), removeItem:k => values.delete(k)}};
const calls = [];
const item = {{repository:'stackchain/api', number:7}};
const payload = {{decision:'approve', body:'Good', expected_head_sha:'abc', comments:[]}};
const first = createReviewController({{storage, createOperationId:() => 'review-op-185', fetchJson:(_u,o) => {{calls.push(o.headers['Idempotency-Key']); return Promise.reject(new Error('timeout'));}}}});
first.submit(item, payload).catch(() => {{
createReviewController({{storage, createOperationId:() => 'wrong-key', fetchJson:(_u,o) => {{calls.push(o.headers['Idempotency-Key']); return Promise.resolve({{id:93}});}}}}).submit(item, payload)
.then(() => process.stdout.write(JSON.stringify({{calls, keys:[...values.keys()]}})));
}});
"""
result = subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True)
assert json.loads(result.stdout) == {"calls": ["review-op-185", "review-op-185"], "keys": []}