From 0d475e81c8b7c9aac1815962e928b904597dd2dc Mon Sep 17 00:00:00 2001 From: slhx agent Date: Mon, 13 Jul 2026 17:40:16 +0200 Subject: [PATCH] feat(kanban): coordinate uploader tabs req: sync/018 --- PLAN.md | 4 +- examples/kanban/static/sync.js | 87 ++++++++++++- examples/kanban/tests/browser_e2e.rs | 178 ++++++++++++++++++++++++++- 3 files changed, 263 insertions(+), 6 deletions(-) diff --git a/PLAN.md b/PLAN.md index 404d08f..3b39bb5 100644 --- a/PLAN.md +++ b/PLAN.md @@ -42,11 +42,11 @@ encryption, retention, backup, and deployment policy remain host concerns. ## Slice 4 — authoritative reconnect and convergence - [ ] **User value:** offline and concurrent work reconnects without duplicate mutation, silent loss, stale authorization, or ambiguous conflict. -- **State:** In progress — one app-owned `move_card` server command validates a durable client command id, applies the authoritative canonical column once, returns the same acknowledgement for an identical retry, rejects id reuse with a different payload, assigns one server sequence, and redelivers that canonical acknowledgement after a real EventSource disconnect/reconnect. Canonical acknowledgements and the next sequence are durably stored in a strict versioned JSON envelope using fsync plus atomic replacement; startup refuses malformed/unknown state, rebuilds the canonical board, and preserves idempotency and event replay across a real process restart. A dedicated opt-in sync route reads one pending IndexedDB command, retries transient failures with capped exponential backoff and randomized jitter, exposes online/offline state plus an accessible manual retry after exhaustion, waits for canonical acknowledgement on the reconnecting transport, and only then removes the durable command; later retry converges without a new local mutation. Canonical payload conflicts are not retried and remain durable with a visible reason. If requested history predates retained events, the stream emits a typed snapshot-required event and the client loads a versioned canonical snapshot. One deterministic rebase rule treats `reorder_card` as converged only when the canonical snapshot already places that card in `done`; it then atomically stores the snapshot/cursor and removes the satisfied command. If a later canonical command instead places the same card in `doing`, the rebase is explicitly `conflicted`, retains the local command and last committed snapshot/cursor unchanged, and exposes the divergent canonical column/reason. Broader conflict decisions, multi-tab leadership, backpressure, and auth isolation remain. +- **State:** In progress — one app-owned `move_card` server command validates a durable client command id, applies the authoritative canonical column once, returns the same acknowledgement for an identical retry, rejects id reuse with a different payload, assigns one server sequence, and redelivers that canonical acknowledgement after a real EventSource disconnect/reconnect. Canonical acknowledgements and the next sequence are durably stored in a strict versioned JSON envelope using fsync plus atomic replacement; startup refuses malformed/unknown state, rebuilds the canonical board, and preserves idempotency and event replay across a real process restart. A dedicated opt-in sync route reads one pending IndexedDB command, retries transient failures with capped exponential backoff and randomized jitter, exposes online/offline state plus an accessible manual retry after exhaustion, waits for canonical acknowledgement on the reconnecting transport, and only then removes the durable command; later retry converges without a new local mutation. Canonical payload conflicts are not retried and remain durable with a visible reason. If requested history predates retained events, the stream emits a typed snapshot-required event and the client loads a versioned canonical snapshot. One deterministic rebase rule treats `reorder_card` as converged only when the canonical snapshot already places that card in `done`; it then atomically stores the snapshot/cursor and removes the satisfied command. If a later canonical command instead places the same card in `doing`, the rebase is explicitly `conflicted`, retains the local command and last committed snapshot/cursor unchanged, and exposes the divergent canonical column/reason. Two same-origin tabs coordinate an app-owned expiring IndexedDB lease plus a preferred-owner hint so only one uploads; the standby exposes its role without issuing a request, and after the leader closes it takes over, receives one canonical acknowledgement/sequence, and removes the queue once. Broader conflict decisions, backpressure, upgrade mid-queue, and auth isolation remain. - **Build:** materialize `hemx-sync` over an integration transport with idempotent server command processing, snapshot/change cursor, durable acknowledgements, bounded ordered replay, current auth checks, rejection/conflict results, canonical replacement, reconnect jitter/backoff, multi-tab coordination, and redacted diagnostics. - **Refusals:** no default CRDT, transport in core, cached enqueue-time permission, unbounded queue, or silent last-write-wins policy. - **Requirements:** `sync/001-023`, `operations/001-005`, `security/002-005`, `performance/004-005`. -- **Proof:** `cargo test -p hemx-kanban-example --test browser_e2e idempotent_server_command_is_acknowledged_after_reconnect -- --exact` proves duplicate POST delivery yields one identical canonical acknowledgement/sequence, conflicting id reuse is rejected, EventSource reconnects after a server-closed first stream, the acknowledgement is delivered once with its sequence as event id, and a page reload shows the authoritative card in the canonical column. `cargo test -p hemx-kanban-example --test browser_e2e pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack -- --exact` proves automatic platform-store upload, one explicit transient failure, bounded exponential backoff plus jitter, reconnect acknowledgement, pending-before-ack ordering, acknowledged removal, canonical board convergence, and non-retried 409 rejection remaining durable with a visible reason. `cargo test -p hemx-kanban-example --test browser_e2e canonical_acknowledgement_survives_server_restart -- --exact` proves the versioned store is materialized before success, a real process restart reloads the same idempotent acknowledgement/sequence, EventSource replays it by id, and canonical board state is rebuilt. `cargo test -p hemx-kanban-example --test browser_e2e exhausted_offline_retries_keep_command_until_later_reconnect -- --exact` proves three bounded retries exhaust into visible offline/manual-recovery state while the command remains durable, then a later retry acknowledges/removes it and converges canonically. `cargo test -p hemx-kanban-example --test browser_e2e missing_history_rebase_converges_without_losing_local_intent -- --exact` proves retained-history gap detection, typed/versioned snapshot fallback, deterministic already-canonical convergence, atomic snapshot/cursor commit with acknowledged removal, then a divergent canonical update producing explicit conflict while the local command and prior committed snapshot/cursor remain intact. The completed slice proof must additionally cover broader partial reject/conflict decisions, two tabs, backpressure, upgrade mid-queue, and multi-user isolation. +- **Proof:** `cargo test -p hemx-kanban-example --test browser_e2e idempotent_server_command_is_acknowledged_after_reconnect -- --exact` proves duplicate POST delivery yields one identical canonical acknowledgement/sequence, conflicting id reuse is rejected, EventSource reconnects after a server-closed first stream, the acknowledgement is delivered once with its sequence as event id, and a page reload shows the authoritative card in the canonical column. `cargo test -p hemx-kanban-example --test browser_e2e pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack -- --exact` proves automatic platform-store upload, one explicit transient failure, bounded exponential backoff plus jitter, reconnect acknowledgement, pending-before-ack ordering, acknowledged removal, canonical board convergence, and non-retried 409 rejection remaining durable with a visible reason. `cargo test -p hemx-kanban-example --test browser_e2e canonical_acknowledgement_survives_server_restart -- --exact` proves the versioned store is materialized before success, a real process restart reloads the same idempotent acknowledgement/sequence, EventSource replays it by id, and canonical board state is rebuilt. `cargo test -p hemx-kanban-example --test browser_e2e exhausted_offline_retries_keep_command_until_later_reconnect -- --exact` proves three bounded retries exhaust into visible offline/manual-recovery state while the command remains durable, then a later retry acknowledges/removes it and converges canonically. `cargo test -p hemx-kanban-example --test browser_e2e missing_history_rebase_converges_without_losing_local_intent -- --exact` proves retained-history gap detection, typed/versioned snapshot fallback, deterministic already-canonical convergence, atomic snapshot/cursor commit with acknowledged removal, then a divergent canonical update producing explicit conflict while the local command and prior committed snapshot/cursor remain intact. `cargo test -p hemx-kanban-example --test browser_e2e two_tabs_coordinate_single_uploader_and_takeover_without_duplicate_application -- --exact` proves one retry-exhausted leader/one explicit standby, zero follower upload before takeover, lease takeover after the leader closes, one canonical sequence/event, one queue removal, and one board application. The completed slice proof must additionally cover broader partial reject/conflict decisions, backpressure, upgrade mid-queue, and multi-user isolation. ## Slice 5 — local-first multiplayer Kanban milestone diff --git a/examples/kanban/static/sync.js b/examples/kanban/static/sync.js index 490354c..9802067 100644 --- a/examples/kanban/static/sync.js +++ b/examples/kanban/static/sync.js @@ -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."); } diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index 3e124d5..8367b12 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/examples/kanban/tests/browser_e2e.rs @@ -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");