fix: fence background delivery claims (Closes #479)

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

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 (lanes.issue.length) selected.push(lanes.issue.shift().id);
-        if (selected.length < maximum && lanes.authored.length) selected.push(lanes.authored.shift());
+        if (selected.length < maximum && lanes.authored.length) selected.push(lanes.authored.shift().id);
       }
-      const claimed = selected.map(item => ({
+      return selected;
-        ...item, status: 'sending', claimUntil: timestamp + claimMs,
-      }));
-      for (const item of claimed) await records.put(item);
-      return claimed;
     });
   }
 
@@ -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,
+    planBatch,
-    complete: (id, deliveredIssue) => update(id, item => ({
+    supportsClaimTokens: true,
-      ...item, status: 'sent', claimUntil: 0,
+    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 })),
+    release: (id, claimToken) => updateClaim(id, claimToken,
-    fail: (id, error, deliveryState) => update(id, item => ({
+      item => clearClaim(item, { status: 'queued' })),
-      ...item, status: 'attention', claimUntil: 0, error,
+    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);
+      deliveredIssue = await requestStage(item, request.url, request.options);
-      await store.update?.(item.id, current => ({ ...current, deliveredIssue }));
+      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(
+        await failClaim(
-          item.id,
+          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;

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);
+  const planned=await store.planBatch('timmy', 4);
-  process.stdout.write(JSON.stringify({{claimed,snapshot:await store.snapshot()}}));
+  process.stdout.write(JSON.stringify({{planned,snapshot:await store.snapshot()}}));
 }})();
 """
     output = run_node(script)
 
-    assert [item["id"] for item in output["claimed"]] == [
+    assert output["planned"] == ["issue-1", "message-1", "issue-2", "message-2"]
-        "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"])
-    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"
 
 
 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');
+  const claimed=await store.claimNext('timmy');
-  await store.complete('done');
+  await store.complete('done', claimed.claimToken);
   const replay=await store.claimNext('timmy');
   const snapshot=await store.snapshot();
   process.stdout.write(JSON.stringify({{replay,snapshot}}));