Merge pull request 'Drain Today edits queued during an in-flight sync' (#368) from timmy/367-drain-today-inflight into main

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-v49';
+const CACHE = 'stackchain-dashboard-shell-v50';
 const OUTAGE_STATUSES = new Set([500, 502, 503, 504]);
 const NAVIGATION_TIMEOUT_MS = self.__STACKCHAIN_NAVIGATION_TIMEOUT_MS || 4000;
 const SHELL = [

frontend/today-sync.js

@@ -143,9 +143,11 @@ function createTodaySync({ storage, getLogin, fetchJson, onRemoteIds, onStatus,
     ensureChannel();
     try {
       let plan = await fetchJson('api/v1/today');
-      const operations = pending();
-      for (const operation of operations) {
+      let operations = pending();
+      let hadConflict = false;
+      while (operations.length) {
         if (key() !== ownerKey) return false;
+        const operation = operations[0];
         try {
           plan = await fetchJson('api/v1/today', {
             method: 'PATCH',
@@ -155,21 +157,25 @@ function createTodaySync({ storage, getLogin, fetchJson, onRemoteIds, onStatus,
         } catch (error) {
           if (error?.status !== 409) throw error;
           const rejected = pending();
-          save(rejected.filter(candidate => candidate.operation_id !== operation.operation_id));
-          adopt(plan);
-          onStatus?.('full');
-          retryAttempt = 0;
-          cancelRetry();
-          return false;
+          if (!save(rejected.filter(candidate => candidate.operation_id !== operation.operation_id))) {
+            throw new Error('Could not persist rejected Today operation');
+          }
+          hadConflict = true;
+          operations = pending();
+          continue;
         }
         const remaining = pending();
-        if (remaining[0]?.operation_id === operation.operation_id) save(remaining.slice(1));
+        const delivered = remaining.findIndex(candidate => candidate.operation_id === operation.operation_id);
+        if (delivered >= 0 && !save(remaining.filter((_, index) => index !== delivered))) {
+          throw new Error('Could not persist Today delivery receipt');
+        }
+        operations = pending();
       }
       adopt(plan);
-      onStatus?.(pending().length ? 'pending' : 'saved');
+      onStatus?.(pending().length ? 'pending' : hadConflict ? 'full' : 'saved');
       retryAttempt = 0;
       cancelRetry();
-      return true;
+      return !hadConflict;
     } catch (error) {
       if (pending().length) scheduleRetry(error, ownerKey);
       else onStatus?.('error');

tests/test_later_sync.py

@@ -233,5 +233,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-v49" in source
+    assert "stackchain-dashboard-shell-v50" in source
     assert "BASE + 'static/later-sync.js'" in source

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-v49" in worker
+    assert "stackchain-dashboard-shell-v50" in worker

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-v49" in worker
+    assert "stackchain-dashboard-shell-v50" in worker

tests/test_service_worker.py

@@ -108,7 +108,7 @@ async function dispatchNotificationClick(route) {{
 def test_navigation_deadline_ships_in_a_new_shell_cache():
     source = WORKER.read_text()
 
-    assert "stackchain-dashboard-shell-v49" in source
+    assert "stackchain-dashboard-shell-v50" in source
     assert "BASE + 'static/dashboard.css'" in source
     assert "BASE + 'static/dashboard.js'" in source
     assert "BASE + 'static/install-app.js'" in source
@@ -117,21 +117,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-v49" in source
+    assert "stackchain-dashboard-shell-v50" 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-v49" in source
+    assert "stackchain-dashboard-shell-v50" 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-v49" in source
+    assert "stackchain-dashboard-shell-v50" in source
     assert "BASE + 'static/update-ownership.js'" in source
 
 

tests/test_today_sync.py

@@ -6,6 +6,67 @@ from pathlib import Path
 TODAY_SYNC = Path(__file__).parents[1] / "frontend" / "today-sync.js"
 
 
+def test_operations_enqueued_during_an_inflight_flush_are_drained_before_it_settles():
+    script = f"""
+const createTodaySync = require({json.dumps(str(TODAY_SYNC))});
+const values = new Map();
+const patches = [];
+let sequence = 0;
+let releaseFirstPatch;
+const firstPatchBlocked = new Promise(resolve => {{ globalThis.firstPatchStarted = resolve; }});
+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:0, ids:[]}};
+    const operation = JSON.parse(options.body);
+    patches.push(operation);
+    if (patches.length === 1) {{
+      globalThis.firstPatchStarted();
+      await new Promise(resolve => {{ releaseFirstPatch = resolve; }});
+    }}
+    return {{revision:patches.length, ids:patches.map(item => item.item_id)}};
+  }},
+  onRemoteIds: ids => {{ globalThis.ids = ids; }},
+  onStatus: status => {{ globalThis.status = status; }},
+}});
+sync.enqueue('add', 'issue:r:1:');
+(async () => {{
+  const firstFlush = sync.flush();
+  await firstPatchBlocked;
+  sync.enqueue('add', 'issue:r:2:');
+  const sharedFlush = sync.flush();
+  const sharesInflight = firstFlush === sharedFlush;
+  releaseFirstPatch();
+  await sharedFlush;
+  process.stdout.write(JSON.stringify({{
+    sharesInflight, patches, pending:sync.pending(), ids:globalThis.ids, status:globalThis.status,
+  }}));
+}})();
+"""
+    result = json.loads(
+        subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True).stdout
+    )
+
+    assert result == {
+        "sharesInflight": True,
+        "patches": [
+            {"operation_id": "op-1", "action": "add", "item_id": "issue:r:1:", "direction": None},
+            {"operation_id": "op-2", "action": "add", "item_id": "issue:r:2:", "direction": None},
+        ],
+        "pending": [],
+        "ids": ["issue:r:1:", "issue:r:2:"],
+        "status": "saved",
+    }
+
+
+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-v50" in source
+    assert "BASE + 'static/today-sync.js'" in source
+
+
 def test_local_operations_replay_once_then_adopt_server_order():
     script = f"""
 const createTodaySync = require({json.dumps(str(TODAY_SYNC))});
@@ -212,6 +273,49 @@ sync.enqueue('add', '6');
     }
 
 
+def test_server_limit_conflict_drops_only_the_rejected_operation_and_drains_later_edits():
+    script = f"""
+const createTodaySync = require({json.dumps(str(TODAY_SYNC))});
+const values = new Map();
+const patches = [];
+let sequence = 0;
+const statuses = [];
+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']}};
+  }},
+  onRemoteIds: ids => {{ globalThis.ids = ids; }},
+  onStatus: status => statuses.push(status),
+}});
+sync.enqueue('add', '6');
+sync.enqueue('remove', '5');
+(async()=>{{
+  const result = await sync.flush();
+  process.stdout.write(JSON.stringify({{result, patches, pending:sync.pending(), ids:globalThis.ids, statuses}}));
+}})();
+"""
+    result = json.loads(
+        subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True).stdout
+    )
+
+    assert result == {
+        "result": False,
+        "patches": [
+            {"operation_id": "op-1", "action": "add", "item_id": "6", "direction": None},
+            {"operation_id": "op-2", "action": "remove", "item_id": "5", "direction": None},
+        ],
+        "pending": [],
+        "ids": ["1", "2", "3", "4"],
+        "statuses": ["pending", "pending", "full"],
+    }
+
+
 def test_tabs_ignore_an_older_response_after_a_newer_revision_is_broadcast():
     script = f"""
 const createTodaySync = require({json.dumps(str(TODAY_SYNC))});