Merge pull request 'Coordinate offline outboxes safely across dashboard tabs' (#243) from timmy/242-cross-tab-outbox-coordination into main

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:

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

frontend/index.html

@@ -739,6 +739,7 @@ textarea { resize: vertical; min-height: 120px; }
 <script src="static/search-preview.js"></script>
 <script src="static/widgets.js"></script>
 <script src="static/drafts.js"></script>
+<script src="static/outbox-coordinator.js"></script>
 <script src="static/issue-outbox.js"></script>
 <script src="static/authored-outbox.js"></script>
 <script src="static/offline-work.js"></script>
@@ -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 ?

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

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;

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',

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 '<script src="static/outbox-coordinator.js"></script>' 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

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",