From c74c63d29c23c88b418d11ba51a4298ae3e549e0 Mon Sep 17 00:00:00 2001 From: timmy Date: Sat, 8 Aug 2026 05:10:11 +0000 Subject: [PATCH] feat: make background queue admission durable (#264) --- frontend/authored-outbox.js | 24 +++++++++--- frontend/index.html | 43 ++++++++++++++++------ frontend/issue-outbox.js | 42 +++++++++++++++++---- frontend/my-work.js | 11 ++++-- tests/test_authored_outbox.py | 33 ++++++++++++++++- tests/test_issue_outbox.py | 69 ++++++++++++++++++++++++++++++++++- 6 files changed, 194 insertions(+), 28 deletions(-) diff --git a/frontend/authored-outbox.js b/frontend/authored-outbox.js index a534242..f0c90da 100644 --- a/frontend/authored-outbox.js +++ b/frontend/authored-outbox.js @@ -14,17 +14,17 @@ function createAuthoredOutbox({ storage, fetchJson, coordinator, backgroundSync, } catch (_error) { return []; } } - function write(items) { + function write(items, mirror = true) { storage?.setItem(storageKey, JSON.stringify({ version: 2, items })); coordinator?.notify('authored'); - if (backgroundSync?.reconcile) { + if (mirror && backgroundSync?.reconcile) { Promise.resolve(backgroundSync.reconcile(items, 'authored')) .then(() => items.length ? backgroundSync.requestSync?.() : undefined) .catch(() => { /* Foreground reconnect remains the compatibility fallback. */ }); } } - function enqueue(message) { + function enqueue(message, mirror = true) { if (!supportedKinds.has(message?.kind)) throw new Error('This action cannot be queued.'); const ownerLogin = String(getOwnerLogin() || '').trim(); if (!ownerLogin) throw new Error('Confirm your Gitea account before queueing a message.'); @@ -47,10 +47,24 @@ function createAuthoredOutbox({ storage, fetchJson, coordinator, backgroundSync, queuedAt: Number(now()), }; items.push(item); - write(items); + write(items, mirror); return item; } + async function enqueueDurably(message) { + const item = enqueue(message, false); + if (!backgroundSync?.reconcile || !backgroundSync?.requestSync) { + return { item, background: false, durability: 'foreground-only' }; + } + try { + await backgroundSync.reconcile(read(), 'authored'); + await backgroundSync.requestSync(); + return { item, background: true, durability: 'background' }; + } catch (error) { + return { item, background: false, durability: 'foreground-only', error }; + } + } + function update(id, changes) { let updated = null; write(read().map(item => { @@ -182,7 +196,7 @@ function createAuthoredOutbox({ storage, fetchJson, coordinator, backgroundSync, return items; } - return { enqueue, update, discard, flush, retry, reconcileBackground, list: () => read().map(item => ({ ...item })) }; + return { enqueue, enqueueDurably, update, discard, flush, retry, reconcileBackground, list: () => read().map(item => ({ ...item })) }; } if (typeof module !== 'undefined' && module.exports) module.exports = createAuthoredOutbox; diff --git a/frontend/index.html b/frontend/index.html index cc1642e..26e473e 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -933,9 +933,10 @@ textarea { resize: vertical; min-height: 120px; } store: backgroundIssueStore, fetchJson: fetchReviewJson, }); backgroundIssueSync.requestSync = async () => { - if (!('serviceWorker' in navigator)) return; + if (!('serviceWorker' in navigator)) throw new Error('Background Sync unavailable'); const registration = await navigator.serviceWorker.ready; - if (registration.sync) await registration.sync.register('stackchain-issue-outbox-v1'); + if (!registration.sync) throw new Error('Background Sync unavailable'); + await registration.sync.register('stackchain-issue-outbox-v1'); }; } const outboxCoordinator = createOutboxCoordinator({ storage: localStorage }); @@ -2819,10 +2820,18 @@ textarea { resize: vertical; min-height: 120px; } } const button = qs('#submit-new-issue'); button.disabled = true; - qs('#create-issue-status').textContent = navigator.onLine ? 'Sending issue…' : 'Saving issue to outbox…'; + qs('#create-issue-status').textContent = 'Saving for background delivery…'; try { - const queued = editingOutboxId ? issueOutbox.update(editingOutboxId, captureDraft) : - issueOutbox.enqueue(captureDraft); + const admission = editingOutboxId ? await issueOutbox.updateDurably(editingOutboxId, captureDraft) : + await issueOutbox.enqueueDurably(captureDraft); + const queued = admission.item; + if (!admission.background) { + editingOutboxId = queued.id; + refreshMyWorkView(); + qs('#create-issue-status').textContent = 'Saved for next launch; background delivery unavailable.'; + button.disabled = false; + return; + } editingOutboxId = null; issueCapture.clearDraft(); closeCreateIssueSheet(); @@ -3038,11 +3047,17 @@ textarea { resize: vertical; min-height: 120px; } } catch (error) { if (canQueueMessage(error)) { const operationId = localStorage.getItem('stackchain.issue-comment.v1:' + selectedIssue.repository + '#' + selectedIssue.number + ':operation'); - authoredOutbox.enqueue({ kind:'issue-comment', repository:selectedIssue.repository, + qs('#issue-comment-status').textContent = 'Saving for background delivery…'; + const admission = await authoredOutbox.enqueueDurably({ kind:'issue-comment', repository:selectedIssue.repository, number:selectedIssue.number, body, operationId }); - qs('#issue-comment').value = ''; - qs('#issue-comment-status').textContent = 'Queued for sync when the connection returns.'; refreshMyWorkView(); + if (admission.background) { + qs('#issue-comment').value = ''; + qs('#issue-comment-status').textContent = 'Queued for sync when the connection returns.'; + } else { + qs('#issue-comment-status').textContent = 'Saved for next launch; background delivery unavailable.'; + qs('#issue-comment').focus(); + } } else { qs('#issue-comment-status').textContent = error.message + ' Your draft is safe; retry.'; qs('#issue-comment').focus(); @@ -3139,11 +3154,17 @@ textarea { resize: vertical; min-height: 120px; } } catch (error) { if (canQueueMessage(error)) { const operationId = localStorage.getItem('stackchain.pull-comment.v1:' + selectedPull.repository + '#' + selectedPull.number + ':operation'); - authoredOutbox.enqueue({ kind:'pull-comment', repository:selectedPull.repository, + qs('#pull-comment-status').textContent = 'Saving for background delivery…'; + const admission = await authoredOutbox.enqueueDurably({ kind:'pull-comment', repository:selectedPull.repository, number:selectedPull.number, body, operationId }); - qs('#pull-comment').value = ''; - qs('#pull-comment-status').textContent = 'Queued for sync when the connection returns.'; refreshMyWorkView(); + if (admission.background) { + qs('#pull-comment').value = ''; + qs('#pull-comment-status').textContent = 'Queued for sync when the connection returns.'; + } else { + qs('#pull-comment-status').textContent = 'Saved for next launch; background delivery unavailable.'; + qs('#pull-comment').focus(); + } } else { qs('#pull-comment-status').textContent = error.message + ' Your draft is safe; retry.'; qs('#pull-comment').focus(); diff --git a/frontend/issue-outbox.js b/frontend/issue-outbox.js index 45a803c..b3fca33 100644 --- a/frontend/issue-outbox.js +++ b/frontend/issue-outbox.js @@ -13,17 +13,17 @@ function createIssueOutbox({ storage, fetchJson, coordinator, backgroundSync, ge } catch (_error) { return []; } } - function write(items) { + function write(items, mirror = true) { storage?.setItem(storageKey, JSON.stringify({ version: 2, items })); coordinator?.notify('issue'); - if (backgroundSync?.reconcile) { + if (mirror && backgroundSync?.reconcile) { Promise.resolve(backgroundSync.reconcile(items)) .then(() => items.length ? backgroundSync.requestSync?.() : undefined) .catch(() => { /* Foreground reconnect remains the compatibility fallback. */ }); } } - function enqueue(draft) { + function enqueue(draft, mirror = true) { const ownerLogin = String(getOwnerLogin() || '').trim(); if (!ownerLogin) throw new Error('Confirm your Gitea account before queueing an issue.'); const items = read(); @@ -45,11 +45,25 @@ function createIssueOutbox({ storage, fetchJson, coordinator, backgroundSync, ge } if (/^\d{4}-\d{2}-\d{2}$/.test(String(draft?.dueDate || ''))) item.dueDate = String(draft.dueDate); items.push(item); - write(items); + write(items, mirror); return item; } - function update(id, draft) { + async function enqueueDurably(draft) { + const item = enqueue(draft, false); + if (!backgroundSync?.reconcile || !backgroundSync?.requestSync) { + return { item, background: false, durability: 'foreground-only' }; + } + try { + await backgroundSync.reconcile(read()); + await backgroundSync.requestSync(); + return { item, background: true, durability: 'background' }; + } catch (error) { + return { item, background: false, durability: 'foreground-only', error }; + } + } + + function update(id, draft, mirror = true) { let updated = null; write(read().map(item => { if (item.id !== id) return item; @@ -62,10 +76,24 @@ function createIssueOutbox({ storage, fetchJson, coordinator, backgroundSync, ge }; delete updated.error; return updated; - })); + }), mirror); return updated; } + async function updateDurably(id, draft) { + const item = update(id, draft, false); + if (!item || !backgroundSync?.reconcile || !backgroundSync?.requestSync) { + return { item, background: false, durability: 'foreground-only' }; + } + try { + await backgroundSync.reconcile(read()); + await backgroundSync.requestSync(); + return { item, background: true, durability: 'background' }; + } catch (error) { + return { item, background: false, durability: 'foreground-only', error }; + } + } + function discard(id) { const items = read(); if (!items.some(item => item.id === id)) return false; @@ -170,7 +198,7 @@ function createIssueOutbox({ storage, fetchJson, coordinator, backgroundSync, ge return items; } - return { enqueue, update, discard, flush, retry, reconcileBackground, list: () => read().map(item => ({ ...item })) }; + return { enqueue, enqueueDurably, update, updateDurably, discard, flush, retry, reconcileBackground, list: () => read().map(item => ({ ...item })) }; } if (typeof module !== 'undefined' && module.exports) module.exports = createIssueOutbox; diff --git a/frontend/my-work.js b/frontend/my-work.js index b9891ca..a75f3cf 100644 --- a/frontend/my-work.js +++ b/frontend/my-work.js @@ -383,11 +383,16 @@ function createNotificationReplier({ } catch (error) { const status = Number(error?.status || 0); if (authoredOutbox && (!status || status >= 500)) { - authoredOutbox.enqueue({ + onStatus('Saving for background delivery…'); + const admission = await authoredOutbox.enqueueDurably({ kind: 'update-reply', notificationId: item.notification_id, body, operationId, }); - onStatus('Queued for sync when the connection returns.'); - return { queued: true }; + if (admission.background) { + onStatus('Queued for sync when the connection returns.'); + return { queued: true }; + } + onStatus('Saved for next launch; background delivery unavailable.'); + return { queued: false, degraded: true }; } onStatus('Could not send reply. Your draft is safe; retry.'); return false; diff --git a/tests/test_authored_outbox.py b/tests/test_authored_outbox.py index 65afc48..cb461b8 100644 --- a/tests/test_authored_outbox.py +++ b/tests/test_authored_outbox.py @@ -182,6 +182,35 @@ setTimeout(() => {{ ] +def test_authored_outbox_waits_for_durable_background_admission(): + script = f""" +const createAuthoredOutbox = require({json.dumps(str(OUTBOX))}); +const values = new Map(); +const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}}; +let release; const gate = new Promise(resolve => release = resolve); const events = []; +const outbox = createAuthoredOutbox({{ + storage, getOwnerLogin:()=> 'timmy', + backgroundSync: {{ + reconcile: async (_items, lane) => {{ events.push('mirror:' + lane); await gate; events.push('committed'); }}, + requestSync: async () => events.push('registered'), + }}, +}}); +const admission = outbox.enqueueDurably({{ + kind:'issue-comment',repository:'o/r',number:1,body:'Do not lose',operationId:'message-1' +}}).then(result => {{ events.push('confirmed'); return result; }}); +Promise.resolve().then(async () => {{ + const pending = events.slice(); release(); const result = await admission; + process.stdout.write(JSON.stringify({{pending,events,result,items:outbox.list()}})); +}}); +""" + output = run_node(script) + + assert output["pending"] == ["mirror:authored"] + assert output["events"] == ["mirror:authored", "committed", "registered", "confirmed"] + assert output["result"]["durability"] == "background" + assert output["items"][0]["operationId"] == "message-1" + + def test_authored_outbox_foreground_send_uses_atomic_background_delivery(): script = f""" const createAuthoredOutbox = require({json.dumps(str(OUTBOX))}); @@ -217,7 +246,9 @@ async def test_mobile_dashboard_loads_and_operates_authored_message_outbox(): assert "const authoredOutbox = createAuthoredOutbox({" in html assert "activeFlushLogin = contextIdentityFresh ?" in html assert "flushAuthoredOutbox();" in html - assert "authoredOutbox.enqueue" in html + assert "await authoredOutbox.enqueueDurably" in html + assert "Saving for background delivery…" in html + assert "Saved for next launch; background delivery unavailable." in html assert "authoredOutbox.retry(item.outbox_id, activeFlushLogin)" in html assert "authoredOutbox.discard(item.outbox_id)" in html assert "if (result?.queued)" in html diff --git a/tests/test_issue_outbox.py b/tests/test_issue_outbox.py index 9c2d1c2..9706c55 100644 --- a/tests/test_issue_outbox.py +++ b/tests/test_issue_outbox.py @@ -186,6 +186,70 @@ setTimeout(() => process.stdout.write(JSON.stringify(state)), 0); assert output["syncs"] == 1 +def test_issue_outbox_confirms_durable_admission_only_after_mirror_and_sync(): + script = f""" +const createIssueOutbox = require({json.dumps(str(OUTBOX))}); +const values = new Map(); +const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}}; +let releaseMirror; let releaseSync; +const mirrorGate = new Promise(resolve => releaseMirror = resolve); +const syncGate = new Promise(resolve => releaseSync = resolve); +const events = []; +const outbox = createIssueOutbox({{ + storage, getOwnerLogin:()=> 'timmy', createOperationId:()=> 'durable-1', + backgroundSync: {{ + reconcile: async items => {{ events.push('mirror-start'); await mirrorGate; events.push('mirror-committed'); }}, + requestSync: async () => {{ events.push('sync-start'); await syncGate; events.push('sync-registered'); }}, + }}, +}}); +const admission = outbox.enqueueDurably({{repository:'stackchain/api',title:'Keep this',body:'Draft'}}) + .then(result => {{ events.push('confirmed'); return result; }}); +Promise.resolve().then(async () => {{ + const beforeMirror = events.slice(); + releaseMirror(); await new Promise(resolve => setTimeout(resolve, 0)); + const beforeSync = events.slice(); + releaseSync(); + const result = await admission; + process.stdout.write(JSON.stringify({{beforeMirror,beforeSync,events,result,items:outbox.list()}})); +}}); +""" + output = run_node(script) + + assert output["beforeMirror"] == ["mirror-start"] + assert output["beforeSync"] == ["mirror-start", "mirror-committed", "sync-start"] + assert output["events"][-1] == "confirmed" + assert output["result"]["background"] is True + assert output["result"]["item"]["id"] == "durable-1" + assert output["items"][0]["title"] == "Keep this" + + +def test_issue_outbox_reports_degraded_admission_without_losing_foreground_item(): + script = f""" +const createIssueOutbox = require({json.dumps(str(OUTBOX))}); +const values = new Map(); +const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v)}}; +const outbox = createIssueOutbox({{ + storage, getOwnerLogin:()=> 'timmy', createOperationId:()=> 'foreground-1', + backgroundSync: {{ + reconcile: async () => {{}}, + requestSync: async () => {{ throw new Error('Background Sync unavailable'); }}, + }}, +}}); +outbox.enqueueDurably({{repository:'stackchain/api',title:'Retain me',body:'Draft'}}).then(result => + process.stdout.write(JSON.stringify({{ + background:result.background, durability:result.durability, + error:result.error.message, item:result.item, persisted:outbox.list() + }})) +); +""" + output = run_node(script) + + assert output["background"] is False + assert output["durability"] == "foreground-only" + assert output["error"] == "Background Sync unavailable" + assert output["persisted"] == [output["item"]] + + def test_foreground_delivery_uses_same_atomic_background_claim(): script = f""" const createIssueOutbox = require({json.dumps(str(OUTBOX))}); @@ -263,7 +327,10 @@ async def test_mobile_dashboard_queues_offline_captures_and_exposes_outbox_actio assert '' in html assert "const issueOutbox = createIssueOutbox({" in html - assert "issueOutbox.enqueue(captureDraft)" in html + assert "await issueOutbox.enqueueDurably(captureDraft)" in html + assert "Saving for background delivery…" in html + assert "Saved for next launch; background delivery unavailable." in html + assert "throw new Error('Background Sync unavailable')" in html assert "issueOutbox.retry(queued.id, activeFlushLogin)" in html assert "navigator.onLine" in html assert "activeFlushLogin = contextIdentityFresh ?" in html