stackchain/stackchain-dashboard
1
0
Fork 0
Code Issues 9 Pull Requests 1 Actions Packages Projects Releases 713 Wiki Activity

Merge pull request 'Make background outbox claims lease-safe' (#480) from timmy/479-lease-safe-background-claims into main
All checks were successful
CI / lint (push) Successful in 55s
Details
CI / build-release (push) Successful in 6s
Details
CI / release-candidate (push) Successful in 5s
Details

Browse Source 
Merge pull request #480: Make background outbox claims lease-safe
This commit is contained in:
timmy 2026-08-10 11:23:39 +00:00
commit b667963968
3 changed files with 244 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@ -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

143
frontend/background-issue-sync.js
View File

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

138
tests/test_background_issue_sync.py
View File

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