From 8f7a836d28785d669627fd884aeafc53e61a8745 Mon Sep 17 00:00:00 2001 From: timmy Date: Sun, 9 Aug 2026 06:05:40 +0000 Subject: [PATCH] feat: batch queued planning edits (#375) --- frontend/later-sync.js | 22 ++++-- frontend/service-worker.js | 2 +- frontend/today-sync.js | 39 +++++------ src/later_store.py | 66 ++++++++++++++++++ src/main.py | 24 ++++++- src/today_store.py | 81 +++++++++++++++++++++++ tests/test_later_store.py | 43 ++++++++++-- tests/test_later_sync.py | 39 ++++++++--- tests/test_markdown_renderer.py | 2 +- tests/test_mobile_composer_integration.py | 2 +- tests/test_service_worker.py | 12 ++-- tests/test_today_store.py | 47 +++++++++++-- tests/test_today_sync.py | 50 +++++++++----- 13 files changed, 353 insertions(+), 76 deletions(-) diff --git a/frontend/later-sync.js b/frontend/later-sync.js index d97db23..c8c4f4a 100644 --- a/frontend/later-sync.js +++ b/frontend/later-sync.js @@ -184,19 +184,29 @@ function createLaterSync({ storage, getLogin, fetchJson, onRemoteRecords, onStat if (!ownerKey) return false; ensureChannel(); try { - let plan = await fetchJson('api/v1/later'); let operations = pending(); + let plan; + if (!operations.length) plan = await fetchJson('api/v1/later'); while (operations.length) { if (key() !== ownerKey) return false; - const operation = operations[0]; + const batch = operations.slice(0, 50); plan = await fetchJson('api/v1/later', { method: 'PATCH', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(operation), + body: JSON.stringify({ operations: batch }), }); - if (pending().some(candidate => candidate.operation_id === operation.operation_id) && - !removeOperation(operation.operation_id)) { - throw new Error('Could not persist Later delivery receipt'); + const hasReceipts = Array.isArray(plan.accepted_operation_ids) || + Array.isArray(plan.duplicate_operation_ids) || Array.isArray(plan.rejected_operations); + const received = hasReceipts ? [ + ...(plan.accepted_operation_ids || []), + ...(plan.duplicate_operation_ids || []), + ...(plan.rejected_operations || []).map(item => item.operation_id), + ] : batch.map(item => item.operation_id); + for (const operationId of received) { + if (pending().some(candidate => candidate.operation_id === operationId) && + !removeOperation(operationId)) { + throw new Error('Could not persist Later delivery receipt'); + } } operations = pending(); } diff --git a/frontend/service-worker.js b/frontend/service-worker.js index 50187e3..76d1653 100644 --- a/frontend/service-worker.js +++ b/frontend/service-worker.js @@ -1,6 +1,6 @@ const BASE = new URL('./', self.location.href).pathname; importScripts(BASE + 'static/background-issue-sync.js'); -const CACHE = 'stackchain-dashboard-shell-v53'; +const CACHE = 'stackchain-dashboard-shell-v54'; const OUTAGE_STATUSES = new Set([500, 502, 503, 504]); const NAVIGATION_TIMEOUT_MS = self.__STACKCHAIN_NAVIGATION_TIMEOUT_MS || 4000; const SHELL = [ diff --git a/frontend/today-sync.js b/frontend/today-sync.js index 24ffeda..72009cd 100644 --- a/frontend/today-sync.js +++ b/frontend/today-sync.js @@ -170,30 +170,31 @@ function createTodaySync({ storage, getLogin, fetchJson, onRemoteIds, onStatus, if (!ownerKey) return false; ensureChannel(); try { - let plan = await fetchJson('api/v1/today'); let operations = pending(); + let plan; let hadConflict = false; + if (!operations.length) plan = await fetchJson('api/v1/today'); while (operations.length) { if (key() !== ownerKey) return false; - const operation = operations[0]; - try { - plan = await fetchJson('api/v1/today', { - method: 'PATCH', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(operation), - }); - } catch (error) { - if (error?.status !== 409) throw error; - if (!removeOperation(operation.operation_id)) { - throw new Error('Could not persist rejected Today operation'); + const batch = operations.slice(0, 50); + plan = await fetchJson('api/v1/today', { + method: 'PATCH', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ operations: batch }), + }); + const hasReceipts = Array.isArray(plan.accepted_operation_ids) || + Array.isArray(plan.duplicate_operation_ids) || Array.isArray(plan.rejected_operations); + const received = hasReceipts ? [ + ...(plan.accepted_operation_ids || []), + ...(plan.duplicate_operation_ids || []), + ...(plan.rejected_operations || []).map(item => item.operation_id), + ] : batch.map(item => item.operation_id); + hadConflict = hadConflict || Boolean(plan.rejected_operations?.length); + for (const operationId of received) { + if (pending().some(candidate => candidate.operation_id === operationId) && + !removeOperation(operationId)) { + throw new Error('Could not persist Today delivery receipt'); } - hadConflict = true; - operations = pending(); - continue; - } - if (pending().some(candidate => candidate.operation_id === operation.operation_id) && - !removeOperation(operation.operation_id)) { - throw new Error('Could not persist Today delivery receipt'); } operations = pending(); } diff --git a/src/later_store.py b/src/later_store.py index d93ea24..5580b92 100644 --- a/src/later_store.py +++ b/src/later_store.py @@ -122,3 +122,69 @@ class LaterStore: (login, operation_id), ) return {"revision": revision, "records": records} + + def apply_batch(self, login: str, operations: list[dict]) -> dict: + """Apply ordered deferrals with one SQLite write transaction.""" + login = self._normalize_login(login) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + "SELECT revision, records FROM later_plans WHERE login = ?", (login,) + ).fetchone() + snapshot = self._snapshot(row) + records = dict(snapshot["records"]) + revision = snapshot["revision"] + accepted: list[str] = [] + duplicates: list[str] = [] + + for operation in operations: + operation_id = operation.get("operation_id", "") + action = operation.get("action", "") + item_id = operation.get("item_id", "") + if not operation_id or not item_id: + raise ValueError("operation_id and item_id are required") + if action not in {"defer", "restore"}: + raise ValueError("unsupported Later action") + wake_at = operation.get("wake_at") + if action == "defer": + wake_at = self._validate_wake_at(wake_at) + duplicate = connection.execute( + "SELECT 1 FROM later_operations WHERE login = ? AND operation_id = ?", + (login, operation_id), + ).fetchone() + if duplicate: + duplicates.append(operation_id) + continue + + before = records.get(item_id) + if action == "defer": + records[item_id] = wake_at + changed = before != wake_at + else: + changed = item_id in records + records.pop(item_id, None) + revision += 1 if changed else 0 + connection.execute( + "INSERT INTO later_operations(login, operation_id) VALUES (?, ?)", + (login, operation_id), + ) + accepted.append(operation_id) + + serialized = json.dumps(records, separators=(",", ":"), sort_keys=True) + if row is None: + connection.execute( + "INSERT INTO later_plans(login, revision, records) VALUES (?, ?, ?)", + (login, revision, serialized), + ) + elif accepted: + connection.execute( + "UPDATE later_plans SET revision = ?, records = ? WHERE login = ?", + (revision, serialized, login), + ) + return { + "revision": revision, + "records": records, + "accepted_operation_ids": accepted, + "duplicate_operation_ids": duplicates, + "rejected_operations": [], + } diff --git a/src/main.py b/src/main.py index 33356b4..d1450bc 100644 --- a/src/main.py +++ b/src/main.py @@ -224,6 +224,14 @@ class LaterOperation(BaseModel): wake_at: str | None = Field(default=None, max_length=100) +class TodayOperationBatch(BaseModel): + operations: list[TodayOperation] = Field(min_length=1, max_length=50) + + +class LaterOperationBatch(BaseModel): + operations: list[LaterOperation] = Field(min_length=1, max_length=50) + + class NotificationReply(BaseModel): body: str = Field(min_length=1, max_length=10_000) @@ -904,9 +912,15 @@ async def get_today_plan(): @app.patch("/api/v1/today") -async def update_today_plan(payload: TodayOperation): +async def update_today_plan(payload: TodayOperation | TodayOperationBatch): login = await _confirmed_login() try: + if isinstance(payload, TodayOperationBatch): + return await asyncio.to_thread( + _today_store().apply_batch, + login, + [operation.model_dump() for operation in payload.operations], + ) return await asyncio.to_thread( _today_store().apply, login, @@ -939,9 +953,15 @@ async def get_later_plan(): @app.patch("/api/v1/later") -async def update_later_plan(payload: LaterOperation): +async def update_later_plan(payload: LaterOperation | LaterOperationBatch): login = await _confirmed_login() try: + if isinstance(payload, LaterOperationBatch): + return await asyncio.to_thread( + _later_store().apply_batch, + login, + [operation.model_dump() for operation in payload.operations], + ) return await asyncio.to_thread( _later_store().apply, login, diff --git a/src/today_store.py b/src/today_store.py index 96d3ab0..972624c 100644 --- a/src/today_store.py +++ b/src/today_store.py @@ -128,3 +128,84 @@ class TodayStore: (login, operation_id), ) return {"revision": revision, "ids": ids} + + def apply_batch(self, login: str, operations: list[dict]) -> dict: + """Apply an ordered batch with one lock and receipt per operation.""" + login = self._normalize_login(login) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + "SELECT revision, ids FROM today_plans WHERE login = ?", (login,) + ).fetchone() + snapshot = self._snapshot(row) + ids = list(snapshot["ids"]) + revision = snapshot["revision"] + accepted: list[str] = [] + duplicates: list[str] = [] + rejected: list[dict[str, str]] = [] + + for operation in operations: + operation_id = operation.get("operation_id", "") + action = operation.get("action", "") + item_id = operation.get("item_id", "") + direction = operation.get("direction") + if not operation_id or not item_id: + raise ValueError("operation_id and item_id are required") + if action not in {"add", "remove", "move"}: + raise ValueError("unsupported Today action") + if action == "move" and direction not in {"up", "down"}: + raise ValueError("move direction must be up or down") + duplicate = connection.execute( + "SELECT 1 FROM today_operations WHERE login = ? AND operation_id = ?", + (login, operation_id), + ).fetchone() + if duplicate: + duplicates.append(operation_id) + continue + + changed = False + if action == "add": + if item_id not in ids: + if len(ids) >= self.limit: + rejected.append({"operation_id": operation_id, "reason": "today_full"}) + continue + ids.append(item_id) + changed = True + elif action == "remove": + if item_id in ids: + ids.remove(item_id) + changed = True + else: + try: + index = ids.index(item_id) + except ValueError: + index = -1 + target = index - 1 if direction == "up" else index + 1 + if index >= 0 and 0 <= target < len(ids): + ids[index], ids[target] = ids[target], ids[index] + changed = True + revision += 1 if changed else 0 + connection.execute( + "INSERT INTO today_operations(login, operation_id) VALUES (?, ?)", + (login, operation_id), + ) + accepted.append(operation_id) + + serialized = json.dumps(ids, separators=(",", ":")) + if row is None: + connection.execute( + "INSERT INTO today_plans(login, revision, ids) VALUES (?, ?, ?)", + (login, revision, serialized), + ) + elif accepted: + connection.execute( + "UPDATE today_plans SET revision = ?, ids = ? WHERE login = ?", + (revision, serialized, login), + ) + return { + "revision": revision, + "ids": ids, + "accepted_operation_ids": accepted, + "duplicate_operation_ids": duplicates, + "rejected_operations": rejected, + } diff --git a/tests/test_later_store.py b/tests/test_later_store.py index a511992..6d88352 100644 --- a/tests/test_later_store.py +++ b/tests/test_later_store.py @@ -65,6 +65,27 @@ def test_deferrals_are_durable_revisioned_idempotent_and_account_scoped(tmp_path ) == {"revision": 2, "records": {}} +def test_batch_applies_in_one_ordered_idempotent_unit(tmp_path): + store = LaterStore(tmp_path / "later.sqlite3") + operations = [ + {"operation_id": "first", "action": "defer", "item_id": "issue:r:1:", "wake_at": "2026-08-10T09:00:00.000Z"}, + {"operation_id": "second", "action": "restore", "item_id": "issue:r:1:"}, + ] + + result = store.apply_batch("timmy", operations) + assert result == { + "revision": 2, + "records": {}, + "accepted_operation_ids": ["first", "second"], + "duplicate_operation_ids": [], + "rejected_operations": [], + } + replay = store.apply_batch("timmy", operations) + assert replay["revision"] == 2 + assert replay["accepted_operation_ids"] == [] + assert replay["duplicate_operation_ids"] == ["first", "second"] + + @pytest.mark.anyio async def test_authenticated_later_api_uses_confirmed_account_and_csrf(monkeypatch, tmp_path): monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator") @@ -98,10 +119,10 @@ async def test_authenticated_later_api_uses_confirmed_account_and_csrf(monkeypat changed = await client.patch( "/api/v1/later", json={ - "operation_id": "mobile-1", - "action": "defer", - "item_id": "issue:stackchain/dashboard:363:", - "wake_at": "2026-08-10T09:00:00.000Z", + "operations": [ + {"operation_id": "mobile-1", "action": "defer", "item_id": "issue:stackchain/dashboard:363:", "wake_at": "2026-08-10T09:00:00.000Z"}, + {"operation_id": "mobile-2", "action": "defer", "item_id": "issue:stackchain/dashboard:365:", "wake_at": "2026-08-11T09:00:00.000Z"}, + ], }, headers={ "Origin": "https://test", @@ -112,10 +133,18 @@ async def test_authenticated_later_api_uses_confirmed_account_and_csrf(monkeypat assert forbidden.status_code == 403 assert changed.status_code == 200 - assert changed.json() == fetched.json() == { - "revision": 1, + assert changed.json() == { + "revision": 2, "records": { - "issue:stackchain/dashboard:363:": "2026-08-10T09:00:00.000Z" + "issue:stackchain/dashboard:363:": "2026-08-10T09:00:00.000Z", + "issue:stackchain/dashboard:365:": "2026-08-11T09:00:00.000Z", }, + "accepted_operation_ids": ["mobile-1", "mobile-2"], + "duplicate_operation_ids": [], + "rejected_operations": [], } + assert fetched.json() == {"revision": 2, "records": { + "issue:stackchain/dashboard:363:": "2026-08-10T09:00:00.000Z", + "issue:stackchain/dashboard:365:": "2026-08-11T09:00:00.000Z", + }} assert fetched.headers["cache-control"] == "no-store" diff --git a/tests/test_later_sync.py b/tests/test_later_sync.py index 005e1f8..67d0eae 100644 --- a/tests/test_later_sync.py +++ b/tests/test_later_sync.py @@ -28,7 +28,7 @@ const values=new Map();const storage={{get length(){{return values.size}},key:i= const channels=[];const channelFactory=()=>{{const channel={{onmessage:null,postMessage:data=>channels.filter(x=>x!==channel).forEach(x=>x.onmessage?.({{data}})),close(){{}}}};channels.push(channel);return channel}}; 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 sequence=0,releaseFirst;const delivered=[]; -const make=tab=>createLaterSync({{storage,getLogin:()=> 'timmy',createOperationId:()=>tab+'-'+(++sequence),coordinator:createCoordinator({{storage,locks,channelFactory,tabId:tab}}),fetchJson:async(_url,options={{}})=>{{if(!options.method)return {{revision:0,records:{{}}}};const operation=JSON.parse(options.body);delivered.push(operation);if(delivered.length===1)await new Promise(resolve=>releaseFirst=resolve);return {{revision:delivered.length,records:{{[operation.item_id]:operation.wake_at}}}}}},onRemoteRecords:()=>{{}},onStatus:()=>{{}}}}); +const make=tab=>createLaterSync({{storage,getLogin:()=> 'timmy',createOperationId:()=>tab+'-'+(++sequence),coordinator:createCoordinator({{storage,locks,channelFactory,tabId:tab}}),fetchJson:async(_url,options={{}})=>{{if(!options.method)return {{revision:0,records:{{}}}};const operation=JSON.parse(options.body).operations[0];delivered.push(operation);if(delivered.length===1)await new Promise(resolve=>releaseFirst=resolve);return {{revision:delivered.length,records:{{[operation.item_id]:operation.wake_at}}}}}},onRemoteRecords:()=>{{}},onStatus:()=>{{}}}}); const first=make('first'),second=make('second');first.enqueue('defer','issue:r:1:','2026-08-10T09:00:00.000Z'); (async()=>{{const draining=first.flush();while(!releaseFirst)await Promise.resolve();second.enqueue('defer','issue:r:2:','2026-08-11T09:00:00.000Z');const competing=second.flush();releaseFirst();await Promise.all([draining,competing]);await first.flush();process.stdout.write(JSON.stringify({{delivered,pending:first.pending(),keys:[...values.keys()]}}));}})(); """ @@ -46,7 +46,7 @@ const values=new Map();const storage={{get length(){{return values.size}},key:i= const prefix='stackchain.later-sync.v1.timmy.operation.'; values.set(prefix+'old',JSON.stringify({{queued_at:100,operation:{{operation_id:'old',action:'defer',item_id:'issue:r:2:',wake_at:'2026-08-10T09:00:00.000Z'}}}})); values.set(prefix+'new',JSON.stringify({{queued_at:101,operation:{{operation_id:'new',action:'restore',item_id:'issue:r:2:',wake_at:null}}}})); -const delivered=[];const sync=createLaterSync({{storage,getLogin:()=> 'timmy',fetchJson:async(_url,options={{}})=>{{if(!options.method)return {{revision:0,records:{{}}}};delivered.push(JSON.parse(options.body));return {{revision:1,records:{{}}}}}},onRemoteRecords:()=>{{}},onStatus:()=>{{}}}}); +const delivered=[];const sync=createLaterSync({{storage,getLogin:()=> 'timmy',fetchJson:async(_url,options={{}})=>{{if(!options.method)return {{revision:0,records:{{}}}};delivered.push(...JSON.parse(options.body).operations);return {{revision:1,records:{{}}}}}},onRemoteRecords:()=>{{}},onStatus:()=>{{}}}}); (async()=>{{await sync.flush();process.stdout.write(JSON.stringify({{delivered,pending:sync.pending(),keys:[...values.keys()]}}));}})(); """ result = run_node(script) @@ -67,7 +67,7 @@ const sync = createLaterSync({{ fetchJson:async (url,options={{}})=>{{ requests.push({{url,body:options.body&&JSON.parse(options.body)}}); if (!options.method) return remote; - const operation=JSON.parse(options.body); + const operation=JSON.parse(options.body).operations[0]; remote={{revision:2,records:{{...remote.records,[operation.item_id]:operation.wake_at}}}}; return remote; }}, @@ -84,19 +84,38 @@ sync.enqueue('defer','issue:r:2:','2026-08-10T09:00:00.000Z'); assert [request["url"] for request in result["requests"]] == [ "api/v1/later", "api/v1/later", - "api/v1/later", ] - assert result["requests"][1]["body"] == { + assert result["requests"][0]["body"] == {"operations": [{ "operation_id": "offline-op", "action": "defer", "item_id": "issue:r:2:", "wake_at": "2026-08-10T09:00:00.000Z", - } + }]} assert result["records"]["issue:r:2:"] == "2026-08-10T09:00:00.000Z" assert result["status"] == "saved" assert result["pending"] == [] +def test_pending_later_edits_are_sent_as_one_batch_without_a_preflight_get(): + script = f""" +const createLaterSync=require({json.dumps(str(LATER_SYNC))}); +const values=new Map();const requests=[];let sequence=0; +const sync=createLaterSync({{ + storage:{{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}}, + getLogin:()=> 'timmy',createOperationId:()=> 'op-'+(++sequence), + fetchJson:async(_url,options={{}})=>{{const body=JSON.parse(options.body);requests.push(body);return {{revision:2,records:{{}},accepted_operation_ids:body.operations.map(x=>x.operation_id),duplicate_operation_ids:[],rejected_operations:[]}}}}, + onRemoteRecords:()=>{{}},onStatus:()=>{{}}, +}}); +sync.enqueue('defer','issue:r:1:','2026-08-10T09:00:00.000Z'); +sync.enqueue('restore','issue:r:2:'); +(async()=>{{await sync.flush();process.stdout.write(JSON.stringify({{requests,pending:sync.pending()}}));}})(); +""" + result=run_node(script) + assert len(result["requests"]) == 1 + assert [item["operation_id"] for item in result["requests"][0]["operations"]] == ["op-1", "op-2"] + assert result["pending"] == [] + + def test_latest_offline_intent_wins_and_failed_delivery_stays_pending(): script = f""" const createLaterSync = require({json.dumps(str(LATER_SYNC))}); @@ -145,7 +164,7 @@ sync.enqueue('defer','issue:r:2:','2026-08-10T09:00:00.000Z'); assert run_node(script) == { "scheduledBeforeSwitch": 1, "timers": 1, - "requests": ["GET"], + "requests": ["PATCH"], "pendingForAlexander": [], } @@ -159,7 +178,7 @@ const sync=createLaterSync({{ getLogin:()=> 'timmy',createOperationId:()=> 'op-'+(++sequence), fetchJson:async (_url,options={{}})=>{{ if (!options.method) return {{revision:0,records:{{}}}}; - const operation=JSON.parse(options.body); actions.push(operation.action); + const operation=JSON.parse(options.body).operations[0]; actions.push(operation.action); if (operation.action==='defer') await new Promise(resolve=>releaseFirst=resolve); return {{revision:actions.length,records:operation.action==='defer'?{{[operation.item_id]:operation.wake_at}}:{{}}}}; }},onRemoteRecords:r=>{{globalThis.records=r}},onStatus:()=>{{}}, @@ -214,7 +233,7 @@ sync.startLifecycle({{window:{{addEventListener:(n,h)=>handlers[n]=h}},document: assert run_node(script) == { "first": True, "second": False, - "requests": ["GET", "PATCH"], + "requests": ["PATCH"], "pending": [], } @@ -269,5 +288,5 @@ async def test_dashboard_syncs_every_later_change_and_exposes_account_status(): def test_later_sync_ships_atomically_in_the_offline_shell(): source = (Path(__file__).parents[1] / "frontend" / "service-worker.js").read_text() - assert "stackchain-dashboard-shell-v53" in source + assert "stackchain-dashboard-shell-v54" in source assert "BASE + 'static/later-sync.js'" in source diff --git a/tests/test_markdown_renderer.py b/tests/test_markdown_renderer.py index 94d21fb..ab9fd69 100644 --- a/tests/test_markdown_renderer.py +++ b/tests/test_markdown_renderer.py @@ -137,4 +137,4 @@ def test_markdown_work_bodies_are_mobile_safe_block_containers(): assert ".markdown-content { min-width:0; max-width:100%; overflow-wrap:anywhere;" in css assert ".markdown-content pre { max-width:100%; overflow-x:auto;" in css assert ".markdown-content a { min-height:44px;" in css - assert "stackchain-dashboard-shell-v53" in worker + assert "stackchain-dashboard-shell-v54" in worker diff --git a/tests/test_mobile_composer_integration.py b/tests/test_mobile_composer_integration.py index 8e944d9..a3179c1 100644 --- a/tests/test_mobile_composer_integration.py +++ b/tests/test_mobile_composer_integration.py @@ -35,4 +35,4 @@ def test_offline_shell_contains_every_local_dashboard_runtime_asset(): shell_assets = set(re.findall(r"BASE \+ '([^']+)'", worker.split("async function sessionCsrf", 1)[0])) assert local_assets <= shell_assets, f"Offline shell is missing: {sorted(local_assets - shell_assets)}" - assert "stackchain-dashboard-shell-v53" in worker + assert "stackchain-dashboard-shell-v54" in worker diff --git a/tests/test_service_worker.py b/tests/test_service_worker.py index 1daf9bd..e138b43 100644 --- a/tests/test_service_worker.py +++ b/tests/test_service_worker.py @@ -108,7 +108,7 @@ async function dispatchNotificationClick(route) {{ def test_duplicate_aware_capture_ships_in_a_new_offline_shell(): source = WORKER.read_text() - assert "stackchain-dashboard-shell-v53" in source + assert "stackchain-dashboard-shell-v54" in source assert "BASE + 'static/create-issue-sheet.js'" in source assert "BASE + 'static/dashboard.js'" in source @@ -116,14 +116,14 @@ def test_duplicate_aware_capture_ships_in_a_new_offline_shell(): def test_exact_later_picker_ships_atomically_in_a_new_offline_shell(): source = WORKER.read_text() - assert "stackchain-dashboard-shell-v53" in source + assert "stackchain-dashboard-shell-v54" in source assert "BASE + 'static/later-picker.js'" in source def test_navigation_deadline_ships_in_a_new_shell_cache(): source = WORKER.read_text() - assert "stackchain-dashboard-shell-v53" in source + assert "stackchain-dashboard-shell-v54" in source assert "BASE + 'static/dashboard.css'" in source assert "BASE + 'static/dashboard.js'" in source assert "BASE + 'static/install-app.js'" in source @@ -132,21 +132,21 @@ def test_navigation_deadline_ships_in_a_new_shell_cache(): def test_today_convergence_ships_in_a_new_shell_cache(): source = WORKER.read_text() - assert "stackchain-dashboard-shell-v53" in source + assert "stackchain-dashboard-shell-v54" in source assert "BASE + 'static/today-sync.js'" in source def test_mobile_search_viewport_ships_in_a_new_offline_shell(): source = WORKER.read_text() - assert "stackchain-dashboard-shell-v53" in source + assert "stackchain-dashboard-shell-v54" in source assert "BASE + 'static/mobile-search-viewport.js'" in source def test_update_ownership_flow_ships_atomically_in_a_new_offline_shell(): source = WORKER.read_text() - assert "stackchain-dashboard-shell-v53" in source + assert "stackchain-dashboard-shell-v54" in source assert "BASE + 'static/update-ownership.js'" in source diff --git a/tests/test_today_store.py b/tests/test_today_store.py index ab8c033..b748b05 100644 --- a/tests/test_today_store.py +++ b/tests/test_today_store.py @@ -77,6 +77,36 @@ def test_limit_is_atomic_and_remove_frees_capacity(tmp_path): ] +def test_batch_applies_ordered_operations_once_and_rejects_only_capacity_conflicts(tmp_path): + store = TodayStore(tmp_path / "today.sqlite3", limit=2) + + result = store.apply_batch( + "timmy", + [ + {"operation_id": "one", "action": "add", "item_id": "issue:r:1:"}, + {"operation_id": "two", "action": "add", "item_id": "issue:r:2:"}, + {"operation_id": "full", "action": "add", "item_id": "issue:r:3:"}, + {"operation_id": "remove", "action": "remove", "item_id": "issue:r:1:"}, + {"operation_id": "three", "action": "add", "item_id": "issue:r:3:"}, + ], + ) + + assert result == { + "revision": 4, + "ids": ["issue:r:2:", "issue:r:3:"], + "accepted_operation_ids": ["one", "two", "remove", "three"], + "duplicate_operation_ids": [], + "rejected_operations": [{"operation_id": "full", "reason": "today_full"}], + } + replay = store.apply_batch("timmy", [ + {"operation_id": "two", "action": "add", "item_id": "issue:r:2:"}, + {"operation_id": "three", "action": "add", "item_id": "issue:r:3:"}, + ]) + assert replay["revision"] == 4 + assert replay["accepted_operation_ids"] == [] + assert replay["duplicate_operation_ids"] == ["two", "three"] + + @pytest.mark.anyio async def test_authenticated_today_api_uses_confirmed_account_and_csrf(monkeypatch, tmp_path): monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator") @@ -109,9 +139,10 @@ async def test_authenticated_today_api_uses_confirmed_account_and_csrf(monkeypat changed = await client.patch( "/api/v1/today", json={ - "operation_id": "mobile-1", - "action": "add", - "item_id": "issue:stackchain/dashboard:357:", + "operations": [ + {"operation_id": "mobile-1", "action": "add", "item_id": "issue:stackchain/dashboard:357:"}, + {"operation_id": "mobile-2", "action": "add", "item_id": "issue:stackchain/dashboard:359:"}, + ], }, headers={ "Origin": "https://test", @@ -122,8 +153,12 @@ async def test_authenticated_today_api_uses_confirmed_account_and_csrf(monkeypat assert forbidden.status_code == 403 assert changed.status_code == 200 - assert changed.json() == fetched.json() == { - "revision": 1, - "ids": ["issue:stackchain/dashboard:357:"], + assert changed.json() == { + "revision": 2, + "ids": ["issue:stackchain/dashboard:357:", "issue:stackchain/dashboard:359:"], + "accepted_operation_ids": ["mobile-1", "mobile-2"], + "duplicate_operation_ids": [], + "rejected_operations": [], } + assert fetched.json() == {"revision": 2, "ids": ["issue:stackchain/dashboard:357:", "issue:stackchain/dashboard:359:"]} assert fetched.headers["cache-control"] == "no-store" diff --git a/tests/test_today_sync.py b/tests/test_today_sync.py index 5c09615..9e9963d 100644 --- a/tests/test_today_sync.py +++ b/tests/test_today_sync.py @@ -18,7 +18,7 @@ const channels=[]; const channelFactory=()=>{{const channel={{onmessage:null,pos 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 sequence=0,releaseFirst;const delivered=[]; const make=tab=>createTodaySync({{storage,getLogin:()=> 'timmy',createOperationId:()=>tab+'-'+(++sequence),coordinator:createCoordinator({{storage,locks,channelFactory,tabId:tab}}), - fetchJson:async(_url,options={{}})=>{{if(!options.method)return {{revision:0,ids:[]}};const operation=JSON.parse(options.body);delivered.push(operation.operation_id);if(delivered.length===1)await new Promise(resolve=>releaseFirst=resolve);return {{revision:delivered.length,ids:delivered}}}},onRemoteIds:()=>{{}},onStatus:()=>{{}}}}); + fetchJson:async(_url,options={{}})=>{{if(!options.method)return {{revision:0,ids:[]}};const operation=JSON.parse(options.body).operations[0];delivered.push(operation.operation_id);if(delivered.length===1)await new Promise(resolve=>releaseFirst=resolve);return {{revision:delivered.length,ids:delivered}}}},onRemoteIds:()=>{{}},onStatus:()=>{{}}}}); const first=make('first'),second=make('second');first.enqueue('add','issue:r:1:'); (async()=>{{const draining=first.flush();while(!releaseFirst)await Promise.resolve();second.enqueue('add','issue:r:2:');const competing=second.flush();releaseFirst();await Promise.all([draining,competing]);await first.flush();process.stdout.write(JSON.stringify({{delivered,pending:first.pending(),keys:[...values.keys()]}}));}})(); """ @@ -42,7 +42,7 @@ const sync = createTodaySync({{ getLogin: () => 'timmy', createOperationId: () => 'op-' + (++sequence), fetchJson: async (_url, options={{}}) => {{ if (!options.method) return {{revision:0, ids:[]}}; - const operation = JSON.parse(options.body); + const operation = JSON.parse(options.body).operations[0]; patches.push(operation); if (patches.length === 1) {{ globalThis.firstPatchStarted(); @@ -86,7 +86,7 @@ sync.enqueue('add', 'issue:r:1:'); def test_inflight_today_drain_ships_in_a_new_offline_shell(): source = (Path(__file__).parents[1] / "frontend" / "service-worker.js").read_text() - assert "stackchain-dashboard-shell-v53" in source + assert "stackchain-dashboard-shell-v54" in source assert "BASE + 'static/today-sync.js'" in source @@ -108,7 +108,7 @@ const sync = createTodaySync({{ fetchJson: async (url, options={{}}) => {{ requests.push({{url, body: options.body && JSON.parse(options.body)}}); if (!options.method) return remote; - remote = {{revision: remote.revision + 1, ids:['issue:r:9:', options.body && JSON.parse(options.body).item_id]}}; + remote = {{revision: remote.revision + 1, ids:['issue:r:9:', JSON.parse(options.body).operations[0].item_id]}}; return remote; }}, onRemoteIds: ids => {{ globalThis.adopted = ids; }}, @@ -131,19 +131,37 @@ sync.enqueue('add', 'issue:r:2:'); assert [request["url"] for request in result["requests"]] == [ "api/v1/today", "api/v1/today", - "api/v1/today", ] - assert result["requests"][1]["body"] == { + assert result["requests"][0]["body"] == {"operations": [{ "operation_id": "fixed-op", "action": "add", "item_id": "issue:r:2:", "direction": None, - } + }]} assert result["adopted"] == ["issue:r:9:", "issue:r:2:"] assert result["status"] == "saved" assert result["pending"] == [] +def test_pending_today_edits_are_sent_as_one_batch_without_a_preflight_get(): + script = f""" +const createTodaySync = require({json.dumps(str(TODAY_SYNC))}); +const values = new Map(); const requests=[]; let sequence=0; +const sync=createTodaySync({{ + storage:{{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}}, + getLogin:()=> 'timmy',createOperationId:()=> 'op-'+(++sequence), + fetchJson:async (_url,options={{}})=>{{const body=JSON.parse(options.body);requests.push(body);return {{revision:2,ids:['issue:r:1:','issue:r:2:'],accepted_operation_ids:body.operations.map(x=>x.operation_id),duplicate_operation_ids:[],rejected_operations:[]}}}}, + onRemoteIds:()=>{{}},onStatus:()=>{{}}, +}}); +sync.enqueue('add','issue:r:1:');sync.enqueue('add','issue:r:2:'); +(async()=>{{await sync.flush();process.stdout.write(JSON.stringify({{requests,pending:sync.pending()}}));}})(); +""" + result = json.loads(subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True).stdout) + assert len(result["requests"]) == 1 + assert [item["operation_id"] for item in result["requests"][0]["operations"]] == ["op-1", "op-2"] + assert result["pending"] == [] + + def test_failed_delivery_stays_pending_for_offline_replay(): script = f""" const createTodaySync = require({json.dumps(str(TODAY_SYNC))}); @@ -183,7 +201,7 @@ const sync = createTodaySync({{ fetchJson:async (_url,options={{}})=>{{ attempts += 1; if (attempts === 1) {{ const error = new Error('busy'); error.status=503; error.retryAfter=2; throw error; }} - if (options.method) patches.push(JSON.parse(options.body)); + if (options.method) patches.push(...JSON.parse(options.body).operations); return options.method ? {{revision:1,ids:['issue:r:2:']}} : {{revision:0,ids:[]}}; }}, onRemoteIds:()=>{{}}, onStatus:(state,detail)=>statuses.push([state,detail?.delayMs||null]), @@ -201,7 +219,7 @@ sync.enqueue('add','issue:r:2:'); ) assert result["scheduled"] == [2000] - assert result["attempts"] == 3 + assert result["attempts"] == 2 assert result["patches"] == [{ "operation_id": "stable-op", "action": "add", "item_id": "issue:r:2:", "direction": None, }] @@ -277,8 +295,8 @@ const sync = createTodaySync({{ storage: {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}}, getLogin: () => 'timmy', createOperationId: () => 'sixth', fetchJson: async (_url, options={{}}) => {{ - if (options.method) {{ patch = true; const error = new Error('full'); error.status = 409; throw error; }} - return {{revision:5, ids:['1','2','3','4','5']}}; + patch = true; + return {{revision:5, ids:['1','2','3','4','5'], accepted_operation_ids:[], duplicate_operation_ids:[], rejected_operations:[{{operation_id:'sixth',reason:'today_full'}}]}}; }}, onRemoteIds:ids=>{{globalThis.ids=ids}}, onStatus:value=>{{globalThis.status=value}}, }}); @@ -307,11 +325,9 @@ const sync = createTodaySync({{ storage: {{getItem:k=>values.get(k)||null,setItem:(k,v)=>values.set(k,v),removeItem:k=>values.delete(k)}}, getLogin: () => 'timmy', createOperationId: () => 'op-' + (++sequence), fetchJson: async (_url, options={{}}) => {{ - if (!options.method) return {{revision:5, ids:['1','2','3','4','5']}}; - const operation = JSON.parse(options.body); - patches.push(operation); - if (operation.action === 'add') {{ const error = new Error('full'); error.status = 409; throw error; }} - return {{revision:6, ids:['1','2','3','4']}}; + const operations = JSON.parse(options.body).operations; + patches.push(...operations); + return {{revision:6, ids:['1','2','3','4'], accepted_operation_ids:['op-2'], duplicate_operation_ids:[], rejected_operations:[{{operation_id:'op-1',reason:'today_full'}}]}}; }}, onRemoteIds: ids => {{ globalThis.ids = ids; }}, onStatus: status => statuses.push(status), @@ -413,7 +429,7 @@ sync.startLifecycle({{ subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True).stdout ) - assert result == {"requests": ["GET", "PATCH"], "pending": []} + assert result == {"requests": ["PATCH"], "pending": []} def test_foregrounding_a_stale_tab_refreshes_the_saved_plan():