From add45365ab6e4521e25c808324f2a660e9f89775 Mon Sep 17 00:00:00 2001 From: timmy Date: Sat, 8 Aug 2026 13:46:46 +0000 Subject: [PATCH] fix: recover stalled background delivery (#305) --- README.md | 7 ++- frontend/background-issue-sync.js | 40 +++++++++++++-- frontend/service-worker.js | 24 +++++++-- tests/test_background_issue_sync.py | 77 +++++++++++++++++++++++++++++ tests/test_service_worker.py | 41 ++++++++++++--- 5 files changed, 172 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 6227e40..51e3a7e 100644 --- a/README.md +++ b/README.md @@ -32,8 +32,11 @@ and durable delivery flow. A different or unconfirmed account can only copy or d the private content. Issue capture and authored mobile actions (issue comments, pull-request comments, notification replies, and reviews) persist per-draft idempotency keys, so retrying after a timeout, reload, process restart, or handoff to -another worker replays a confirmed result instead of posting duplicate content. Results -are coordinated through a bounded SQLite ledger. Set `STACKCHAIN_STATE_DIR` to a +another worker replays a confirmed result instead of posting duplicate content. 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 +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 writes run outside the request event loop, and lock admission is bounded to 100 ms by diff --git a/frontend/background-issue-sync.js b/frontend/background-issue-sync.js index 50ea7fe..66509c5 100644 --- a/frontend/background-issue-sync.js +++ b/frontend/background-issue-sync.js @@ -196,10 +196,35 @@ function createIssueSyncStore({ transaction, indexedDB = globalThis.indexedDB, n function createBackgroundIssueSync({ store, fetchJson, base = '', maxConcurrency = 3, batchSize = 70, - batch = work => work(), + batch = work => work(), requestTimeoutMs = 15000, }) { let purgeRequested = false; let activeFlush = null; + const activeRequests = new Set(); + const timeoutMs = Math.max(1, Number(requestTimeoutMs) || 15000); + + async function requestJson(url, options = {}) { + const controller = new AbortController(); + activeRequests.add(controller); + let timer; + const interrupted = new Promise((resolve, reject) => { + controller.signal.addEventListener('abort', () => { + reject(new Error(purgeRequested + ? 'Background request canceled.' + : 'Background request timed out.')); + }, { once: true }); + timer = setTimeout(() => controller.abort(), timeoutMs); + }); + try { + return await Promise.race([ + Promise.resolve().then(() => fetchJson(url, { ...options, signal: controller.signal })), + interrupted, + ]); + } finally { + clearTimeout(timer); + activeRequests.delete(controller); + } + } function receiptFor(item, status, delivered = {}) { if (status === 'attention') { return { id: item.id, status, kind: item.kind ? 'message' : 'issue', route: '#/my-work/drafts' }; @@ -265,7 +290,7 @@ function createBackgroundIssueSync({ async function deliver(item) { const request = deliveryRequest(item); try { - const delivered = await fetchJson(request.url, request.options); + const delivered = await requestJson(request.url, request.options); await store.complete(item.id); const receipt = receiptFor(item, 'confirmed', delivered); return item.kind ? { message: delivered, receipt } : { issue: delivered, receipt }; @@ -294,7 +319,7 @@ function createBackgroundIssueSync({ } async function runFlush() { - const identity = await fetchJson(base + 'api/v1/background-identity', { + const identity = await requestJson(base + 'api/v1/background-identity', { headers: { Accept: 'application/json' }, cache: 'no-store', }); const login = String(identity?.login || '').trim(); @@ -314,7 +339,7 @@ function createBackgroundIssueSync({ let authenticationError = null; let transientError = null; const worker = async () => { - while (!authenticationError && next < claimed.length) { + while (!purgeRequested && !authenticationError && next < claimed.length) { const item = claimed[next++]; try { collect(await deliver(item)); @@ -326,13 +351,17 @@ function createBackgroundIssueSync({ }; const concurrency = Math.max(1, Math.min(Number(maxConcurrency) || 1, claimed.length)); await Promise.all(Array.from({ length: concurrency }, worker)); + if (purgeRequested) { + await Promise.all(claimed.slice(next).map(item => store.release(item.id))); + throw new Error('Background delivery canceled.'); + } if (authenticationError) { await Promise.all(claimed.slice(next).map(item => store.release(item.id))); throw authenticationError; } if (transientError) throw transientError; } else { - while (true) { + while (!purgeRequested) { const item = await store.claimNext(login); if (!item) break; collect(await deliver(item)); @@ -351,6 +380,7 @@ function createBackgroundIssueSync({ async function purge() { purgeRequested = true; + activeRequests.forEach(controller => controller.abort()); if (activeFlush) await activeFlush.catch(() => {}); await store.close?.(); } diff --git a/frontend/service-worker.js b/frontend/service-worker.js index b9bd972..1a0a299 100644 --- a/frontend/service-worker.js +++ b/frontend/service-worker.js @@ -1,6 +1,6 @@ const BASE = new URL('./', self.location.href).pathname; importScripts(BASE + 'static/background-issue-sync.js'); -const CACHE = 'stackchain-dashboard-shell-v29'; +const CACHE = 'stackchain-dashboard-shell-v30'; const OUTAGE_STATUSES = new Set([500, 502, 503, 504]); const SHELL = [ BASE, @@ -38,9 +38,10 @@ const SHELL = [ BASE + 'static/background-issue-sync.js', ]; -async function sessionCsrf() { +async function sessionCsrf(signal) { const response = await fetch(new URL(BASE + 'api/v1/session', self.location.origin), { headers: { Accept: 'application/json' }, + signal, }); if (!response.ok) return ''; const payload = await response.json().catch(() => ({})); @@ -48,12 +49,16 @@ async function sessionCsrf() { } let batchedCsrf = null; +let batchedCsrfController = null; async function withSessionCsrf(work) { - batchedCsrf = sessionCsrf(); + batchedCsrfController = new AbortController(); + batchedCsrf = sessionCsrf(batchedCsrfController.signal); try { return await work(); } finally { + batchedCsrfController.abort(); batchedCsrf = null; + batchedCsrfController = null; } } @@ -62,7 +67,18 @@ async function fetchJson(url, options = {}) { const method = String(options.method || 'GET').toUpperCase(); if (!['GET', 'HEAD', 'OPTIONS'].includes(method)) { const headers = new Headers(options.headers || {}); - const csrf = await (batchedCsrf || sessionCsrf()); + const cancelBatchedCsrf = () => batchedCsrfController?.abort(); + if (batchedCsrf && options.signal) { + if (options.signal.aborted) cancelBatchedCsrf(); + else options.signal.addEventListener('abort', cancelBatchedCsrf, { once: true }); + } + let csrf; + try { + csrf = await (batchedCsrf || sessionCsrf(options.signal)); + } finally { + options.signal?.removeEventListener('abort', cancelBatchedCsrf); + } + if (options.signal?.aborted) throw new Error('Background request aborted.'); if (csrf) headers.set('X-CSRF-Token', csrf); requestOptions.headers = headers; } diff --git a/tests/test_background_issue_sync.py b/tests/test_background_issue_sync.py index 31b0fdc..5fb3e16 100644 --- a/tests/test_background_issue_sync.py +++ b/tests/test_background_issue_sync.py @@ -359,6 +359,83 @@ const fetchJson = async url => {{ assert output["error"] == "Gitea unavailable" +def test_stalled_delivery_times_out_releases_claim_and_retries_with_same_idempotency_key(): + script = f""" +const createBackgroundIssueSync = require({json.dumps(str(SYNC))}); +const item = {{id:'capture-timeout',operationId:'stable-operation',ownerLogin:'timmy',repository:'o/r',title:'Retry',body:'Later',labelIds:[]}}; +const state = {{queued:true,released:[],completed:[],keys:[],attempts:0}}; +const store = {{ + claimNext: async () => state.queued ? (state.queued=false, {{...item}}) : null, + complete: async id => state.completed.push(id), + release: async id => {{state.released.push(id); state.queued=true;}}, + fail: async () => {{}}, countBlocked: async () => 0, +}}; +const fetchJson = async (url, options={{}}) => {{ + if (url === 'api/v1/background-identity') return {{login:'timmy'}}; + state.keys.push(options.headers['Idempotency-Key']); + state.attempts += 1; + if (state.attempts === 1) return new Promise(() => {{}}); + return {{repository:'o/r',number:9}}; +}}; +(async () => {{ + const sync=createBackgroundIssueSync({{store,fetchJson,requestTimeoutMs:10}}); + let firstError=''; + try {{ await sync.flush(); }} catch (error) {{ firstError=error.message; }} + const second=await sync.flush(); + process.stdout.write(JSON.stringify({{state,firstError,second}})); +}})(); +""" + output = run_node(script) + + assert output["firstError"] == "Background request timed out." + assert output["state"]["released"] == ["capture-timeout"] + assert output["state"]["completed"] == ["capture-timeout"] + assert output["state"]["keys"] == ["stable-operation", "stable-operation"] + assert output["second"]["confirmed"][0]["number"] == 9 + + +def test_purge_cancels_stalled_delivery_without_reclaiming_it(): + script = f""" +const createBackgroundIssueSync = require({json.dumps(str(SYNC))}); +const items=[ + {{id:'capture-purge-1',operationId:'capture-purge-1',ownerLogin:'timmy',repository:'o/r',title:'Private',body:'Draft',labelIds:[]}}, + {{id:'capture-purge-2',operationId:'capture-purge-2',ownerLogin:'timmy',repository:'o/r',title:'Private 2',body:'Draft',labelIds:[]}}, +]; +const state={{released:[],closed:0,mutations:0}}; +let mutationStarted; +const started=new Promise(resolve => mutationStarted=resolve); +const store={{ + claimBatch:async()=>items.map(item=>({{...item}})), + complete:async()=>{{}}, + release:async id=>state.released.push(id), + fail:async()=>{{}}, countBlocked:async()=>0, close:async()=>{{state.closed+=1;}}, +}}; +const fetchJson=async url=>{{ + if(url==='api/v1/background-identity')return{{login:'timmy'}}; + state.mutations+=1;mutationStarted();return new Promise(()=>{{}}); +}}; +(async()=>{{ + const sync=createBackgroundIssueSync({{store,fetchJson,maxConcurrency:1,requestTimeoutMs:1000}}); + sync.flush().catch(()=>{{}}); + await started; + const outcome=await Promise.race([ + sync.purge().then(()=> 'purged'), + new Promise(resolve=>setTimeout(()=>resolve('blocked'),80)), + ]); + process.stdout.write(JSON.stringify({{state,outcome}})); + process.exit(0); +}})(); +""" + output = run_node(script) + + assert output["outcome"] == "purged" + assert output["state"] == { + "released": ["capture-purge-1", "capture-purge-2"], + "closed": 1, + "mutations": 1, + } + + def test_bounded_flush_delivers_healthy_records_after_transient_failure_without_same_run_retry(): script = f""" const createBackgroundIssueSync = require({json.dumps(str(SYNC))}); diff --git a/tests/test_service_worker.py b/tests/test_service_worker.py index 1bee7c1..9863307 100644 --- a/tests/test_service_worker.py +++ b/tests/test_service_worker.py @@ -20,7 +20,7 @@ const cache = {{ put: async (request, response) => {{ state.puts.push(String(request.url || request)); }}, }}; const context = {{ - URL, Request, Response, + URL, Request, Response, Headers, AbortController, console, self: {{ location: {{ href: 'https://forge.example/dashboard/service-worker.js', origin: 'https://forge.example' }}, @@ -52,7 +52,7 @@ const context = {{ }}, }}; vm.createContext(context); -vm.runInContext(fs.readFileSync({json.dumps(str(WORKER))}, 'utf8'), context); +vm.runInContext(fs.readFileSync({json.dumps(str(WORKER))}, 'utf8') + '\\nself.__testFetchJson = fetchJson; self.__testWithSessionCsrf = withSessionCsrf;', context); async function dispatch(name, request) {{ let pending; let response; @@ -95,7 +95,7 @@ async function dispatchNotificationClick(route) {{ def test_strict_browser_assets_ship_in_a_new_shell_cache(): source = WORKER.read_text() - assert "stackchain-dashboard-shell-v29" in source + assert "stackchain-dashboard-shell-v30" in source assert "BASE + 'static/dashboard.css'" in source assert "BASE + 'static/dashboard.js'" in source assert "BASE + 'static/install-app.js'" in source @@ -124,6 +124,31 @@ def test_authenticated_page_message_resumes_queued_background_delivery(): assert result["backgroundFlushes"] == 1 +def test_background_mutation_abort_also_cancels_stalled_csrf_lookup(): + result = run_worker_scenario( + """ + let mutations=0; + context.fetch=(request, options={})=>{ + if(String(request).includes('/api/v1/session')){ + return new Promise((resolve,reject)=>options.signal?.addEventListener('abort',()=>reject(new Error('aborted')), {once:true})); + } + mutations+=1; + return Promise.resolve(new Response('{}',{status:200,headers:{'Content-Type':'application/json'}})); + }; + const controller=new AbortController(); + const request=context.self.__testWithSessionCsrf(() => + context.self.__testFetchJson('/dashboard/api/v1/repos/o/r/issues',{method:'POST',signal:controller.signal}) + ).then(()=> 'completed', error=>error.message); + controller.abort(); + const outcome=await Promise.race([request,new Promise(resolve=>setTimeout(()=>resolve('blocked'),40))]); + process.stdout.write(JSON.stringify({outcome,mutations})); + process.exit(0); +""" + ) + + assert result == {"outcome": "aborted", "mutations": 0} + + def test_device_purge_message_stops_worker_outbox_and_acknowledges_completion(): result = run_worker_scenario( """ @@ -180,8 +205,9 @@ def test_opted_in_background_sync_notifies_privately_and_receipt_tap_focuses_rou def test_background_mutations_obtain_session_bound_csrf_proof(): source = WORKER.read_text() - assert "async function sessionCsrf()" in source + assert "async function sessionCsrf(signal)" in source assert "BASE + 'api/v1/session'" in source + assert "signal," in source assert "headers.set('X-CSRF-Token', csrf)" in source @@ -189,8 +215,11 @@ def test_one_session_bound_csrf_proof_is_reused_for_a_background_drain(): source = WORKER.read_text() assert "async function withSessionCsrf(work)" in source - assert "batchedCsrf = sessionCsrf()" in source - assert "await (batchedCsrf || sessionCsrf())" in source + assert "batchedCsrfController = new AbortController()" in source + assert "batchedCsrf = sessionCsrf(batchedCsrfController.signal)" in source + assert "batchedCsrfController.abort()" in source + assert "await (batchedCsrf || sessionCsrf(options.signal))" in source + assert "if (options.signal?.aborted)" in source assert "batch: withSessionCsrf" in source