test(sync): prove deterministic reconciliation

req: sync/022

examples/kanban/static/sync.js

@@ -114,6 +114,41 @@ function decideRebase(snapshot, command) {
   return { kind: "conflicted", reason: "canonical-state-diverged", canonicalColumn: canonical.column };
 }
 
+// The built-in policy is deliberately a named module export: applications that
+// need custom merge or CRDT semantics must import and wire a different policy.
+export function reconcileServerAuthoritative(snapshot, commandSequence, serverResults) {
+  if (!snapshot || !Array.isArray(snapshot.cards) || !Number.isSafeInteger(snapshot.serverSequence)) {
+    throw new TypeError("reconciliation snapshot is invalid");
+  }
+  if (!Array.isArray(commandSequence) || !Array.isArray(serverResults)) {
+    throw new TypeError("reconciliation commands and server results must be arrays");
+  }
+  const resultCursor = serverResults.reduce((cursor, result) => {
+    if (!result || !Number.isSafeInteger(result.serverSequence)) {
+      throw new TypeError("reconciliation server result is invalid");
+    }
+    return Math.max(cursor, result.serverSequence);
+  }, 0);
+  if (resultCursor > snapshot.serverSequence) {
+    throw new RangeError("reconciliation server result is newer than the canonical snapshot");
+  }
+  const command = commandSequence[0];
+  const decision = command
+    ? decideRebase(snapshot, command)
+    : { kind: "idle", reason: "no-pending-command", canonicalColumn: "unchanged" };
+  return {
+    model: "server-authoritative-v1",
+    snapshotSequence: snapshot.serverSequence,
+    serverResultCursor: resultCursor,
+    serverResultCount: serverResults.length,
+    commandCount: commandSequence.length,
+    retainedCommandCount: decision.kind === "converged"
+      ? Math.max(0, commandSequence.length - 1)
+      : commandSequence.length,
+    decision,
+  };
+}
+
 async function claimUploaderLease(database) {
   const transaction = database.transaction("meta", "readwrite");
   const done = transactionDone(transaction);
@@ -423,8 +458,15 @@ async function synchronize(command) {
       if (!response.ok) throw new Error(`snapshot failed with ${response.status}`);
       const snapshot = await response.json();
       const queued = await pendingCommands(database);
-      const decision = decideRebase(snapshot, command);
+      const reconciliation = reconcileServerAuthoritative(snapshot, queued, [{
+        status: "snapshot-required",
+        serverSequence: missing.latest,
+      }]);
+      const decision = reconciliation.decision;
       const converged = decision.kind === "converged";
+      root.setAttribute("data-sync-reconciliation-model", reconciliation.model);
+      root.setAttribute("data-sync-reconciliation-result-cursor", String(reconciliation.serverResultCursor));
+      root.setAttribute("data-sync-reconciliation-retained-count", String(reconciliation.retainedCommandCount));
       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));

examples/kanban/templates/sync_shell.heml

@@ -16,6 +16,6 @@
     <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>
+  <script type="module" src="/sync.js"></script>
 </body>
 </html>

examples/kanban/tests/browser_e2e.rs

@@ -2077,6 +2077,90 @@ async fn keep_local_retry_preserves_conflicted_command_and_suffix_order() -> Web
     result.and(quit)
 }
 
+#[tokio::test]
+async fn identical_sync_inputs_reconcile_deterministically() -> WebDriverResult<()> {
+    // test req: sync/022
+    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);
+    let _app = TestProcess::start(app_command, "hemx-kanban", &app_addr, STARTUP_TIMEOUT)
+        .expect("start 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}/sync-demo")).await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.hasAttribute('data-sync-database-version')",
+        )
+        .await?;
+        let proof = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                (async () => {
+                  const acceptedResponse = await fetch('/sync/commands?command_id=deterministic%3A1&card_id=1&column=done', { method: 'POST' });
+                  const accepted = await acceptedResponse.json();
+                  const snapshotResponse = await fetch('/sync/snapshot');
+                  const snapshot = await snapshotResponse.json();
+                  const commands = [{ id: accepted.commandId, cardId: String(accepted.cardId), kind: 'reorder_card' }];
+                  const results = [accepted];
+                  const before = JSON.stringify({ snapshot, commands, results });
+                  const { reconcileServerAuthoritative } = await import('/sync.js');
+                  const first = reconcileServerAuthoritative(snapshot, commands, results);
+                  const second = reconcileServerAuthoritative(
+                    structuredClone(snapshot),
+                    structuredClone(commands),
+                    structuredClone(results),
+                  );
+                  done({
+                    acceptedStatus: acceptedResponse.status,
+                    snapshotStatus: snapshotResponse.status,
+                    first,
+                    second,
+                    inputsUnchanged: before === JSON.stringify({ snapshot, commands, results }),
+                  });
+                })().catch((error) => done({ error: String(error), stack: error?.stack }));
+                "#,
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert!(proof["error"].is_null(), "reconciliation failed: {proof}");
+        assert_eq!(proof["acceptedStatus"], 200);
+        assert_eq!(proof["snapshotStatus"], 200);
+        assert_eq!(proof["first"], proof["second"]);
+        assert_eq!(proof["inputsUnchanged"], true);
+        assert_eq!(proof["first"]["model"], "server-authoritative-v1");
+        assert_eq!(proof["first"]["snapshotSequence"], 1);
+        assert_eq!(proof["first"]["serverResultCursor"], 1);
+        assert_eq!(proof["first"]["serverResultCount"], 1);
+        assert_eq!(proof["first"]["commandCount"], 1);
+        assert_eq!(proof["first"]["retainedCommandCount"], 0);
+        assert_eq!(proof["first"]["decision"]["kind"], "converged");
+        assert_eq!(
+            proof["first"]["decision"]["reason"],
+            "intent-already-canonical"
+        );
+        assert_eq!(proof["first"]["decision"]["canonicalColumn"], "done");
+        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