Make mobile authored actions idempotent across retries #186
|
|
@ -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
|
||||||
|
|
|
||||||
|
|
@ -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(() => ({}));
|
||||||
|
|
|
||||||
|
|
@ -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;
|
||||||
|
|
|
||||||
|
|
@ -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) {
|
||||||
|
|
|
||||||
|
|
@ -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;
|
||||||
|
|
|
||||||
|
|
@ -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;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
109
src/main.py
109
src/main.py
|
|
@ -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
|
||||||
|
|
|
||||||
142
tests/test_authored_idempotency.py
Normal file
142
tests/test_authored_idempotency.py
Normal 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")]
|
||||||
|
|
@ -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": []}
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user