Merge pull request 'Recover queued delivery when background requests stall' (#306) from timmy/305-background-request-deadlines into main

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

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?.();
   }

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;
   }

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))});

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