diff --git a/README.md b/README.md index 45d1b7d..590e2d2 100644 --- a/README.md +++ b/README.md @@ -97,9 +97,13 @@ API responses and mutations are never cached by the service worker. New issue ca issue comments, pull-request comments, and unread-update replies use bounded local outboxes when connectivity or a retryable server failure prevents delivery. Drafts shows queued and needs-attention messages with explicit send/discard controls; reconnect -flushes messages sequentially with their original idempotency keys. State-sensitive -actions such as reviews, merges, closures, labels, milestones, and assignments are never -queued. Service-worker upgrades are atomic and remove only older +flushes messages sequentially with their original idempotency keys. Tabs coordinate each +queue through the browser lock manager (with an expiring local lease fallback), so an +installed PWA and a browser tab cannot submit the same operation concurrently. Queue +changes are broadcast to other tabs and refresh Drafts without polling; an abandoned +lease can be reclaimed after expiry. State-sensitive actions such as reviews, merges, +closures, labels, milestones, and assignments are never queued. Service-worker upgrades +are atomic and remove only older `stackchain-dashboard-*` caches, preserving unrelated caches on the same origin. Run the test suite with: diff --git a/frontend/authored-outbox.js b/frontend/authored-outbox.js index 1b3d57d..bc45413 100644 --- a/frontend/authored-outbox.js +++ b/frontend/authored-outbox.js @@ -1,4 +1,4 @@ -function createAuthoredOutbox({ storage, fetchJson, getOwnerLogin = () => '', createOperationId, now = () => Date.now(), maxItems = 50 }) { +function createAuthoredOutbox({ storage, fetchJson, coordinator, getOwnerLogin = () => '', createOperationId, now = () => Date.now(), maxItems = 50 }) { const storageKey = 'stackchain.authored-outbox.v1'; const makeId = createOperationId || (() => globalThis.crypto?.randomUUID?.() || String(Date.now()) + '-' + Math.random().toString(16).slice(2) @@ -16,6 +16,7 @@ function createAuthoredOutbox({ storage, fetchJson, getOwnerLogin = () => '', cr function write(items) { storage?.setItem(storageKey, JSON.stringify({ version: 2, items })); + coordinator?.notify('authored'); } function enqueue(message) { @@ -112,7 +113,7 @@ function createAuthoredOutbox({ storage, fetchJson, getOwnerLogin = () => '', cr finally { if (pending.get(item.id) === request) pending.delete(item.id); } } - async function flush(currentLogin) { + async function flushQueue(currentLogin) { const confirmed = []; let blocked = 0; currentLogin = String(currentLogin || '').trim(); @@ -126,7 +127,12 @@ function createAuthoredOutbox({ storage, fetchJson, getOwnerLogin = () => '', cr return { confirmed, remaining: read(), blocked }; } - async function retry(id, currentLogin) { + async function flush(currentLogin) { + if (coordinator) return coordinator.runExclusive('authored', () => flushQueue(currentLogin)); + return flushQueue(currentLogin); + } + + async function retryItem(id, currentLogin) { const item = read().find(candidate => candidate.id === id); if (!item) return { confirmed: [], remaining: read() }; currentLogin = String(currentLogin || '').trim(); @@ -138,6 +144,11 @@ function createAuthoredOutbox({ storage, fetchJson, getOwnerLogin = () => '', cr return { confirmed: outcome.result ? [outcome.result] : [], remaining: read(), blocked: 0 }; } + async function retry(id, currentLogin) { + if (coordinator) return coordinator.runExclusive('authored', () => retryItem(id, currentLogin)); + return retryItem(id, currentLogin); + } + return { enqueue, update, discard, flush, retry, list: () => read().map(item => ({ ...item })) }; } diff --git a/frontend/index.html b/frontend/index.html index 1c25f49..02f798f 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -739,6 +739,7 @@ textarea { resize: vertical; min-height: 120px; } + @@ -884,11 +885,14 @@ textarea { resize: vertical; min-height: 120px; } loadMilestones: item => issueController.loadMilestones(item), }); const issueCapture = createIssueCapture({ fetchJson: fetchReviewJson, storage: localStorage }); + const outboxCoordinator = createOutboxCoordinator({ storage: localStorage }); const issueOutbox = createIssueOutbox({ - storage: localStorage, fetchJson: fetchReviewJson, getOwnerLogin: () => confirmedOwnerLogin, + storage: localStorage, fetchJson: fetchReviewJson, coordinator: outboxCoordinator, + getOwnerLogin: () => confirmedOwnerLogin, }); const authoredOutbox = createAuthoredOutbox({ - storage: localStorage, fetchJson: fetchReviewJson, getOwnerLogin: () => confirmedOwnerLogin, + storage: localStorage, fetchJson: fetchReviewJson, coordinator: outboxCoordinator, + getOwnerLogin: () => confirmedOwnerLogin, }); const shareParams = new URLSearchParams(location.search); const sharedLaunch = { @@ -900,6 +904,7 @@ textarea { resize: vertical; min-height: 120px; } issueCapture.stageSharedContent(sharedLaunch) : null; const pullController = createPullSheet({ fetchJson: fetchReviewJson, storage: localStorage }); const draftInbox = createDraftInbox({ storage: localStorage, getCurrentLogin: () => activeFlushLogin }); + outboxCoordinator.subscribe(() => refreshMyWorkView()); const offlineWorkStore = createOfflineWorkStore({ storage: localStorage }); const findWorkController = createFindWork({ fetchJson: fetchReviewJson, @@ -2027,6 +2032,7 @@ textarea { resize: vertical; min-height: 120px; } } function applyOutboxResult(result, openCreated = false) { + if (result.lease_skipped) { refreshMyWorkView(); return; } (result.confirmed || []).forEach(confirmed => { if (lastContextSnapshot) lastContextSnapshot.issues = [confirmed].concat(lastContextSnapshot.issues || []); }); @@ -2053,6 +2059,7 @@ textarea { resize: vertical; min-height: 120px; } } function applyAuthoredOutboxResult(result) { + if (result.lease_skipped) { refreshMyWorkView(); return; } refreshMyWorkView(); const attention = (result.remaining || []).some(item => item.status === 'attention'); qs('#my-work-action-status').textContent = result.confirmed?.length ? diff --git a/frontend/issue-outbox.js b/frontend/issue-outbox.js index 44be1ff..3ce7483 100644 --- a/frontend/issue-outbox.js +++ b/frontend/issue-outbox.js @@ -1,4 +1,4 @@ -function createIssueOutbox({ storage, fetchJson, getOwnerLogin = () => '', createOperationId, now = () => Date.now(), maxItems = 20 }) { +function createIssueOutbox({ storage, fetchJson, coordinator, getOwnerLogin = () => '', createOperationId, now = () => Date.now(), maxItems = 20 }) { const storageKey = 'stackchain.issue-outbox.v1'; const operationId = createOperationId || (() => globalThis.crypto?.randomUUID?.() || String(Date.now()) + '-' + Math.random().toString(16).slice(2) @@ -15,6 +15,7 @@ function createIssueOutbox({ storage, fetchJson, getOwnerLogin = () => '', creat function write(items) { storage?.setItem(storageKey, JSON.stringify({ version: 2, items })); + coordinator?.notify('issue'); } function enqueue(draft) { @@ -102,7 +103,7 @@ function createIssueOutbox({ storage, fetchJson, getOwnerLogin = () => '', creat finally { if (pending.get(item.id) === request) pending.delete(item.id); } } - async function flush(currentLogin) { + async function flushQueue(currentLogin) { const confirmed = []; let blocked = 0; currentLogin = String(currentLogin || '').trim(); @@ -116,7 +117,12 @@ function createIssueOutbox({ storage, fetchJson, getOwnerLogin = () => '', creat return { confirmed, remaining: read(), blocked }; } - async function retry(id, currentLogin) { + async function flush(currentLogin) { + if (coordinator) return coordinator.runExclusive('issue', () => flushQueue(currentLogin)); + return flushQueue(currentLogin); + } + + async function retryItem(id, currentLogin) { const item = read().find(candidate => candidate.id === id); if (!item) return { confirmed: [], remaining: read() }; currentLogin = String(currentLogin || '').trim(); @@ -128,6 +134,11 @@ function createIssueOutbox({ storage, fetchJson, getOwnerLogin = () => '', creat return { confirmed: result.issue ? [result.issue] : [], remaining: read(), blocked: 0 }; } + async function retry(id, currentLogin) { + if (coordinator) return coordinator.runExclusive('issue', () => retryItem(id, currentLogin)); + return retryItem(id, currentLogin); + } + return { enqueue, update, discard, flush, retry, list: () => read().map(item => ({ ...item })) }; } diff --git a/frontend/outbox-coordinator.js b/frontend/outbox-coordinator.js new file mode 100644 index 0000000..557de5c --- /dev/null +++ b/frontend/outbox-coordinator.js @@ -0,0 +1,98 @@ +function createOutboxCoordinator({ + storage, + locks = globalThis.navigator?.locks, + channelFactory = typeof globalThis.BroadcastChannel === 'function' ? + () => new globalThis.BroadcastChannel('stackchain-outbox-v1') : null, + addStorageListener = globalThis.addEventListener?.bind(globalThis), + removeStorageListener = globalThis.removeEventListener?.bind(globalThis), + tabId = globalThis.crypto?.randomUUID?.() || String(Date.now()) + '-' + Math.random().toString(16).slice(2), + now = () => Date.now(), + leaseMs = 15000, +} = {}) { + const changeKey = 'stackchain.outbox-change.v1'; + const subscribers = new Set(); + const channel = channelFactory ? channelFactory() : null; + let sequence = 0; + + function validChange(value) { + return value && (value.queue === 'issue' || value.queue === 'authored') ? value : null; + } + + function publish(change) { + subscribers.forEach(listener => listener(change)); + } + + if (channel) { + channel.onmessage = event => { + const change = validChange(event?.data); + if (change && change.tabId !== tabId) publish(change); + }; + } + + const onStorage = event => { + if (event?.key !== changeKey || !event.newValue) return; + try { + const change = validChange(JSON.parse(event.newValue)); + if (change && change.tabId !== tabId) publish(change); + } catch (_error) { /* Ignore malformed cross-tab signals. */ } + }; + if (addStorageListener) addStorageListener('storage', onStorage); + + function notify(queue) { + const change = { queue, tabId, changedAt: Number(now()), sequence: ++sequence }; + if (!validChange(change)) return; + try { storage?.setItem(changeKey, JSON.stringify(change)); } catch (_error) { /* Broadcast may still work. */ } + try { channel?.postMessage(change); } catch (_error) { /* Storage events remain available. */ } + } + + function skipped() { + return { confirmed: [], remaining: [], blocked: 0, lease_skipped: true }; + } + + async function withFallbackLease(queue, work) { + const key = 'stackchain.outbox-lease.' + queue + '.v1'; + const token = tabId + ':' + (++sequence); + const timestamp = Number(now()); + try { + const current = JSON.parse(storage?.getItem(key) || 'null'); + if (current?.token && Number(current.expiresAt) > timestamp) return skipped(); + storage?.setItem(key, JSON.stringify({ token, expiresAt: timestamp + leaseMs })); + const claimed = JSON.parse(storage?.getItem(key) || 'null'); + if (claimed?.token !== token) return skipped(); + } catch (_error) { + return work(); + } + try { + return await work(); + } finally { + try { + const current = JSON.parse(storage?.getItem(key) || 'null'); + if (current?.token === token) storage?.removeItem(key); + } catch (_error) { /* An expired lease will be reclaimed. */ } + } + } + + async function runExclusive(queue, work) { + if (locks?.request) { + return locks.request('stackchain-outbox-' + queue + '-v1', { mode: 'exclusive', ifAvailable: true }, + lock => lock ? work() : skipped()); + } + return withFallbackLease(queue, work); + } + + function subscribe(listener) { + if (typeof listener !== 'function') return () => {}; + subscribers.add(listener); + return () => subscribers.delete(listener); + } + + function close() { + subscribers.clear(); + try { channel?.close(); } catch (_error) { /* No-op. */ } + if (removeStorageListener) removeStorageListener('storage', onStorage); + } + + return { runExclusive, notify, subscribe, close }; +} + +if (typeof module !== 'undefined' && module.exports) module.exports = createOutboxCoordinator; diff --git a/frontend/service-worker.js b/frontend/service-worker.js index 32bf00d..97deee1 100644 --- a/frontend/service-worker.js +++ b/frontend/service-worker.js @@ -1,4 +1,4 @@ -const CACHE = 'stackchain-dashboard-shell-v9'; +const CACHE = 'stackchain-dashboard-shell-v10'; const OUTAGE_STATUSES = new Set([500, 502, 503, 504]); const BASE = new URL('./', self.location.href).pathname; const SHELL = [ @@ -11,6 +11,7 @@ const SHELL = [ BASE + 'static/search-preview.js', BASE + 'static/widgets.js', BASE + 'static/drafts.js', + BASE + 'static/outbox-coordinator.js', BASE + 'static/issue-outbox.js', BASE + 'static/authored-outbox.js', BASE + 'static/offline-work.js', diff --git a/tests/test_outbox_coordinator.py b/tests/test_outbox_coordinator.py new file mode 100644 index 0000000..40610f3 --- /dev/null +++ b/tests/test_outbox_coordinator.py @@ -0,0 +1,134 @@ +import json +import subprocess +from pathlib import Path + +import pytest + +from src.views import dashboard + + +ROOT = Path(__file__).parents[1] +COORDINATOR = ROOT / "frontend" / "outbox-coordinator.js" +ISSUE_OUTBOX = ROOT / "frontend" / "issue-outbox.js" +AUTHORED_OUTBOX = ROOT / "frontend" / "authored-outbox.js" + + +def run_node(script: str): + result = subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True) + return json.loads(result.stdout) + + +def test_two_tabs_share_one_flush_lease_for_each_outbox(): + script = f""" +const createCoordinator = require({json.dumps(str(COORDINATOR))}); +const createIssueOutbox = require({json.dumps(str(ISSUE_OUTBOX))}); +const createAuthoredOutbox = require({json.dumps(str(AUTHORED_OUTBOX))}); +const values = new Map(); +const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}}; +const held = new Set(); +const locks = {{request: async (name, options, work) => {{ + if (held.has(name)) return work(null); + held.add(name); + try {{ return await work({{name}}); }} finally {{ held.delete(name); }} +}}}}; +let releaseIssue, releaseMessage; +const issueGate = new Promise(resolve => releaseIssue = resolve); +const messageGate = new Promise(resolve => releaseMessage = resolve); +let issueCalls = 0, messageCalls = 0; +const issueA = createIssueOutbox({{storage,getOwnerLogin:()=> 'timmy',createOperationId:()=> 'issue-1', + coordinator:createCoordinator({{storage,locks,channelFactory:null,tabId:'a'}}),fetchJson:async()=>{{issueCalls++;await issueGate;return {{number:1}};}}}}); +const issueB = createIssueOutbox({{storage,getOwnerLogin:()=> 'timmy', + coordinator:createCoordinator({{storage,locks,channelFactory:null,tabId:'b'}}),fetchJson:async()=>{{issueCalls++;return {{number:1}};}}}}); +issueA.enqueue({{repository:'o/r',title:'One'}}); +const messageA = createAuthoredOutbox({{storage,getOwnerLogin:()=> 'timmy', + coordinator:createCoordinator({{storage,locks,channelFactory:null,tabId:'a'}}),fetchJson:async()=>{{messageCalls++;await messageGate;return {{id:1}};}}}}); +const messageB = createAuthoredOutbox({{storage,getOwnerLogin:()=> 'timmy', + coordinator:createCoordinator({{storage,locks,channelFactory:null,tabId:'b'}}),fetchJson:async()=>{{messageCalls++;return {{id:1}};}}}}); +messageA.enqueue({{kind:'issue-comment',repository:'o/r',number:1,body:'Reply',operationId:'message-1'}}); +(async()=>{{ + const issueFirst=issueA.flush('timmy'); const issueSecond=issueB.retry('issue-1', 'timmy'); + const messageFirst=messageA.flush('timmy'); const messageSecond=messageB.retry('message-1', 'timmy'); + await Promise.resolve(); + releaseIssue(); releaseMessage(); + const results=await Promise.all([issueFirst,issueSecond,messageFirst,messageSecond]); + process.stdout.write(JSON.stringify({{issueCalls,messageCalls,results,issueRemaining:issueA.list(),messageRemaining:messageA.list()}})); +}})(); +""" + output = run_node(script) + + assert output["issueCalls"] == 1 + assert output["messageCalls"] == 1 + assert output["issueRemaining"] == [] + assert output["messageRemaining"] == [] + assert sum(result.get("lease_skipped", False) for result in output["results"]) == 2 + + +def test_fallback_lease_expires_and_stale_owner_cannot_release_successor(): + script = f""" +const createCoordinator = require({json.dumps(str(COORDINATOR))}); +const values = new Map(); +const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}}; +let clock=100; +let releaseFirst; +const firstGate=new Promise(resolve=>releaseFirst=resolve); +const first=createCoordinator({{storage,locks:null,channelFactory:null,tabId:'first',now:()=>clock,leaseMs:50}}); +const second=createCoordinator({{storage,locks:null,channelFactory:null,tabId:'second',now:()=>clock,leaseMs:50}}); +(async()=>{{ + const running=first.runExclusive('issue', async()=>{{await firstGate;return 'first-done';}}); + await Promise.resolve(); + const blocked=await second.runExclusive('issue', async()=> 'too-early'); + clock=151; + const takeover=await second.runExclusive('issue', async()=> 'taken-over'); + releaseFirst(); + const original=await running; + const after=await second.runExclusive('issue', async()=> 'after-release'); + process.stdout.write(JSON.stringify({{blocked,takeover,original,after,lease:storage.getItem('stackchain.outbox-lease.issue.v1')}})); +}})(); +""" + output = run_node(script) + + assert output["blocked"]["lease_skipped"] is True + assert output["takeover"] == "taken-over" + assert output["original"] == "first-done" + assert output["after"] == "after-release" + assert output["lease"] is None + + +def test_queue_change_notifications_cross_tabs_and_can_unsubscribe(): + script = f""" +const createCoordinator = require({json.dumps(str(COORDINATOR))}); +const values = new Map(); +const storage = {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}}; +const channels=[]; +function channelFactory() {{ + const channel={{onmessage:null,postMessage(message){{channels.filter(x=>x!==channel).forEach(x=>x.onmessage?.({{data:message}}));}},close(){{}}}}; + channels.push(channel); return channel; +}} +const first=createCoordinator({{storage,channelFactory,tabId:'first'}}); +const second=createCoordinator({{storage,channelFactory,tabId:'second'}}); +const seen=[]; +const unsubscribe=second.subscribe(change=>seen.push(change)); +first.notify('issue'); +first.notify('authored'); +unsubscribe(); +first.notify('issue'); +process.stdout.write(JSON.stringify({{seen,signal:JSON.parse(storage.getItem('stackchain.outbox-change.v1'))}})); +""" + output = run_node(script) + + assert [change["queue"] for change in output["seen"]] == ["issue", "authored"] + assert output["signal"]["queue"] == "issue" + assert output["signal"]["tabId"] == "first" + + +@pytest.mark.anyio +async def test_dashboard_wires_cross_tab_draft_refresh_and_precaches_coordinator(): + html = await dashboard() + worker = (ROOT / "frontend" / "service-worker.js").read_text() + + assert '' in html + assert "const outboxCoordinator = createOutboxCoordinator" in html + assert "coordinator: outboxCoordinator" in html + assert "outboxCoordinator.subscribe" in html + assert "refreshMyWorkView()" in html + assert "BASE + 'static/outbox-coordinator.js'" in worker diff --git a/tests/test_service_worker.py b/tests/test_service_worker.py index c29b6f4..9e370b9 100644 --- a/tests/test_service_worker.py +++ b/tests/test_service_worker.py @@ -63,10 +63,10 @@ async function dispatch(name, request) {{ return json.loads(completed.stdout) -def test_review_to_merge_flow_ships_in_a_new_shell_cache(): +def test_cross_tab_outbox_coordination_ships_in_a_new_shell_cache(): source = WORKER.read_text() - assert "stackchain-dashboard-shell-v9" in source + assert "stackchain-dashboard-shell-v10" in source def test_install_precaches_complete_subpath_scoped_app_shell(): @@ -89,6 +89,7 @@ def test_install_precaches_complete_subpath_scoped_app_shell(): "/dashboard/static/search-preview.js", "/dashboard/static/widgets.js", "/dashboard/static/drafts.js", + "/dashboard/static/outbox-coordinator.js", "/dashboard/static/issue-outbox.js", "/dashboard/static/authored-outbox.js", "/dashboard/static/offline-work.js",