fix: recover orphaned authored operations (#319)
CI / lint (pull_request) Successful in 32s
CI / build-frontend (pull_request) Successful in 4s

This commit is contained in:
timmy committed 2026-08-08 16:40:31 +00:00
1 parent b5858046ee
commit dc18004e02
15 files changed
+321 -18

No files matched your search

+93
View File
@@ -116,6 +116,54 @@ async def test_notification_reply_retry_recovers_after_caller_timeout(monkeypatc
assert calls == 1
@pytest.mark.anyio
async def test_live_authored_task_remains_joinable_after_reservation_age_threshold(
monkeypatch, tmp_path
):
calls = 0
now = [1_000.0]
release = asyncio.Event()
async def reply(_thread_id, body):
nonlocal calls
calls += 1
await release.wait()
return {"id": 319, "body": body}
monkeypatch.setattr(
main,
"_idempotency_ledger",
IdempotencyLedger(
tmp_path / "live.sqlite3",
ttl_seconds=1,
max_entries=256,
clock=lambda: now[0],
),
)
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": "live-after-threshold-319"}
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
timed_out = await client.post(
"/api/v1/notifications/42/reply", json={"body": "Join me"}, headers=headers,
)
now[0] += 2
monkeypatch.setattr(main, "NOTIFICATION_MUTATION_TIMEOUT_SECONDS", 1.0)
joined = asyncio.create_task(client.post(
"/api/v1/notifications/42/reply", json={"body": "Join me"}, headers=headers,
))
await asyncio.sleep(0.02)
assert not joined.done(), "a live local operation must be joined instead of rejected as stale"
release.set()
recovered = await joined
assert timed_out.status_code == 503
assert recovered.status_code == 201
assert calls == 1
@pytest.mark.anyio
async def test_review_replays_same_key_and_rejects_changed_head(monkeypatch):
calls = []
@@ -233,6 +281,51 @@ async def test_orphaned_pending_comment_fails_closed_without_upstream_retry(
assert calls == 0
@pytest.mark.anyio
async def test_stale_orphaned_comment_returns_non_retryable_uncertain_delivery(
monkeypatch, tmp_path
):
calls = 0
now = [1_000.0]
async def assigned(_repository, _number):
return True
async def comment(_repository, _number, _body):
nonlocal calls
calls += 1
return {"id": 319}
ledger = IdempotencyLedger(
tmp_path / "stale-orphan.sqlite3",
ttl_seconds=10,
max_entries=256,
clock=lambda: now[0],
)
fingerprint = ("issue-comment", "stackchain/api", 7, "Verify first")
assert ledger.reserve("stale-orphan-319", fingerprint).state == "reserved"
now[0] += 11
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": "stale-orphan-319"},
)
assert response.status_code == 422
assert response.json()["detail"] == {
"code": "delivery_uncertain",
"message": "Delivery could not be confirmed. Verify it was not posted before retrying.",
}
assert "Retry-After" not in response.headers
assert calls == 0
@pytest.mark.anyio
async def test_slow_ledger_reservation_does_not_block_health_requests(monkeypatch):
async def assigned(_repository, _number):
+46
View File
@@ -113,6 +113,37 @@ outbox.flush('timmy').then(async first => {{
assert output["remaining"][1]["status"] == "queued"
def test_authored_outbox_requires_deliberate_rekey_after_uncertain_delivery():
script = f"""
const createAuthoredOutbox = require({json.dumps(str(OUTBOX))});
const values = new Map(); let sequence=0; const calls=[];
const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
const outbox = createAuthoredOutbox({{
storage, getOwnerLogin:()=>'timmy', createOperationId:()=> 'verified-' + (++sequence),
fetchJson:async (_url, options) => {{
const key=options.headers['Idempotency-Key']; calls.push(key);
if (key === 'uncertain-key') {{ const e=new Error('Verify it was not posted before retrying.'); e.status=422; e.code='delivery_uncertain'; throw e; }}
return {{id:calls.length}};
}},
}});
const uncertain=outbox.enqueue({{kind:'issue-comment',repository:'o/r',number:1,body:'Check',operationId:'uncertain-key'}});
outbox.enqueue({{kind:'pull-comment',repository:'o/r',number:2,body:'Continue',operationId:'other-key'}});
(async()=>{{
const first=await outbox.flush('timmy');
const attention=outbox.list()[0];
const retried=await outbox.retry(uncertain.id,'timmy');
process.stdout.write(JSON.stringify({{first,attention,retried,calls,remaining:outbox.list()}}));
}})();
"""
output = run_node(script)
assert output["calls"] == ["uncertain-key", "other-key", "verified-1"]
assert len(output["first"]["confirmed"]) == 1
assert output["attention"]["status"] == "attention"
assert output["attention"]["deliveryState"] == "uncertain"
assert output["remaining"] == []
def test_authored_outbox_retry_is_single_flight_and_edit_rotates_identity():
script = f"""
const createAuthoredOutbox = require({json.dumps(str(OUTBOX))});
@@ -182,6 +213,21 @@ setTimeout(() => {{
]
def test_authored_outbox_reconciles_uncertain_worker_delivery_state():
script = f"""
const createAuthoredOutbox = require({json.dumps(str(OUTBOX))});
const values=new Map(); const storage={{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
const outbox=createAuthoredOutbox({{storage,getOwnerLogin:()=>'timmy'}});
const item=outbox.enqueue({{kind:'issue-comment',repository:'o/r',number:7,body:'Maybe',operationId:'maybe'}});
const reconciled=outbox.reconcileBackground([{{...item,status:'attention',deliveryState:'uncertain',error:'Verify first'}}]);
process.stdout.write(JSON.stringify(reconciled[0]));
"""
output = run_node(script)
assert output["status"] == "attention"
assert output["deliveryState"] == "uncertain"
def test_authored_outbox_waits_for_durable_background_admission():
script = f"""
const createAuthoredOutbox = require({json.dumps(str(OUTBOX))});
+25
View File
@@ -585,6 +585,31 @@ const fetchJson = async url => {{
assert output["result"]["attention"] == 1
def test_uncertain_background_delivery_preserves_verification_state():
script = f"""
const createBackgroundIssueSync = require({json.dumps(str(SYNC))});
let claimed=false; const state={{failed:[]}};
const store={{
claimNext:async()=>claimed?null:(claimed=true,{{id:'maybe',operationId:'maybe',ownerLogin:'timmy',repository:'o/r',title:'Maybe',labelIds:[]}}),
fail:async(id,message,deliveryState)=>state.failed.push({{id,message,deliveryState}}),
release:async()=>{{}}, countBlocked:async()=>0,
}};
const fetchJson=async url=>{{
if(url==='api/v1/background-identity') return {{login:'timmy'}};
const error=new Error('Verify first'); error.status=422; error.code='delivery_uncertain'; throw error;
}};
(async()=>{{
await createBackgroundIssueSync({{store,fetchJson}}).flush();
process.stdout.write(JSON.stringify(state));
}})();
"""
output = run_node(script)
assert output["failed"] == [
{"id": "maybe", "message": "Verify first", "deliveryState": "uncertain"}
]
def test_issue_store_atomically_grants_one_delivery_claim():
script = f"""
const createBackgroundIssueSync = require({json.dumps(str(SYNC))});
+19
View File
@@ -121,6 +121,22 @@ process.stdout.write(JSON.stringify(drafts));
assert output[1]["label"] == "Queued issue"
def test_draft_inbox_exposes_uncertain_delivery_for_explicit_user_verification():
script = f"""
const createDraftInbox = require({json.dumps(str(DRAFTS))});
const values = new Map([['stackchain.authored-outbox.v1', JSON.stringify({{version:2,items:[
{{id:'uncertain',kind:'issue-comment',repository:'o/r',number:7,body:'Possibly posted',ownerLogin:'timmy',status:'attention',deliveryState:'uncertain',error:'Verify it was not posted before retrying.',queuedAt:200}}
]}})]]);
const storage = {{get length(){{return values.size}},key:i=>Array.from(values.keys())[i]||null,getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
const drafts = createDraftInbox({{storage,getCurrentLogin:()=>'timmy'}}).list();
process.stdout.write(JSON.stringify(drafts[0]));
"""
output = run_node(script)
assert output["delivery_state"] == "uncertain"
assert output["label"] == "Verify delivery"
def test_draft_inbox_marks_mismatched_and_legacy_outbox_content_copy_only():
script = f"""
const createDraftInbox = require({json.dumps(str(DRAFTS))});
@@ -160,6 +176,9 @@ async def test_mobile_dashboard_exposes_touch_safe_draft_recovery_lane():
assert '.draft-actions button { min-height:44px;' in html
assert 'createDraftInbox({ storage: localStorage' in html
assert "captureDraft.repository && !repositories.includes(captureDraft.repository)" in html
assert "Verified not posted — retry" in html
assert "payload.detail?.message" in html
assert "error.code = payload.detail?.code" in html
@pytest.mark.anyio
+18
View File
@@ -71,3 +71,21 @@ def test_independent_connections_atomically_reserve_one_operation(tmp_path):
states = list(executor.map(reserve, ledgers))
assert sorted(states) == ["pending", "reserved"]
def test_stale_pending_reservation_becomes_uncertain_without_consuming_capacity(tmp_path):
now = [1_000.0]
ledger = IdempotencyLedger(
tmp_path / "orphaned.sqlite3",
ttl_seconds=10,
max_entries=1,
clock=lambda: now[0],
)
fingerprint = ("issue-comment", "stackchain/api", 7, "Verify first")
assert ledger.reserve("orphan", fingerprint).state == "reserved"
now[0] += 11
assert ledger.reserve("orphan", fingerprint).state == "uncertain"
assert ledger.reserve("unrelated", ("issue-comment", "stackchain/api", 8, "Continue")).state == "reserved"
assert ledger.reserve("orphan", ("issue-comment", "stackchain/api", 7, "Changed")).state == "conflict"
+28
View File
@@ -139,6 +139,34 @@ outbox.flush('timmy').then(async () => {{
assert output["remaining"] == []
def test_issue_outbox_rekeys_only_after_user_retries_uncertain_delivery():
script = f"""
const createIssueOutbox = require({json.dumps(str(OUTBOX))});
const values = new Map(); let sequence=0; const calls=[];
const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}};
const outbox = createIssueOutbox({{
storage, getOwnerLogin:()=>'timmy', createOperationId:()=> 'operation-' + (++sequence),
fetchJson:async (_url, options) => {{
const key=options.headers['Idempotency-Key']; calls.push(key);
if (key === 'operation-1') {{ const e=new Error('Verify it was not posted before retrying.'); e.status=422; e.code='delivery_uncertain'; throw e; }}
return {{number:319}};
}},
}});
const item=outbox.enqueue({{repository:'o/r',title:'Possibly sent',body:'Check first'}});
(async()=>{{
await outbox.flush('timmy'); const attention=outbox.list()[0];
await outbox.retry(item.id,'timmy');
process.stdout.write(JSON.stringify({{attention,calls,remaining:outbox.list()}}));
}})();
"""
output = run_node(script)
assert output["attention"]["status"] == "attention"
assert output["attention"]["deliveryState"] == "uncertain"
assert output["calls"] == ["operation-1", "operation-2"]
assert output["remaining"] == []
def test_issue_outbox_rotates_operation_id_only_when_delivery_payload_changes():
script = f"""
const createIssueOutbox = require({json.dumps(str(OUTBOX))});
+19 -2
View File
@@ -97,7 +97,7 @@ async function dispatchNotificationClick(route) {{
def test_share_target_sign_in_fix_ships_in_a_new_shell_cache():
source = WORKER.read_text()
assert "stackchain-dashboard-shell-v33" in source
assert "stackchain-dashboard-shell-v34" in source
assert "BASE + 'static/dashboard.css'" in source
assert "BASE + 'static/dashboard.js'" in source
assert "BASE + 'static/install-app.js'" in source
@@ -106,7 +106,7 @@ def test_share_target_sign_in_fix_ships_in_a_new_shell_cache():
def test_mobile_search_viewport_ships_in_a_new_offline_shell():
source = WORKER.read_text()
assert "stackchain-dashboard-shell-v33" in source
assert "stackchain-dashboard-shell-v34" in source
assert "BASE + 'static/mobile-search-viewport.js'" in source
@@ -122,6 +122,23 @@ def test_background_sync_event_flushes_closed_app_issue_outbox_only_for_its_tag(
assert result["backgroundFlushes"] == 1
def test_background_delivery_preserves_structured_uncertain_error():
result = run_worker_scenario(
"""
context.fetch=async()=>new Response(JSON.stringify({detail:{code:'delivery_uncertain',message:'Verify it was not posted before retrying.'}}),{status:422,headers:{'Content-Type':'application/json'}});
const outcome=await context.self.__testFetchJson('/dashboard/api/v1/repos/o/r/issues',{method:'POST'})
.then(()=>({ok:true}),error=>({status:error.status,code:error.code,message:error.message}));
process.stdout.write(JSON.stringify(outcome));
"""
)
assert result == {
"status": 422,
"code": "delivery_uncertain",
"message": "Verify it was not posted before retrying.",
}
def test_authenticated_page_message_resumes_queued_background_delivery():
result = run_worker_scenario(
"""