feat(kanban): coordinate uploader tabs

req: sync/018
2026-07-13 17:40:16 +02:00
parent e4688bb6df
commit 0d475e81c8
3 changed files with 263 additions and 6 deletions
@@ -3,8 +3,15 @@ const COMMANDS = "commands";
 const MAX_ATTEMPTS = 3;
 const BACKOFF_MS = [25, 50];
 const root = document.querySelector("[data-kanban-sync]");
+const TAB_ID = sessionStorage.getItem("hemx-kanban-sync-tab-id") || crypto.randomUUID();
+const LEASE_MS = 5000;
+const LEASE_POLL_MS = 100;
+const LEASE_KEY = "uploaderLease";
 let database;
 let retryTimer;
+let leaseTimer;
+let synchronizing = false;
+let stopped = false;

 class UploadError extends Error {
  constructor(status, retryable) {
@@ -63,6 +70,38 @@ function decideRebase(snapshot, command) {
  return { kind: "conflicted", reason: "canonical-state-diverged", canonicalColumn: canonical.column };
 }

+async function claimUploaderLease(database) {
+  const transaction = database.transaction("meta", "readwrite");
+  const done = transactionDone(transaction);
+  const meta = transaction.objectStore("meta");
+  const now = Date.now();
+  const current = await requestResult(meta.get(LEASE_KEY));
+  if (current && current.owner !== TAB_ID && current.expiresAt > now) {
+    await done;
+    return { leader: false, owner: current.owner, expiresAt: current.expiresAt };
+  }
+  const lease = { owner: TAB_ID, expiresAt: now + LEASE_MS };
+  meta.put(lease, LEASE_KEY);
+  await done;
+  return { leader: true, ...lease };
+}
+
+async function releaseUploaderLease(database) {
+  const transaction = database.transaction("meta", "readwrite");
+  const done = transactionDone(transaction);
+  const meta = transaction.objectStore("meta");
+  const current = await requestResult(meta.get(LEASE_KEY));
+  if (current?.owner === TAB_ID) meta.delete(LEASE_KEY);
+  await done;
+}
+
+function publishLease(lease) {
+  root.setAttribute("data-sync-tab-id", TAB_ID);
+  root.setAttribute("data-sync-leader", String(lease.leader));
+  root.setAttribute("data-sync-lease-owner", lease.owner || TAB_ID);
+  root.setAttribute("data-sync-lease-expires", String(lease.expiresAt));
+}
+
 async function commitConvergedRebase(database, snapshot, command) {
  const transaction = database.transaction([COMMANDS, "meta"], "readwrite");
  const done = transactionDone(transaction);
@@ -130,6 +169,21 @@ async function upload(command) {
 }

 async function synchronize(command) {
+  if (synchronizing) return;
+  synchronizing = true;
+  const lease = await claimUploaderLease(database);
+  publishLease(lease);
+  if (!lease.leader) {
+    synchronizing = false;
+    setPhase("standby", "Another tab owns sync; waiting for lease takeover.");
+    return;
+  }
+  clearTimeout(leaseTimer);
+  leaseTimer = setTimeout(() => {
+    if (!stopped && root.getAttribute("data-sync-phase") !== "acknowledged") {
+      synchronize(command).catch(failPermanently);
+    }
+  }, LEASE_MS / 2);
  root.removeAttribute("data-sync-error");
  root.removeAttribute("data-sync-manual-retry");
  setOnline(navigator.onLine);
@@ -156,6 +210,10 @@ async function synchronize(command) {
      root.setAttribute("data-sync-canonical-column", canonical.canonicalColumn);
      setPhase("acknowledged", `Command ${command.id} acknowledged in ${canonical.canonicalColumn}.`);
      root.dispatchEvent(new CustomEvent("kanban:sync-acknowledged", { detail: canonical }));
+      synchronizing = false;
+      clearTimeout(leaseTimer);
+      await releaseUploaderLease(database);
+      root.setAttribute("data-sync-leader", "false");
      source.close();
    });
    source.addEventListener("snapshot-required", async (event) => {
@@ -182,15 +240,35 @@ async function synchronize(command) {
        setPhase("conflicted", `Canonical snapshot ${snapshot.serverSequence} conflicts with ${command.id} (${decision.reason}); the pending command remains queued.`);
        root.dispatchEvent(new CustomEvent("kanban:sync-conflicted", { detail: { snapshot, command, decision } }));
      }
+      synchronizing = false;
      source.close();
    });
  } catch (error) {
+    synchronizing = false;
    setOnline(false);
    if (error instanceof UploadError && !error.retryable) throw error;
    scheduleManualRetry(command, error);
  }
 }

+async function runLeaseLoop(command) {
+  if (stopped) return;
+  const phase = root.getAttribute("data-sync-phase");
+  if (phase === "acknowledged" || phase === "rebased" || phase === "conflicted" || phase === "failed") return;
+  if (root.getAttribute("data-sync-leader") === "true") {
+    await synchronize(command);
+    return;
+  }
+  const lease = await claimUploaderLease(database);
+  publishLease(lease);
+  if (lease.leader) {
+    await synchronize(command);
+    return;
+  }
+  setPhase("standby", "Another tab owns sync; waiting for lease takeover.");
+  leaseTimer = setTimeout(() => runLeaseLoop(command).catch(failPermanently), LEASE_POLL_MS);
+}
+
 async function start() {
  if (!root) return;
  database = await openLog();
@@ -208,10 +286,17 @@ async function start() {
  window.addEventListener("online", () => {
    if (root.getAttribute("data-sync-phase") === "offline") synchronize(command).catch(failPermanently);
  });
-  await synchronize(command);
+  await runLeaseLoop(command);
 }

+window.addEventListener("pagehide", () => {
+  stopped = true;
+  clearTimeout(leaseTimer);
+  if (database) releaseUploaderLease(database).catch(() => {});
+});
+
 function failPermanently(error) {
+  synchronizing = false;
  root.setAttribute("data-sync-error", error instanceof Error ? error.message : String(error));
  setPhase("failed", "Sync failed; the durable command remains queued.");
 }
@@ -273,7 +273,7 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
        assert_eq!(proof["phase"], "acknowledged", "sync failed: {proof}");
        assert_eq!(proof["pending"], "0");
        assert_eq!(proof["pendingBeforeAck"], "1");
-        assert_eq!(proof["attempts"], "2");
+        assert_eq!(proof["attempts"], "2", "unexpected retry state: {proof}");
        assert_eq!(proof["maxAttempts"], "3");
        assert_eq!(proof["backoffBase"], "25");
        assert!(
@@ -369,6 +369,178 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
    result.and(quit)
 }

+#[tokio::test]
+async fn two_tabs_coordinate_single_uploader_and_takeover_without_duplicate_application(
+) -> WebDriverResult<()> {
+    // test req: sync/018
+    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_SYNC_FAILURES", "3");
+    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 = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                const open = indexedDB.open('hemx-kanban-v1', 1);
+                open.onupgradeneeded = () => {
+                  const database = open.result;
+                  if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' });
+                  if (!database.objectStoreNames.contains('meta')) database.createObjectStore('meta');
+                };
+                open.onsuccess = () => {
+                  const tx = open.result.transaction('commands', 'readwrite');
+                  tx.objectStore('commands').add({
+                    id: 'tabs:1', schemaVersion: 1, actor: 'tabs', session: 'tabs-session',
+                    causal: 1, kind: 'reorder_card', cardId: '1', 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["seeded"], true);
+
+        driver
+            .execute(
+                "sessionStorage.setItem('hemx-kanban-sync-tab-id', 'leader-seed'); return true;",
+                Vec::new(),
+            )
+            .await?;
+        driver.goto(&format!("http://{app_addr}/sync-demo")).await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'offline'",
+        )
+        .await?;
+        let leader = driver.window().await?;
+        let follower = driver.new_tab().await?;
+        driver.switch_to_window(follower.clone()).await?;
+        driver.goto(&format!("http://{app_addr}/")).await?;
+        driver
+            .execute(
+                "sessionStorage.setItem('hemx-kanban-sync-tab-id', 'follower-seed'); return true;",
+                Vec::new(),
+            )
+            .await?;
+        driver.goto(&format!("http://{app_addr}/sync-demo")).await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'standby'",
+        )
+        .await?;
+        let standby = driver
+            .execute(
+                "const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), leader: root.getAttribute('data-sync-leader'), attempts: root.getAttribute('data-sync-attempts'), owner: root.getAttribute('data-sync-lease-owner'), tab: root.getAttribute('data-sync-tab-id'), status: root.querySelector('[role=status]').textContent }",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(standby["phase"], "standby");
+        assert_eq!(standby["leader"], "false");
+        assert!(standby["attempts"].is_null());
+        assert_ne!(standby["owner"], standby["tab"]);
+        assert_eq!(standby["status"], "Another tab owns sync; waiting for lease takeover.");
+
+        driver.switch_to_window(leader).await?;
+        let first = driver
+            .execute(
+                "const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), leader: root.getAttribute('data-sync-leader'), attempts: root.getAttribute('data-sync-attempts'), pending: root.getAttribute('data-sync-pending-count'), owner: root.getAttribute('data-sync-lease-owner'), tab: root.getAttribute('data-sync-tab-id') }",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(first["phase"], "offline", "leader lost ownership: {first}");
+        assert_eq!(first["leader"], "true");
+        assert_eq!(first["attempts"], "3");
+        assert_eq!(first["pending"], "1");
+        driver.close_window().await?;
+
+        driver.switch_to_window(follower).await?;
+        wait_until(
+            &driver,
+            "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'acknowledged'",
+        )
+        .await?;
+        let takeover = driver
+            .execute(
+                "const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), attempts: root.getAttribute('data-sync-attempts'), pending: root.getAttribute('data-sync-pending-count'), sequence: root.getAttribute('data-sync-ack-sequence'), status: root.querySelector('[role=status]').textContent }",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(takeover["phase"], "acknowledged");
+        assert_eq!(takeover["attempts"], "1");
+        assert_eq!(takeover["pending"], "0");
+        assert_eq!(takeover["sequence"], "1");
+        assert_eq!(takeover["status"], "Command tabs:1 acknowledged in done.");
+        assert_eq!(command_count(&driver).await?, 0);
+
+        driver
+            .execute(
+                r#"
+                window.__tabReplay = [];
+                const source = new EventSource('/sync/acknowledgements?after=0');
+                source.addEventListener('acknowledgement', (event) => {
+                  window.__tabReplay.push({ id: event.lastEventId, body: JSON.parse(event.data) });
+                  setTimeout(() => source.close(), 25);
+                });
+                return true;
+                "#,
+                Vec::new(),
+            )
+            .await?;
+        wait_until(&driver, "return window.__tabReplay.length === 1").await?;
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        let replay = driver
+            .execute("return window.__tabReplay", Vec::new())
+            .await?
+            .json()
+            .clone();
+        assert_eq!(replay.as_array().map(Vec::len), Some(1));
+        assert_eq!(replay[0]["id"], "1");
+        assert_eq!(replay[0]["body"]["commandId"], "tabs:1");
+
+        driver.goto(&format!("http://{app_addr}/")).await?;
+        let canonical = driver
+            .execute(
+                "return [...document.querySelectorAll('section.column')].map((column) => ({ title: column.querySelector('h2').textContent, cards: [...column.querySelectorAll('[data-key]')].map((card) => card.dataset.key) }))",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(canonical[2]["cards"], serde_json::json!(["1", "3"]));
+        Ok(())
+    }
+    .await;
+    let quit = driver.quit().await;
+    result.and(quit)
+}
+
 #[tokio::test]
 async fn exhausted_offline_retries_keep_command_until_later_reconnect() -> WebDriverResult<()> {
    // test req: sync/004 req: sync/010 req: sync/011 req: sync/014 req: sync/016
@@ -582,7 +754,7 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
        assert_eq!(fallback["snapshotSchema"], "1");
        assert_eq!(fallback["snapshotCards"], "3");
        assert_eq!(fallback["pending"], "0");
-        assert_eq!(fallback["rebasePending"], "1");
+        assert_eq!(fallback["rebasePending"], "1", "unexpected rebase state: {fallback}");
        assert_eq!(fallback["decision"], "converged");
        assert_eq!(fallback["reason"], "intent-already-canonical");
        assert_eq!(fallback["canonicalColumn"], "done");
@@ -687,7 +859,7 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
            .await?
            .json()
            .clone();
-        assert_eq!(conflicted["phase"], "conflicted");
+        assert_eq!(conflicted["phase"], "conflicted", "unexpected conflict state: {conflicted}");
        assert_eq!(conflicted["uploadSequence"], "3");
        assert_eq!(conflicted["snapshotSequence"], "4");
        assert_eq!(conflicted["pending"], "1");