From 31b0e79aa1f751d95805d1668af83bcd040f8d29 Mon Sep 17 00:00:00 2001 From: timmy Date: Mon, 10 Aug 2026 11:21:53 +0000 Subject: [PATCH] fix: fence background delivery claims (Closes #479) --- README.md | 6 +- frontend/background-issue-sync.js | 143 +++++++++++++++++++++------- tests/test_background_issue_sync.py | 138 +++++++++++++++++++++++++-- 3 files changed, 244 insertions(+), 43 deletions(-) diff --git a/README.md b/README.md index 9672492..18f682d 100644 --- a/README.md +++ b/README.md @@ -106,7 +106,11 @@ refresh still run immediately; a successful response resets transport backoff. Closed-app delivery requests are deadline-bounded: a stalled identity, CSRF, or mutation request is aborted after 15 seconds, its durable claim returns to the queue, and the next sync retries -with the unchanged idempotency key. Device purge cancels an active drain before closing +with the unchanged idempotency key. Each finite, lane-fair drain admits at most 70 records +and runs up to three deliveries concurrently, but claims a record only when a delivery slot +is ready. Claims have unique fencing tokens and are renewed before every network stage; +completion, release, failure, and delivery checkpoints are token-fenced so an expired worker +cannot alter a newer crash-recovery claim. Device purge cancels an active drain before closing private outbox storage. Results are coordinated through a bounded SQLite ledger. Set `STACKCHAIN_STATE_DIR` to a persistent, writable service directory (or set `STACKCHAIN_IDEMPOTENCY_DB` to an explicit SQLite path); the local default is `.stackchain-state/idempotency.sqlite3`. Ledger reads and diff --git a/frontend/background-issue-sync.js b/frontend/background-issue-sync.js index 9709cbc..5b8e09b 100644 --- a/frontend/background-issue-sync.js +++ b/frontend/background-issue-sync.js @@ -55,7 +55,11 @@ function createIndexedDbTransaction(indexedDB, dbName = 'stackchain-background-o return transact; } -function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, now = () => Date.now(), claimMs = 30000 } = {}) { +function createIssueSyncStore({ + transaction, indexedDB = globalThis.indexedDB, now = () => Date.now(), claimMs = 30000, + createToken = () => globalThis.crypto?.randomUUID?.() || + (Date.now().toString(36) + '-' + Math.random().toString(36).slice(2)), +} = {}) { const transact = transaction || createIndexedDbTransaction(indexedDB); async function reconcile(items, outboxLane = 'issue') { @@ -96,13 +100,15 @@ function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, n (candidate.status === 'queued' || candidate.status === 'sending') && (candidate.status !== 'sending' || Number(candidate.claimUntil) <= timestamp)); if (!item) return null; - const claimed = { ...item, status: 'sending', claimUntil: timestamp + claimMs }; + const claimed = { + ...item, status: 'sending', claimUntil: timestamp + claimMs, claimToken: createToken(), + }; await records.put(claimed); return claimed; }); } - async function claimBatch(ownerLogin, limit = 70) { + async function planBatch(ownerLogin, limit = 70) { return transact(async records => { const timestamp = Number(now()); const eligible = (await records.getAll()).filter(candidate => @@ -117,14 +123,10 @@ function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, n const selected = []; const maximum = Math.max(0, Number(limit) || 0); while (selected.length < maximum && (lanes.issue.length || lanes.authored.length)) { - if (lanes.issue.length) selected.push(lanes.issue.shift()); - if (selected.length < maximum && lanes.authored.length) selected.push(lanes.authored.shift()); + if (lanes.issue.length) selected.push(lanes.issue.shift().id); + if (selected.length < maximum && lanes.authored.length) selected.push(lanes.authored.shift().id); } - const claimed = selected.map(item => ({ - ...item, status: 'sending', claimUntil: timestamp + claimMs, - })); - for (const item of claimed) await records.put(item); - return claimed; + return selected; }); } @@ -135,7 +137,9 @@ function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, n if (!item || item.ownerLogin !== ownerLogin || !['queued', 'sending'].includes(item.status) || (item.status === 'sending' && Number(item.claimUntil) > timestamp)) return null; - const claimed = { ...item, status: 'sending', claimUntil: timestamp + claimMs }; + const claimed = { + ...item, status: 'sending', claimUntil: timestamp + claimMs, claimToken: createToken(), + }; await records.put(claimed); return claimed; }); @@ -166,6 +170,32 @@ function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, n }); } + async function updateClaim(id, claimToken, transform) { + return transact(async records => { + const item = records.get ? await records.get(id) : + (await records.getAll()).find(candidate => candidate.id === id); + if (!item || item.status !== 'sending' || !claimToken || item.claimToken !== claimToken) { + return false; + } + await records.put(transform(item)); + return true; + }); + } + + function clearClaim(item, changes) { + const { claimToken: _claimToken, ...unclaimed } = item; + return { ...unclaimed, ...changes, claimUntil: 0 }; + } + + async function renew(id, claimToken) { + let renewed = null; + await updateClaim(id, claimToken, item => { + renewed = { ...item, claimUntil: Number(now()) + claimMs }; + return renewed; + }); + return renewed; + } + function preferenceId(ownerLogin) { return 'receipt-preference:' + String(ownerLogin || '').trim(); } @@ -194,14 +224,18 @@ function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, n update, claim, claimNext, - claimBatch, - complete: (id, deliveredIssue) => update(id, item => ({ - ...item, status: 'sent', claimUntil: 0, + planBatch, + supportsClaimTokens: true, + renew, + checkpoint: updateClaim, + complete: (id, claimToken, deliveredIssue) => updateClaim(id, claimToken, item => clearClaim(item, { + status: 'sent', ...(item.completionIntent === 'create-and-start' && deliveredIssue ? { deliveredIssue } : {}), })), - release: id => update(id, item => ({ ...item, status: 'queued', claimUntil: 0 })), - fail: (id, error, deliveryState) => update(id, item => ({ - ...item, status: 'attention', claimUntil: 0, error, + release: (id, claimToken) => updateClaim(id, claimToken, + item => clearClaim(item, { status: 'queued' })), + fail: (id, claimToken, error, deliveryState) => updateClaim(id, claimToken, item => clearClaim(item, { + status: 'attention', error, ...(deliveryState ? { deliveryState } : {}), })), snapshot: () => transact(async records => @@ -225,6 +259,24 @@ function createBackgroundIssueSync({ const activeRequests = new Set(); const timeoutMs = Math.max(1, Number(requestTimeoutMs) || 15000); + const completeClaim = (item, delivered) => store.supportsClaimTokens + ? store.complete(item.id, item.claimToken, delivered) : store.complete(item.id, delivered); + const releaseClaim = item => store.supportsClaimTokens + ? store.release(item.id, item.claimToken) : store.release(item.id); + const failClaim = (item, error, deliveryState) => store.supportsClaimTokens + ? store.fail(item.id, item.claimToken, error, deliveryState) + : store.fail(item.id, error, deliveryState); + const checkpointClaim = (item, transform) => store.supportsClaimTokens + ? store.checkpoint(item.id, item.claimToken, transform) : store.update?.(item.id, transform); + + async function requestStage(item, url, options) { + if (store.renew) { + const renewed = await store.renew(item.id, item.claimToken); + if (!renewed) throw new Error('Background delivery claim was lost.'); + } + return requestJson(url, options); + } + async function requestJson(url, options = {}) { const controller = new AbortController(); activeRequests.add(controller); @@ -368,12 +420,13 @@ function createBackgroundIssueSync({ let deliveredIssue = item.deliveredIssue; if (!deliveredIssue) { const request = deliveryRequest(item); - deliveredIssue = await requestJson(request.url, request.options); - await store.update?.(item.id, current => ({ ...current, deliveredIssue })); + deliveredIssue = await requestStage(item, request.url, request.options); + await checkpointClaim(item, current => ({ ...current, deliveredIssue })); } let attachmentMarkdown = item.attachmentMarkdown; if (!attachmentMarkdown) { - const uploaded = await requestJson( + const uploaded = await requestStage( + item, base + 'api/v1/repos/' + repository + '/issues/' + encodeURIComponent(deliveredIssue.number) + '/attachments', { method: 'POST', @@ -394,9 +447,10 @@ function createBackgroundIssueSync({ error.status = 422; throw error; } - await store.update?.(item.id, current => ({ ...current, deliveredIssue, attachmentMarkdown })); + await checkpointClaim(item, current => ({ ...current, deliveredIssue, attachmentMarkdown })); } - await requestJson( + await requestStage( + item, base + 'api/v1/repos/' + repository + '/issues/' + encodeURIComponent(deliveredIssue.number) + '/comments', { method: 'POST', @@ -414,7 +468,8 @@ function createBackgroundIssueSync({ const repository = String(item.repository || '').split('/').map(encodeURIComponent).join('/'); let attachmentMarkdown = item.attachmentMarkdown; if (!attachmentMarkdown) { - const uploaded = await requestJson( + const uploaded = await requestStage( + item, base + 'api/v1/repos/' + repository + '/issues/' + encodeURIComponent(item.number) + '/attachments', { method: 'POST', @@ -435,10 +490,11 @@ function createBackgroundIssueSync({ error.status = 422; throw error; } - await store.update?.(item.id, current => ({ ...current, attachmentMarkdown })); + await checkpointClaim(item, current => ({ ...current, attachmentMarkdown })); } const text = String(item.body || '').trim(); - return requestJson( + return requestStage( + item, base + 'api/v1/repos/' + repository + '/issues/' + encodeURIComponent(item.number) + '/comments', { method: 'POST', @@ -456,30 +512,30 @@ function createBackgroundIssueSync({ try { const delivered = item.attachment && item.kind === 'issue-comment' ? await deliverScreenshotComment(item) : item.attachment && !item.kind ? - await deliverIssueCapture(item) : await requestJson(request.url, request.options); + await deliverIssueCapture(item) : await requestStage(item, request.url, request.options); if (item.kind === 'issue-close' && delivered?.state !== 'closed') { const error = new Error('Issue closure was not confirmed.'); error.status = 422; throw error; } - await store.complete(item.id, delivered); + await completeClaim(item, delivered); const receipt = receiptFor(item, 'confirmed', delivered); return item.kind ? { message: delivered, receipt } : { issue: delivered, receipt }; } catch (error) { const status = Number(error?.status || 0); if (status === 401) { - await store.release(item.id); + await releaseClaim(item); throw error; } if (status >= 400 && status < 500) { - await store.fail( - item.id, + await failClaim( + item, String(error?.message || 'Issue needs attention').slice(0, 240), error?.code === 'delivery_uncertain' ? 'uncertain' : undefined, ); return { attention: true, error, receipt: receiptFor(item, 'attention') }; } - await store.release(item.id); + await releaseClaim(item); throw error; } } @@ -508,7 +564,30 @@ function createBackgroundIssueSync({ if (result.attention) attention += 1; if (result.receipt) receipts.push(result.receipt); }; - if (store.claimBatch) { + if (store.planBatch) { + const planned = await store.planBatch(login, batchSize); + let next = 0; + let authenticationError = null; + let transientError = null; + const worker = async () => { + while (!purgeRequested && !authenticationError && next < planned.length) { + const id = planned[next++]; + const item = await store.claim(id, login); + if (!item) continue; + try { + collect(await deliver(item)); + } catch (error) { + if (Number(error?.status || 0) === 401) authenticationError = error; + else if (!transientError) transientError = error; + } + } + }; + const concurrency = Math.max(1, Math.min(Number(maxConcurrency) || 1, planned.length)); + await Promise.all(Array.from({ length: concurrency }, worker)); + if (purgeRequested) throw new Error('Background delivery canceled.'); + if (authenticationError) throw authenticationError; + if (transientError) throw transientError; + } else if (store.claimBatch) { const claimed = await store.claimBatch(login, batchSize); let next = 0; let authenticationError = null; diff --git a/tests/test_background_issue_sync.py b/tests/test_background_issue_sync.py index 313cc38..5ccbd6b 100644 --- a/tests/test_background_issue_sync.py +++ b/tests/test_background_issue_sync.py @@ -129,6 +129,40 @@ createBackgroundIssueSync({{store,fetchJson}}).flush().then(()=>process.stdout.w assert all(len(key) <= 128 for key in keys) +def test_capture_renews_claim_before_every_network_stage_and_fences_completion(): + script = f""" +const createBackgroundIssueSync=require({json.dumps(str(SYNC))}); +const item={{id:'capture',claimToken:'claim-7',operationId:'capture',ownerLogin:'timmy',repository:'o/r',title:'Bug',body:'',labelIds:[],attachment:{{filename:'a.png',contentType:'image/png',data:'abc'}}}}; +const events=[];let claimed=false; +const store={{ + supportsClaimTokens:true, + claimNext:async()=>claimed?null:(claimed=true,item), + renew:async(id,token)=>{{events.push(['renew',id,token]);return item;}}, + checkpoint:async(id,token,transform)=>{{events.push(['checkpoint',id,token]);transform(item);return true;}}, + complete:async(id,token)=>events.push(['complete',id,token]), + release:async(id,token)=>events.push(['release',id,token]), + fail:async()=>{{}},countBlocked:async()=>0, +}}; +const fetchJson=async(url,options={{}})=>{{ + if(url==='api/v1/background-identity')return{{login:'timmy'}}; + events.push(['network',options.headers['Idempotency-Key']]); + if(url.endsWith('/issues'))return{{number:3}}; + if(url.endsWith('/attachments'))return{{markdown:'![a](url)'}}; + return{{id:4}}; +}}; +(async()=>{{await createBackgroundIssueSync({{store,fetchJson}}).flush();process.stdout.write(JSON.stringify(events));}})(); +""" + events = run_node(script) + + networks = [index for index, event in enumerate(events) if event[0] == "network"] + assert [events[index - 1][0] for index in networks] == ["renew", "renew", "renew"] + assert [events[index][1] for index in networks] == [ + "capture", "capture:attachment", "capture:attachment-comment" + ] + assert events[-1] == ["complete", "capture", "claim-7"] + assert all(event[2] == "claim-7" for event in events if event[0] in {"renew", "checkpoint"}) + + def test_screenshot_comment_retry_reuses_upload_checkpoint_and_posts_combined_body_once(): script = f""" const createBackgroundIssueSync=require({json.dumps(str(SYNC))}); @@ -758,6 +792,52 @@ const fetchJson = async (url, options={{}}) => {{ assert output["error"] == "Temporary outage" +def test_bounded_flush_claims_just_in_time_and_keeps_unadmitted_records_queued(): + script = f""" +const createBackgroundIssueSync=require({json.dumps(str(SYNC))}); +const records=new Map();let tail=Promise.resolve(); +const transaction=work=>{{const run=tail.then(()=>work({{ + get:async id=>records.get(id), + getAll:async()=>[...records.values()].map(value=>({{...value}})), + put:async value=>records.set(value.id,{{...value}}),delete:async id=>records.delete(id), +}}));tail=run.catch(()=>{{}});return run;}}; +let allow=false;const waiting=[];const attempted=[]; +const fetchJson=async(url,options={{}})=>{{ + if(url==='api/v1/background-identity')return{{login:'timmy'}}; + attempted.push(options.headers['Idempotency-Key']); + if(!allow)await new Promise(resolve=>waiting.push(resolve)); + return{{number:attempted.length}}; +}}; +(async()=>{{ + const store=createBackgroundIssueSync.createIssueSyncStore({{transaction,createToken:(()=>{{let n=0;return()=>`token-${{++n}}`;}})()}}); + await store.reconcile(Array.from({{length:4}},(_,i)=>({{id:`issue-${{i}}`,operationId:`issue-${{i}}`,ownerLogin:'timmy',status:'queued',repository:'o/r',title:'Issue',labelIds:[]}})),'issue'); + await store.reconcile(Array.from({{length:2}},(_,i)=>({{id:`message-${{i}}`,operationId:`message-${{i}}`,ownerLogin:'timmy',status:'queued',kind:'issue-comment',repository:'o/r',number:i,body:'Reply'}})),'authored'); + const sync=createBackgroundIssueSync({{store,fetchJson,maxConcurrency:2,batchSize:5}}); + const flushing=sync.flush(); + while(waiting.length<2)await new Promise(resolve=>setTimeout(resolve,0)); + const during=await store.snapshot(); + allow=true;waiting.splice(0).forEach(resolve=>resolve()); + await flushing; + const after=await store.snapshot(); + process.stdout.write(JSON.stringify({{attempted,during,after}})); +}})(); +""" + output = run_node(script) + + assert [item["id"] for item in output["during"] if item["status"] == "sending"] == [ + "issue-0", "message-0" + ] + assert [item["id"] for item in output["during"] if item["status"] == "queued"] == [ + "issue-1", "issue-2", "issue-3", "message-1" + ] + assert output["attempted"] == [ + "issue-0", "message-0", "issue-1", "message-1", "issue-2" + ] + unadmitted = next(item for item in output["after"] if item["id"] == "issue-3") + assert unadmitted["status"] == "queued" + assert "claimToken" not in unadmitted + + def test_bounded_flush_stops_admission_and_releases_unstarted_claims_on_auth_loss(): script = f""" const createBackgroundIssueSync = require({json.dumps(str(SYNC))}); @@ -812,18 +892,15 @@ const transaction=work=>{{const run=tail.then(()=>work({{ {{id:'message-1',ownerLogin:'timmy',status:'queued'}}, {{id:'message-2',ownerLogin:'timmy',status:'queued'}}, ], 'authored'); - const claimed=await store.claimBatch('timmy', 4); - process.stdout.write(JSON.stringify({{claimed,snapshot:await store.snapshot()}})); + const planned=await store.planBatch('timmy', 4); + process.stdout.write(JSON.stringify({{planned,snapshot:await store.snapshot()}})); }})(); """ output = run_node(script) - assert [item["id"] for item in output["claimed"]] == [ - "issue-1", "message-1", "issue-2", "message-2" - ] - assert all(item["status"] == "sending" for item in output["claimed"]) - assert all(item["claimUntil"] == 5100 for item in output["claimed"]) - assert next(item for item in output["snapshot"] if item["id"] == "issue-3")["status"] == "queued" + assert output["planned"] == ["issue-1", "message-1", "issue-2", "message-2"] + assert all(item["status"] == "queued" for item in output["snapshot"]) + assert all("claimUntil" not in item for item in output["snapshot"]) def test_permanent_delivery_failure_marks_issue_for_foreground_attention(): @@ -908,6 +985,47 @@ const transaction = work => {{ assert output["records"][0]["claimUntil"] == 6000 +def test_issue_store_renews_and_fences_claim_mutations_with_unique_tokens(): + script = f""" +const createBackgroundIssueSync = require({json.dumps(str(SYNC))}); +const records=new Map();let tail=Promise.resolve();let timestamp=100; +const transaction=work=>{{const run=tail.then(()=>work({{ + get:async id=>records.get(id), + getAll:async()=>[...records.values()].map(value=>({{...value}})), + put:async value=>records.set(value.id,{{...value}}),delete:async id=>records.delete(id), +}}));tail=run.catch(()=>{{}});return run;}}; +(async()=>{{ + const tokens=['claim-a','claim-b']; + const store=createBackgroundIssueSync.createIssueSyncStore({{ + transaction,now:()=>timestamp,claimMs:30,createToken:()=>tokens.shift(), + }}); + await store.reconcile([{{id:'lease',ownerLogin:'timmy',status:'queued'}}]); + const first=await store.claimNext('timmy'); + timestamp=120; + const renewed=await store.renew('lease',first.claimToken); + timestamp=151; + const second=await store.claimNext('timmy'); + const staleComplete=await store.complete('lease',first.claimToken); + const staleRelease=await store.release('lease',first.claimToken); + const staleFail=await store.fail('lease',first.claimToken,'old failure'); + const currentComplete=await store.complete('lease',second.claimToken,{{number:9}}); + process.stdout.write(JSON.stringify({{first,renewed,second,staleComplete,staleRelease,staleFail,currentComplete,record:records.get('lease')}})); +}})(); +""" + output = run_node(script) + + assert output["first"]["claimToken"] == "claim-a" + assert output["first"]["claimUntil"] == 130 + assert output["renewed"]["claimUntil"] == 150 + assert output["second"]["claimToken"] == "claim-b" + assert output["second"]["claimUntil"] == 181 + assert output["staleComplete"] is False + assert output["staleRelease"] is False + assert output["staleFail"] is False + assert output["currentComplete"] is True + assert output["record"]["status"] == "sent" + + def test_foreground_and_worker_race_still_posts_one_issue(): script = f""" const createBackgroundIssueSync = require({json.dumps(str(SYNC))}); @@ -978,8 +1096,8 @@ const transaction=work=>{{const run=tail.then(()=>work({{ (async()=>{{ const store=createBackgroundIssueSync.createIssueSyncStore({{transaction,now:()=>100}}); await store.reconcile([{{id:'done',ownerLogin:'timmy',status:'queued'}}]); - await store.claimNext('timmy'); - await store.complete('done'); + const claimed=await store.claimNext('timmy'); + await store.complete('done', claimed.claimToken); const replay=await store.claimNext('timmy'); const snapshot=await store.snapshot(); process.stdout.write(JSON.stringify({{replay,snapshot}}));