feat(kanban): retry explicit keep-local choice

req: sync/009

req: sync/010

req: sync/011

req: sync/016

examples/kanban/static/sync.js

@@ -22,6 +22,7 @@ let uploadedTotal = 0;
 let inFlightUploads = 0;
 let maxObservedInFlight = 0;
 let activeConflict;
+let manualRetryCommand;
 let stopped = false;
 
 class UploadError extends Error {
@@ -148,7 +149,7 @@ async function commitConvergedRebase(database, snapshot, command) {
   const done = transactionDone(transaction);
   transaction.objectStore("meta").put(snapshot, "canonicalSnapshot");
   transaction.objectStore("meta").put(snapshot.serverSequence, "acknowledgementCursor");
-  transaction.objectStore(COMMANDS).delete(command.id);
+  transaction.objectStore(COMMANDS).delete(command.queueCommandId || command.id);
   await done;
 }
 
@@ -181,6 +182,28 @@ function setExportAvailable(available) {
 
 function setConflictResolutionAvailable(available) {
   root.querySelector("[data-sync-use-canonical]").disabled = !available;
+  root.querySelector("[data-sync-keep-local]").disabled = !available;
+}
+
+async function keepLocalChange() {
+  if (!activeConflict) return;
+  const { command, snapshot } = activeConflict;
+  const retryCommand = {
+    ...command,
+    id: `${command.id}:keep:${snapshot.serverSequence}`,
+    queueCommandId: command.id,
+    conflictResolution: "keep-local-change",
+    basedOnServerSequence: snapshot.serverSequence,
+  };
+  setConflictResolutionAvailable(false);
+  root.setAttribute("data-sync-conflict-resolution", "keep-local-pending");
+  root.setAttribute("data-sync-resolution-command-id", retryCommand.id);
+  root.setAttribute("data-sync-resolved-command-id", command.id);
+  synchronizing = false;
+  clearTimeout(leaseTimer);
+  await releaseUploaderLease(database);
+  root.setAttribute("data-sync-leader", "false");
+  await synchronize(retryCommand);
 }
 
 async function useCanonicalState() {
@@ -218,6 +241,7 @@ async function exportPendingWork() {
 
 function scheduleManualRetry(command, error) {
   clearTimeout(retryTimer);
+  manualRetryCommand = command;
   root.setAttribute("data-sync-error", error instanceof Error ? error.message : String(error));
   setManualRetryAvailable(true);
   setPhase("offline", "Sync is offline after bounded retries; the durable command remains queued. Retry now when ready.");
@@ -330,7 +354,15 @@ async function synchronize(command) {
       if (canonical.commandId !== command.id) return;
       source.close();
       root.setAttribute("data-sync-pending-before-ack", String((await pendingCommands(database)).length));
-      await removePendingCommand(database, command.id);
+      const queueCommandId = command.queueCommandId || command.id;
+      await removePendingCommand(database, queueCommandId);
+      manualRetryCommand = undefined;
+      if (command.conflictResolution === "keep-local-change") {
+        activeConflict = undefined;
+        setConflictResolutionAvailable(false);
+        root.setAttribute("data-sync-conflict-resolution", "kept-local-change");
+        root.setAttribute("data-sync-resolved-command-id", queueCommandId);
+      }
       uploadsThisRun += 1;
       uploadedTotal += 1;
       root.setAttribute("data-sync-uploaded-this-run", String(uploadsThisRun));
@@ -356,6 +388,7 @@ async function synchronize(command) {
       const snapshot = await response.json();
       const queued = await pendingCommands(database);
       const decision = decideRebase(snapshot, command);
+      const converged = decision.kind === "converged";
       root.setAttribute("data-sync-snapshot-sequence", String(snapshot.serverSequence));
       root.setAttribute("data-sync-snapshot-schema", String(snapshot.schemaVersion));
       root.setAttribute("data-sync-snapshot-card-count", String(snapshot.cards.length));
@@ -363,10 +396,15 @@ async function synchronize(command) {
       root.setAttribute("data-sync-rebase-decision", decision.kind);
       root.setAttribute("data-sync-rebase-reason", decision.reason);
       root.setAttribute("data-sync-canonical-column", decision.canonicalColumn);
-      if (decision.kind === "converged") {
+      if (converged) {
         activeConflict = undefined;
+        manualRetryCommand = undefined;
         setConflictResolutionAvailable(false);
         await commitConvergedRebase(database, snapshot, command);
+        if (command.conflictResolution === "keep-local-change") {
+          root.setAttribute("data-sync-conflict-resolution", "kept-local-change");
+          root.setAttribute("data-sync-resolved-command-id", command.queueCommandId);
+        }
         root.setAttribute("data-sync-pending-count", String((await pendingCommands(database)).length));
         root.setAttribute("data-sync-ack-sequence", String(snapshot.serverSequence));
         setPhase("rebased", `Canonical snapshot ${snapshot.serverSequence} already satisfies ${command.id}; committed and removed the pending command.`);
@@ -382,6 +420,7 @@ async function synchronize(command) {
       clearTimeout(leaseTimer);
       await releaseUploaderLease(database);
       root.setAttribute("data-sync-leader", "false");
+      if (converged) await continuePendingWork();
     });
   } catch (error) {
     synchronizing = false;
@@ -392,7 +431,15 @@ async function synchronize(command) {
       root.setAttribute("data-sync-error-status", String(error.status));
       const remaining = await pendingCommands(database);
       setManualRetryAvailable(false);
-      if (error.kind === "authorization-denial") {
+      if (command.conflictResolution === "keep-local-change") {
+        manualRetryCommand = undefined;
+        setConflictResolutionAvailable(true);
+        root.setAttribute("data-sync-error-kind", error.kind);
+        root.setAttribute("data-sync-error-reason", error.reason);
+        root.setAttribute("data-sync-pending-count", String(remaining.length));
+        root.setAttribute("data-sync-conflict-resolution", "keep-local-rejected");
+        setPhase("resolution-rejected", `Keep-local command ${command.id} was rejected (${error.status}: ${error.reason}); the conflicted command and ${remaining.length - 1} queued suffix command${remaining.length === 2 ? "" : "s"} remain in order.`);
+      } else if (error.kind === "authorization-denial") {
         root.setAttribute("data-sync-error-kind", "authorization-denial");
         root.setAttribute("data-sync-pending-count", "redacted");
         root.setAttribute("data-sync-redacted-pending", "true");
@@ -469,6 +516,10 @@ async function start() {
   }
   const command = validatePending(commands[0]);
   root.addEventListener("click", async (event) => {
+    if (event.target.closest("[data-sync-keep-local]")) {
+      await keepLocalChange();
+      return;
+    }
     if (event.target.closest("[data-sync-use-canonical]")) {
       await useCanonicalState();
       return;
@@ -482,13 +533,15 @@ async function start() {
     root.setAttribute("data-sync-uploaded-this-run", "0");
     setManualRetryAvailable(false);
     const [next] = await pendingCommands(database);
-    if (next) synchronize(validatePending(next)).catch(failPermanently);
+    const retry = manualRetryCommand || (next && validatePending(next));
+    if (retry) synchronize(retry).catch(failPermanently);
   });
   window.addEventListener("online", async () => {
     if (root.getAttribute("data-sync-phase") !== "offline") return;
     setManualRetryAvailable(false);
     const [next] = await pendingCommands(database);
-    if (next) synchronize(validatePending(next)).catch(failPermanently);
+    const retry = manualRetryCommand || (next && validatePending(next));
+    if (retry) synchronize(retry).catch(failPermanently);
   });
   await runLeaseLoop(command);
 }

examples/kanban/templates/sync_shell.heml

@@ -12,6 +12,7 @@
     <button type="button" data-sync-retry>Retry sync now</button>
     <button type="button" data-sync-export disabled>Export this account's queue</button>
     <button type="button" data-sync-use-canonical disabled>Use canonical state and continue</button>
+    <button type="button" data-sync-keep-local disabled>Keep local change and continue</button>
   </section>
   <script +src="self.runtime_src" defer></script>
   <script src="/sync.js" defer></script>

examples/kanban/tests/browser_e2e.rs

@@ -1704,6 +1704,248 @@ async fn missing_history_rebase_and_user_conflict_resolution_preserve_suffix() -
     result.and(quit)
 }
 
+#[tokio::test]
+async fn keep_local_retry_preserves_conflicted_command_and_suffix_order() -> WebDriverResult<()> {
+    // test req: sync/009 req: sync/010 req: sync/011 req: sync/016
+    let app_port = available_port();
+    let app_addr = format!("127.0.0.1:{app_port}");
+    let mut app_command = Command::new(env!("CARGO_BIN_EXE_hemx-kanban-example"));
+    app_command
+        .env("HEMX_KANBAN_ADDR", &app_addr)
+        .env("HEMX_KANBAN_RETAINED_AFTER", "1");
+    let _app = TestProcess::start(app_command, "hemx-kanban", &app_addr, STARTUP_TIMEOUT)
+        .expect("start ready hemx-kanban");
+
+    let webdriver_port = available_port();
+    let webdriver_addr = format!("127.0.0.1:{webdriver_port}");
+    let mut webdriver = Command::new("geckodriver");
+    webdriver.arg("--port").arg(webdriver_port.to_string());
+    let _webdriver = TestProcess::start(webdriver, "geckodriver", &webdriver_addr, STARTUP_TIMEOUT)
+        .expect("start ready geckodriver");
+    let mut caps = DesiredCapabilities::firefox();
+    caps.set_headless()?;
+    let driver = WebDriver::new(&format!("http://{webdriver_addr}"), caps).await?;
+
+    let result = async {
+        driver.goto(&format!("http://{app_addr}/")).await?;
+        let seeded_server = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                const command = (id, card, column = 'done') => fetch(`/sync/commands?command_id=${encodeURIComponent(id)}&card_id=${card}&column=${column}`, { method: 'POST' });
+                (async () => {
+                  const statuses = [];
+                  statuses.push((await command('keep:1', 1)).status);
+                  statuses.push((await command('keep:2', 2)).status);
+                  statuses.push((await command('keep:3', 2)).status);
+                  statuses.push((await command('keep:external', 2, 'doing')).status);
+                  done({ statuses });
+                })().catch((error) => done({ error: String(error) }));
+                "#,
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(seeded_server["statuses"], serde_json::json!([200, 200, 200, 200]));
+
+        let seeded_local = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                const open = indexedDB.open('hemx-kanban-v1', 3);
+                open.onupgradeneeded = () => {
+                  const database = open.result;
+                  const commands = database.createObjectStore('commands', { keyPath: 'id' });
+                  commands.createIndex('byAccountPartition', 'accountPartition');
+                  database.createObjectStore('meta');
+                };
+                open.onsuccess = () => {
+                  const tx = open.result.transaction('commands', 'readwrite');
+                  const commands = tx.objectStore('commands');
+                  commands.add({
+                    id: 'keep:3', schemaVersion: 2, accountPartition: 'demo:demo', actor: 'keep', session: 'keep-session',
+                    causal: 3, kind: 'reorder_card', cardId: '2', targetColumn: 'done', eventKind: 'click', key: null,
+                  });
+                  commands.add({
+                    id: 'keep:4', schemaVersion: 2, accountPartition: 'demo:demo', actor: 'keep', session: 'keep-session',
+                    causal: 4, kind: 'reorder_card', cardId: '1', targetColumn: 'done', eventKind: 'click', key: null,
+                  });
+                  tx.oncomplete = () => done({ seeded: true });
+                  tx.onabort = () => done({ error: tx.error && tx.error.name });
+                };
+                "#,
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(seeded_local["seeded"], true);
+
+        driver.goto(&format!("http://{app_addr}/sync-demo")).await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'conflicted'",
+        )
+        .await?;
+        driver
+            .execute(
+                r#"
+                window.__keepRejections = 0;
+                window.__keepFailures = 0;
+                const originalFetch = window.fetch.bind(window);
+                window.fetch = (input, init = {}) => {
+                  const url = new URL(typeof input === 'string' ? input : input.url, location.href);
+                  if (url.pathname === '/sync/commands' && url.searchParams.get('command_id') === 'keep:3:keep:4') {
+                    if (window.__keepRejections < 1) {
+                      window.__keepRejections += 1;
+                      return Promise.resolve(new Response(JSON.stringify({ kind: 'command-conflict', error: 'injected stale keep-local decision' }), {
+                        status: 409,
+                        headers: { 'content-type': 'application/json' },
+                      }));
+                    }
+                    if (window.__keepFailures < 3) {
+                      window.__keepFailures += 1;
+                      return Promise.resolve(new Response(JSON.stringify({ kind: 'transport-failure', error: 'injected keep-local retry' }), {
+                        status: 503,
+                        headers: { 'content-type': 'application/json' },
+                      }));
+                    }
+                  }
+                  return originalFetch(input, init);
+                };
+                return true;
+                "#,
+                Vec::new(),
+            )
+            .await?;
+        driver
+            .find(By::Css("[data-sync-keep-local]"))
+            .await?
+            .click()
+            .await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'resolution-rejected'",
+        )
+        .await?;
+        let rejected = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                const root = document.querySelector('[data-kanban-sync]');
+                const open = indexedDB.open('hemx-kanban-v1');
+                open.onsuccess = () => {
+                  const request = open.result.transaction('commands', 'readonly').objectStore('commands').getAll();
+                  request.onsuccess = () => done({
+                    resolution: root.getAttribute('data-sync-conflict-resolution'),
+                    resolutionDisabled: root.querySelector('[data-sync-keep-local]').disabled,
+                    pending: root.getAttribute('data-sync-pending-count'),
+                    rejections: window.__keepRejections,
+                    commands: request.result.sort((left, right) => left.causal - right.causal).map(({ id, causal, cardId }) => ({ id, causal, cardId })),
+                  });
+                };
+                "#,
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(rejected["resolution"], "keep-local-rejected");
+        assert_eq!(rejected["resolutionDisabled"], false);
+        assert_eq!(rejected["pending"], "2");
+        assert_eq!(rejected["rejections"], 1);
+        assert_eq!(
+            rejected["commands"],
+            serde_json::json!([
+                { "id": "keep:3", "causal": 3, "cardId": "2" },
+                { "id": "keep:4", "causal": 4, "cardId": "1" }
+            ])
+        );
+
+        driver
+            .find(By::Css("[data-sync-keep-local]"))
+            .await?
+            .click()
+            .await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'offline'",
+        )
+        .await?;
+        let retrying = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                const root = document.querySelector('[data-kanban-sync]');
+                const open = indexedDB.open('hemx-kanban-v1');
+                open.onsuccess = () => {
+                  const request = open.result.transaction('commands', 'readonly').objectStore('commands').getAll();
+                  request.onsuccess = () => done({
+                    phase: root.getAttribute('data-sync-phase'),
+                    resolution: root.getAttribute('data-sync-conflict-resolution'),
+                    resolutionCommand: root.getAttribute('data-sync-resolution-command-id'),
+                    resolvedCommand: root.getAttribute('data-sync-resolved-command-id'),
+                    attempts: root.getAttribute('data-sync-attempts'),
+                    pending: root.getAttribute('data-sync-pending-count'),
+                    manualRetry: root.getAttribute('data-sync-manual-retry'),
+                    failures: window.__keepFailures,
+                    commands: request.result.sort((left, right) => left.causal - right.causal).map(({ id, causal, cardId }) => ({ id, causal, cardId })),
+                  });
+                };
+                "#,
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(retrying["phase"], "offline");
+        assert_eq!(retrying["resolution"], "keep-local-pending");
+        assert_eq!(retrying["resolutionCommand"], "keep:3:keep:4");
+        assert_eq!(retrying["resolvedCommand"], "keep:3");
+        assert_eq!(retrying["attempts"], "3");
+        assert_eq!(retrying["pending"], "2");
+        assert_eq!(retrying["manualRetry"], "available");
+        assert_eq!(retrying["failures"], 3);
+        assert_eq!(
+            retrying["commands"],
+            serde_json::json!([
+                { "id": "keep:3", "causal": 3, "cardId": "2" },
+                { "id": "keep:4", "causal": 4, "cardId": "1" }
+            ])
+        );
+
+        driver.find(By::Css("[data-sync-retry]")).await?.click().await?;
+        wait_until(
+            &driver,
+            "const root = document.querySelector('[data-kanban-sync]'); return root?.getAttribute('data-sync-phase') === 'rebased' && root?.getAttribute('data-sync-pending-count') === '0'",
+        )
+        .await?;
+        let resolved = driver
+            .execute(
+                "const root = document.querySelector('[data-kanban-sync]'); return { resolution: root.getAttribute('data-sync-conflict-resolution'), resolvedCommand: root.getAttribute('data-sync-resolved-command-id'), pending: root.getAttribute('data-sync-pending-count'), ackSequence: root.getAttribute('data-sync-ack-sequence'), canonicalColumn: root.getAttribute('data-sync-canonical-column'), status: root.querySelector('[role=status]').textContent }",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(resolved["resolution"], "kept-local-change");
+        assert_eq!(resolved["resolvedCommand"], "keep:3");
+        assert_eq!(resolved["pending"], "0");
+        assert_eq!(resolved["ackSequence"], "6");
+        assert_eq!(resolved["canonicalColumn"], "done");
+        assert_eq!(
+            resolved["status"],
+            "Canonical snapshot 6 already satisfies keep:4; committed and removed the pending command."
+        );
+        assert_eq!(command_count(&driver).await?, 0);
+        Ok(())
+    }
+    .await;
+    let quit = driver.quit().await;
+    result.and(quit)
+}
+
 #[tokio::test]
 async fn canonical_acknowledgement_survives_server_restart() -> WebDriverResult<()> {
     // test req: sync/001 req: sync/005 req: sync/007 req: sync/008 req: sync/013