From 25dda21811c1f3dec603814f998a8785525cd756 Mon Sep 17 00:00:00 2001 From: timmy Date: Mon, 17 Aug 2026 16:00:59 +0000 Subject: [PATCH] fix: converge Today session conflicts across devices (Closes #1030) --- frontend/index.html | 1 + frontend/today-session-sync.js | 98 ++++++++++--- tests/test_today_session_handoff_ui.py | 4 + tests/test_today_session_sync.py | 196 +++++++++++++++++++++++++ 4 files changed, 278 insertions(+), 21 deletions(-) diff --git a/frontend/index.html b/frontend/index.html index 033a81c..b317852 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -285,6 +285,7 @@
Today is saved on this device.
+
Later is saved on this device.
diff --git a/frontend/today-session-sync.js b/frontend/today-session-sync.js index b3e90ca..f96c90e 100644 --- a/frontend/today-session-sync.js +++ b/frontend/today-session-sync.js @@ -5,9 +5,19 @@ function createTodaySessionSync({ let current = null; let ownedRevision = 0; let pollTimer = null; + let requestTail = Promise.resolve(); + let latestSnapshot = null; + let publishRequested = 0; + let publishSent = 0; const endpoint = 'api/v1/today/session'; const deviceId = () => String(getDeviceId?.() || '').trim(); + function serialize(operation) { + const result = requestTail.then(operation, operation); + requestTail = result.catch(() => null); + return result; + } + function adopt(session) { if (!session || !Number.isInteger(session.revision)) return null; const previousOwned = current?.device_id === deviceId() && current?.running; @@ -27,8 +37,9 @@ function createTodaySessionSync({ return session; } - async function refresh() { + async function refreshNow() { try { + onStatus('syncing'); const session = await fetchJson(endpoint); onStatus('online'); return adopt(session); @@ -38,29 +49,39 @@ function createTodaySessionSync({ } } - async function claim() { - if (!current?.running || !current.identity || current.device_id === deviceId() || !deviceId()) return null; - try { - const session = await fetchJson(endpoint, { - method:'PATCH', headers:{'Content-Type':'application/json'}, - body:JSON.stringify({ - base_revision:current.revision, device_id:deviceId(), identity:current.identity, - elapsed_ms:current.elapsed_ms, running:true, - }), - }); - adopt(session); - timer?.adopt?.(session.identity, session.elapsed_ms, session.running); - return session; - } catch (error) { - onStatus('conflict', error); - await refresh(); - return null; - } + const refresh = () => serialize(refreshNow); + + function claim() { + return serialize(async () => { + if (!current?.running || !current.identity || current.device_id === deviceId() || !deviceId()) return null; + try { + onStatus('syncing'); + const session = await fetchJson(endpoint, { + method:'PATCH', headers:{'Content-Type':'application/json'}, + body:JSON.stringify({ + base_revision:current.revision, device_id:deviceId(), identity:current.identity, + elapsed_ms:current.elapsed_ms, running:true, + }), + }); + adopt(session); + timer?.adopt?.(session.identity, session.elapsed_ms, session.running); + return session; + } catch (error) { + if (error?.status === 409 || error?.code === 'session_changed') { + onStatus('conflict', error); + await refreshNow(); + } else { + onStatus('offline', error); + } + return null; + } + }); } - async function publish(snapshot = timer?.snapshot?.()) { + async function publishNow(snapshot) { if (!snapshot?.identity || !deviceId()) return null; try { + onStatus('syncing'); const session = await fetchJson(endpoint, { method:'PATCH', headers:{'Content-Type':'application/json'}, body:JSON.stringify({ @@ -72,11 +93,36 @@ function createTodaySessionSync({ adopt(session); return session; } catch (error) { - onStatus('offline', error); + if (error?.status === 409 || error?.code === 'session_changed') { + onStatus('conflict', error); + await refreshNow(); + } else { + onStatus('offline', error); + } return null; } } + function publish(snapshot = timer?.snapshot?.()) { + snapshot = snapshot?.identity ? snapshot : latestSnapshot; + if (!snapshot?.identity || !deviceId()) return Promise.resolve(null); + latestSnapshot = snapshot; + publishRequested += 1; + return serialize(async () => { + if (publishSent === publishRequested) return current; + if (current?.device_id && current.device_id !== deviceId()) { + publishSent = publishRequested; + return current; + } + const version = publishRequested; + const result = await publishNow(latestSnapshot); + publishSent = current?.device_id && current.device_id !== deviceId() + ? publishRequested + : version; + return result; + }); + } + const pulse = () => current?.device_id === deviceId() && current?.running ? publish() : refresh(); @@ -113,8 +159,17 @@ function attachTodaySessionHandoff({ return value; }; let offered = null; + const sessionStatus = qs('#today-session-sync-status'); + const showSessionStatus = message => { + sessionStatus.textContent = message; + sessionStatus.hidden = !message; + }; const sync = createTodaySessionSync({ fetchJson, getDeviceId, timer, + onStatus:state => showSessionStatus( + state === 'syncing' ? 'Session syncing…' : + (state === 'offline' ? 'Session offline · will retry.' : '') + ), onRemote:session => { offered = session; const handoff = qs('#today-session-handoff'); @@ -126,6 +181,7 @@ function attachTodaySessionHandoff({ `${item?.title || 'Current Today item'} · ${minutes} min elapsed`; }, onTransferred:() => { + showSessionStatus('Session continued on another device.'); announce('Today continued on another device. Timer paused here.'); renderTimer(); }, diff --git a/tests/test_today_session_handoff_ui.py b/tests/test_today_session_handoff_ui.py index cd001bb..33377ff 100644 --- a/tests/test_today_session_handoff_ui.py +++ b/tests/test_today_session_handoff_ui.py @@ -15,9 +15,13 @@ def test_dashboard_packages_a_mobile_today_session_handoff(): assert 'id="today-session-handoff"' in html assert 'id="continue-today-session"' in html + assert 'id="today-session-sync-status"' in html assert 'aria-live="polite"' in html assert "createTodaySessionSync" in today_bundle assert "attachTodaySessionHandoff" in today_bundle + assert "Session syncing…" in today_bundle + assert "Session offline · will retry." in today_bundle + assert "Session continued on another device." in today_bundle assert "todaySessionSync = attachTodaySessionHandoff" in dashboard assert "todaySessionSync?.publish(snapshot)" in dashboard assert "min-height:44px" in css diff --git a/tests/test_today_session_sync.py b/tests/test_today_session_sync.py index 5095a75..17437df 100644 --- a/tests/test_today_session_sync.py +++ b/tests/test_today_session_sync.py @@ -119,3 +119,199 @@ const sync = createTodaySessionSync({ "running": True, }, } + + +def test_publish_conflict_adopts_remote_owner_and_pauses_once(): + result = run_node( + r""" +const calls = []; +const owned = {revision:2, device_id:'desktop-a', identity:'issue:r:42:', elapsed_ms:1000, running:true}; +const remote = {...owned, revision:3, device_id:'phone-b', elapsed_ms:2500}; +let pauses = 0; +const transfers = []; +const statuses = []; +let reads = 0; +const sync = createTodaySessionSync({ + getDeviceId:()=> 'desktop-a', + fetchJson:async (url, options={}) => { + calls.push(options.method || 'GET'); + if (options.method === 'PATCH') { + const error = new Error('session changed'); + error.status = 409; + error.code = 'session_changed'; + throw error; + } + reads += 1; + return reads === 1 ? owned : remote; + }, + timer:{ + snapshot:()=>({identity:owned.identity, elapsed_ms:1500, running:true}), + pause:()=>{pauses += 1;}, + }, + onTransferred:session=>transfers.push(session.device_id), + onStatus:status=>statuses.push(status), +}); +(async()=>{ + await sync.refresh(); + await sync.publish(); + await sync.refresh(); + process.stdout.write(JSON.stringify({calls, pauses, transfers, statuses, session:sync.session()})); +})().catch(error=>{console.error(error);process.exit(1);}); +""" + ) + + assert result["calls"] == ["GET", "PATCH", "GET", "GET"] + assert result["session"]["device_id"] == "phone-b" + assert result["pauses"] == 1 + assert result["transfers"] == ["phone-b"] + assert "conflict" in result["statuses"] + + +def test_rapid_publishes_are_single_flight_and_coalesce_latest_snapshot(): + result = run_node( + r""" +const owned = {revision:1, device_id:'desktop-a', identity:'issue:r:42:', elapsed_ms:0, running:true}; +const bodies = []; +let active = 0; +let peak = 0; +let releaseFirst; +let patchCount = 0; +const firstPatch = new Promise(resolve=>{releaseFirst=resolve;}); +const sync = createTodaySessionSync({ + getDeviceId:()=> 'desktop-a', + fetchJson:async (url, options={}) => { + if (!options.method) return owned; + active += 1; + peak = Math.max(peak, active); + patchCount += 1; + const body = JSON.parse(options.body); + bodies.push(body); + if (patchCount === 1) await firstPatch; + active -= 1; + return {...owned, revision:1 + patchCount, elapsed_ms:body.elapsed_ms, running:body.running}; + }, +}); +(async()=>{ + await sync.refresh(); + const first = sync.publish({identity:owned.identity, elapsed_ms:1000, running:true}); + await Promise.resolve(); + const middle = sync.publish({identity:owned.identity, elapsed_ms:2000, running:false}); + const latest = sync.publish({identity:owned.identity, elapsed_ms:3000, running:true}); + releaseFirst(); + await Promise.all([first, middle, latest]); + process.stdout.write(JSON.stringify({peak, bodies, session:sync.session()})); +})().catch(error=>{console.error(error);process.exit(1);}); +""" + ) + + assert result["peak"] == 1 + assert [body["elapsed_ms"] for body in result["bodies"]] == [1000, 3000] + assert result["session"]["elapsed_ms"] == 3000 + + +def test_remote_conflict_discards_queued_local_publishes_until_explicit_claim(): + result = run_node( + r""" +const owned = {revision:2, device_id:'desktop-a', identity:'issue:r:42:', elapsed_ms:1000, running:true}; +const remote = {...owned, revision:3, device_id:'phone-b', elapsed_ms:2500}; +const calls = []; +let reads = 0; +let releaseConflict; +const conflictReady = new Promise(resolve=>{releaseConflict=resolve;}); +const sync = createTodaySessionSync({ + getDeviceId:()=> 'desktop-a', + fetchJson:async (url, options={}) => { + calls.push(options.method || 'GET'); + if (options.method === 'PATCH') { + await conflictReady; + const error = new Error('session changed'); + error.status = 409; + throw error; + } + reads += 1; + return reads === 1 ? owned : remote; + }, + timer:{pause:()=>{}}, +}); +(async()=>{ + await sync.refresh(); + const stale = sync.publish({identity:owned.identity, elapsed_ms:1500, running:true}); + await Promise.resolve(); + const queued = sync.publish({identity:owned.identity, elapsed_ms:2000, running:true}); + releaseConflict(); + await Promise.all([stale, queued]); + process.stdout.write(JSON.stringify({calls, session:sync.session()})); +})().catch(error=>{console.error(error);process.exit(1);}); +""" + ) + + assert result["calls"] == ["GET", "PATCH", "GET"] + assert result["session"]["device_id"] == "phone-b" + + +def test_transient_failure_retains_latest_intent_for_retry(): + result = run_node( + r""" +const owned = {revision:4, device_id:'desktop-a', identity:'issue:r:42:', elapsed_ms:1000, running:true}; +const bodies = []; +const statuses = []; +let attempts = 0; +const sync = createTodaySessionSync({ + getDeviceId:()=> 'desktop-a', + fetchJson:async (url, options={}) => { + if (!options.method) return owned; + attempts += 1; + bodies.push(JSON.parse(options.body)); + if (attempts === 1) { + const error = new Error('temporarily unavailable'); + error.status = 503; + throw error; + } + return {...owned, revision:5, elapsed_ms:bodies.at(-1).elapsed_ms}; + }, + onStatus:status=>statuses.push(status), +}); +(async()=>{ + await sync.refresh(); + const first = await sync.publish({identity:owned.identity, elapsed_ms:4500, running:true}); + const retried = await sync.publish(); + process.stdout.write(JSON.stringify({first, retried, bodies, statuses})); +})().catch(error=>{console.error(error);process.exit(1);}); +""" + ) + + assert result["first"] is None + assert [body["elapsed_ms"] for body in result["bodies"]] == [4500, 4500] + assert result["retried"]["revision"] == 5 + assert "offline" in result["statuses"] + + +def test_claim_transport_failure_stays_retryable_without_false_conflict(): + result = run_node( + r""" +const remote = {revision:7, device_id:'phone-b', identity:'issue:r:42:', elapsed_ms:9000, running:true}; +const calls = []; +const statuses = []; +const sync = createTodaySessionSync({ + getDeviceId:()=> 'desktop-a', + fetchJson:async (url, options={}) => { + calls.push(options.method || 'GET'); + if (!options.method) return remote; + const error = new Error('gateway unavailable'); + error.status = 503; + throw error; + }, + onStatus:status=>statuses.push(status), +}); +(async()=>{ + await sync.refresh(); + const claimed = await sync.claim(); + process.stdout.write(JSON.stringify({claimed, calls, statuses})); +})().catch(error=>{console.error(error);process.exit(1);}); +""" + ) + + assert result["claimed"] is None + assert result["calls"] == ["GET", "PATCH"] + assert result["statuses"][-1] == "offline" + assert "conflict" not in result["statuses"]