From a17de85b4f77669cc72d726414e56b160a1715f2 Mon Sep 17 00:00:00 2001 From: slhx agent Date: Mon, 13 Jul 2026 18:06:47 +0200 Subject: [PATCH] feat(kanban): migrate queued commands transactionally req: sync/014 --- PLAN.md | 4 +- examples/kanban/static/command-log.js | 38 ++++- examples/kanban/static/sync.js | 46 ++++-- examples/kanban/tests/browser_e2e.rs | 192 +++++++++++++++++++++++--- hemx-wasm/tests/browser.rs | 9 +- 5 files changed, 251 insertions(+), 38 deletions(-) diff --git a/PLAN.md b/PLAN.md index 786defc..b3d79fa 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. Two same-origin tabs coordinate an app-owned expiring IndexedDB lease 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. Each activation serializes uploads with one in flight, processes at most two acknowledged commands, exposes the retained durable count when backpressured, and resumes the next bounded run only through the visible retry action. A mixed queue commits and removes its accepted prefix exactly once, then stops on the first permanent rejection with the typed server cause visible, the rejected command plus untouched suffix durable, and blind retry disabled. Broader conflict decisions, upgrade mid-queue, 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 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. Each activation serializes uploads with one in flight, processes at most two acknowledged commands, exposes the retained durable count when backpressured, and resumes the next bounded run only through the visible retry action. A mixed queue commits and removes its accepted prefix exactly once, then stops on the first permanent rejection with the typed server cause visible, the rejected command plus untouched suffix durable, and blind retry disabled. The IndexedDB v1-to-v2 command migration transactionally adds the explicit target column, records a typed migration receipt, preserves causal order and interaction intent through an interrupted upload, and later drains in the original order. Broader conflict decisions 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. `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. `cargo test -p hemx-kanban-example --test browser_e2e upload_backpressure_keeps_pending_work_visible_and_recoverable -- --exact` proves one in-flight upload, a two-acknowledgement activation limit, one retained durable command with visible recovery state, and explicit retry draining the final command without loss. `cargo test -p hemx-kanban-example --test browser_e2e mixed_queue_removes_accepted_prefix_and_retains_rejected_tail -- --exact` proves an accepted prefix is canonically applied and removed once before a permanent rejection stops processing, exposes its typed HTTP/server cause, disables blind retry, and leaves both the rejected command and untouched suffix durable. The completed slice proof must additionally cover broader conflict decisions, 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. `cargo test -p hemx-kanban-example --test browser_e2e upload_backpressure_keeps_pending_work_visible_and_recoverable -- --exact` proves one in-flight upload, a two-acknowledgement activation limit, one retained durable command with visible recovery state, and explicit retry draining the final command without loss. `cargo test -p hemx-kanban-example --test browser_e2e mixed_queue_removes_accepted_prefix_and_retains_rejected_tail -- --exact` proves an accepted prefix is canonically applied and removed once before a permanent rejection stops processing, exposes its typed HTTP/server cause, disables blind retry, and leaves both the rejected command and untouched suffix durable. `cargo test -p hemx-kanban-example --test browser_e2e schema_upgrade_preserves_queued_order_and_local_intent -- --exact` proves a three-command v1 queue migrates atomically to the explicit-target v2 schema, remains byte-for-intent ordered after interrupted upload, then receives canonical sequences 1–3 in original order and drains without loss. The completed slice proof must additionally cover broader conflict decisions and multi-user isolation. ## Slice 5 — local-first multiplayer Kanban milestone diff --git a/examples/kanban/static/command-log.js b/examples/kanban/static/command-log.js index 4cbfa20..3dd7b01 100644 --- a/examples/kanban/static/command-log.js +++ b/examples/kanban/static/command-log.js @@ -1,7 +1,10 @@ const DATABASE = "hemx-kanban-v1"; +const DATABASE_VERSION = 2; const COMMANDS = "commands"; const META = "meta"; -const COMMAND_SCHEMA = 1; +const COMMAND_SCHEMA = 2; +const LEGACY_COMMAND_SCHEMA = 1; +const MIGRATION_KEY = "commandSchemaMigration"; const EXPORT_SCHEMA = 1; const MAX_REPLAY_COMMANDS = 64; const REPLAY_BUDGET_MS = 100; @@ -23,13 +26,31 @@ function completed(transaction) { }); } +function migrateCommandLog(request, oldVersion) { + const database = request.result; + if (!database.objectStoreNames.contains(COMMANDS)) database.createObjectStore(COMMANDS, { keyPath: "id" }); + if (!database.objectStoreNames.contains(META)) database.createObjectStore(META); + if (oldVersion === 0 || oldVersion >= DATABASE_VERSION) return; + const transaction = request.transaction; + const commands = transaction.objectStore(COMMANDS); + const meta = transaction.objectStore(META); + const all = commands.getAll(); + all.addEventListener("success", () => { + const legacy = all.result; + if (legacy.some((command) => command.schemaVersion !== LEGACY_COMMAND_SCHEMA)) { + transaction.abort(); + return; + } + for (const command of legacy) { + commands.put({ ...command, schemaVersion: COMMAND_SCHEMA, targetColumn: "done" }); + } + meta.put({ from: LEGACY_COMMAND_SCHEMA, to: COMMAND_SCHEMA, migrated: legacy.length }, MIGRATION_KEY); + }, { once: true }); +} + function openCommandLog() { - const request = indexedDB.open(DATABASE, 1); - request.addEventListener("upgradeneeded", () => { - const database = request.result; - if (!database.objectStoreNames.contains(COMMANDS)) database.createObjectStore(COMMANDS, { keyPath: "id" }); - if (!database.objectStoreNames.contains(META)) database.createObjectStore(META); - }); + const request = indexedDB.open(DATABASE, DATABASE_VERSION); + request.addEventListener("upgradeneeded", (event) => migrateCommandLog(request, event.oldVersion)); return result(request); } @@ -86,6 +107,7 @@ async function appendReorder(database, wire) { causal, kind: "reorder_card", cardId: String(wire[2] || "1"), + targetColumn: "done", eventKind: String(wire[1] || "click"), key: wire[4] ? String(wire[4]) : null, }; @@ -145,6 +167,7 @@ function validate(command) { if (command.id !== `${command.actor}:${command.causal}`) invalidCommand(command, "id"); if (command.kind !== "reorder_card") invalidCommand(command, "kind"); if (typeof command.cardId !== "string" || !command.cardId) invalidCommand(command, "cardId"); + if (command.targetColumn !== "done") invalidCommand(command, "targetColumn"); if (typeof command.eventKind !== "string" || !command.eventKind) invalidCommand(command, "eventKind"); if (command.key !== null && typeof command.key !== "string") invalidCommand(command, "key"); return command; @@ -291,6 +314,7 @@ async function start() { actor: command.actor, session: command.session, causal: command.causal, + targetColumn: command.targetColumn, }, })); try { diff --git a/examples/kanban/static/sync.js b/examples/kanban/static/sync.js index 6826df5..ac2ca45 100644 --- a/examples/kanban/static/sync.js +++ b/examples/kanban/static/sync.js @@ -1,5 +1,9 @@ const DATABASE = "hemx-kanban-v1"; +const DATABASE_VERSION = 2; const COMMANDS = "commands"; +const COMMAND_SCHEMA = 2; +const LEGACY_COMMAND_SCHEMA = 1; +const MIGRATION_KEY = "commandSchemaMigration"; const MAX_ATTEMPTS = 3; const BACKOFF_MS = [25, 50]; const root = document.querySelector("[data-kanban-sync]"); @@ -43,13 +47,31 @@ function transactionDone(transaction) { }); } +function migrateCommandLog(request, oldVersion) { + const database = request.result; + if (!database.objectStoreNames.contains(COMMANDS)) database.createObjectStore(COMMANDS, { keyPath: "id" }); + if (!database.objectStoreNames.contains("meta")) database.createObjectStore("meta"); + if (oldVersion === 0 || oldVersion >= DATABASE_VERSION) return; + const transaction = request.transaction; + const commands = transaction.objectStore(COMMANDS); + const meta = transaction.objectStore("meta"); + const all = commands.getAll(); + all.addEventListener("success", () => { + const legacy = all.result; + if (legacy.some((command) => command.schemaVersion !== LEGACY_COMMAND_SCHEMA)) { + transaction.abort(); + return; + } + for (const command of legacy) { + commands.put({ ...command, schemaVersion: COMMAND_SCHEMA, targetColumn: "done" }); + } + meta.put({ from: LEGACY_COMMAND_SCHEMA, to: COMMAND_SCHEMA, migrated: legacy.length }, MIGRATION_KEY); + }, { once: true }); +} + async function openLog() { - const request = indexedDB.open(DATABASE, 1); - request.addEventListener("upgradeneeded", () => { - const database = request.result; - if (!database.objectStoreNames.contains(COMMANDS)) database.createObjectStore(COMMANDS, { keyPath: "id" }); - if (!database.objectStoreNames.contains("meta")) database.createObjectStore("meta"); - }); + const request = indexedDB.open(DATABASE, DATABASE_VERSION); + request.addEventListener("upgradeneeded", (event) => migrateCommandLog(request, event.oldVersion)); return requestResult(request); } @@ -124,7 +146,7 @@ function setPhase(phase, message) { } function validatePending(command) { - if (!command || command.schemaVersion !== 1 || command.kind !== "reorder_card" || typeof command.id !== "string" || !command.id || typeof command.cardId !== "string" || !command.cardId) { + if (!command || command.schemaVersion !== COMMAND_SCHEMA || command.kind !== "reorder_card" || typeof command.id !== "string" || !command.id || typeof command.cardId !== "string" || !command.cardId || command.targetColumn !== "done") { throw new Error("invalid pending command"); } return command; @@ -155,7 +177,7 @@ async function upload(command) { root.setAttribute("data-sync-attempts", String(attempt)); setPhase(attempt === 1 ? "uploading" : "retrying", `Uploading ${command.id} (attempt ${attempt} of ${MAX_ATTEMPTS}).`); try { - const query = new URLSearchParams({ command_id: command.id, card_id: command.cardId }); + const query = new URLSearchParams({ command_id: command.id, card_id: command.cardId, column: command.targetColumn }); const response = await fetch(`/sync/commands?${query}`, { method: "POST" }); if (response.status === 503 && attempt < MAX_ATTEMPTS) { const base = BACKOFF_MS[attempt - 1]; @@ -342,6 +364,14 @@ async function start() { uploadLimit = Number.parseInt(root.getAttribute("data-sync-upload-limit"), 10); if (!Number.isSafeInteger(uploadLimit) || uploadLimit < 1) throw new Error("data-sync-upload-limit must be a positive integer"); database = await openLog(); + const migration = await requestResult(database.transaction("meta", "readonly").objectStore("meta").get(MIGRATION_KEY)); + root.setAttribute("data-sync-database-version", String(database.version)); + root.setAttribute("data-sync-command-schema", String(COMMAND_SCHEMA)); + if (migration) { + root.setAttribute("data-sync-migration-from", String(migration.from)); + root.setAttribute("data-sync-migration-to", String(migration.to)); + root.setAttribute("data-sync-migrated-count", String(migration.migrated)); + } const commands = await pendingCommands(database); root.setAttribute("data-sync-uploaded-this-run", "0"); root.setAttribute("data-sync-uploaded-total", "0"); diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index e9b6e82..fbb3f03 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/examples/kanban/tests/browser_e2e.rs @@ -232,7 +232,7 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack( .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onupgradeneeded = () => { const database = open.result; if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' }); @@ -297,7 +297,7 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack( .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onsuccess = () => { const request = open.result.transaction('commands', 'readonly').objectStore('commands').count(); request.onsuccess = () => done(request.result); @@ -315,12 +315,12 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack( .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onsuccess = () => { const tx = open.result.transaction('commands', 'readwrite'); tx.objectStore('commands').add({ - id: 'sync-actor:1', schemaVersion: 1, actor: 'sync-actor', session: 'sync-session', - causal: 2, kind: 'reorder_card', cardId: '2', eventKind: 'click', key: null, + id: 'sync-actor:1', schemaVersion: 2, actor: 'sync-actor', session: 'sync-session', + causal: 2, kind: 'reorder_card', cardId: '2', targetColumn: 'done', eventKind: 'click', key: null, }); tx.oncomplete = () => done({ seeded: true }); tx.onabort = () => done({ error: tx.error && tx.error.name }); @@ -372,6 +372,164 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack( result.and(quit) } +#[tokio::test] +async fn schema_upgrade_preserves_queued_order_and_local_intent() -> WebDriverResult<()> { + // test req: sync/004 req: sync/010 req: sync/014 + 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; + database.createObjectStore('commands', { keyPath: 'id' }); + database.createObjectStore('meta'); + }; + open.onsuccess = () => { + const tx = open.result.transaction(['commands', 'meta'], 'readwrite'); + const commands = tx.objectStore('commands'); + for (const [causal, cardId, eventKind, key] of [[1, '1', 'click', null], [2, '2', 'keydown', 'Enter'], [3, '3', 'click', null]]) { + commands.add({ + id: `upgrade:${causal}`, schemaVersion: 1, actor: 'upgrade', session: 'legacy-session', + causal, kind: 'reorder_card', cardId, eventKind, key, + }); + } + tx.objectStore('meta').put(3, 'causal'); + tx.oncomplete = () => done({ version: open.result.version }); + tx.onabort = () => done({ error: tx.error && tx.error.name }); + }; + "#, + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(seeded["version"], 1, "failed to seed legacy queue: {seeded}"); + + 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 migrated = 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 database = open.result; + const tx = database.transaction(['commands', 'meta'], 'readonly'); + const commands = tx.objectStore('commands').getAll(); + const receipt = tx.objectStore('meta').get('commandSchemaMigration'); + tx.oncomplete = () => done({ + databaseVersion: database.version, + commandSchema: root.getAttribute('data-sync-command-schema'), + migrationFrom: root.getAttribute('data-sync-migration-from'), + migrationTo: root.getAttribute('data-sync-migration-to'), + migratedCount: root.getAttribute('data-sync-migrated-count'), + pending: root.getAttribute('data-sync-pending-count'), + receipt: receipt.result, + commands: commands.result.sort((a, b) => a.causal - b.causal).map(({ id, schemaVersion, cardId, targetColumn, eventKind, key }) => ({ id, schemaVersion, cardId, targetColumn, eventKind, key })), + }); + tx.onabort = () => done({ error: tx.error && tx.error.name }); + }; + "#, + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(migrated["databaseVersion"], 2); + assert_eq!(migrated["commandSchema"], "2"); + assert_eq!(migrated["migrationFrom"], "1"); + assert_eq!(migrated["migrationTo"], "2"); + assert_eq!(migrated["migratedCount"], "3"); + assert_eq!(migrated["pending"], "3"); + assert_eq!( + migrated["receipt"], + serde_json::json!({ "from": 1, "to": 2, "migrated": 3 }) + ); + assert_eq!( + migrated["commands"], + serde_json::json!([ + { "id": "upgrade:1", "schemaVersion": 2, "cardId": "1", "targetColumn": "done", "eventKind": "click", "key": null }, + { "id": "upgrade:2", "schemaVersion": 2, "cardId": "2", "targetColumn": "done", "eventKind": "keydown", "key": "Enter" }, + { "id": "upgrade:3", "schemaVersion": 2, "cardId": "3", "targetColumn": "done", "eventKind": "click", "key": null } + ]) + ); + + 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') === 'backpressured' && root?.getAttribute('data-sync-pending-count') === '1'", + ) + .await?; + 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') === 'acknowledged' && root?.getAttribute('data-sync-pending-count') === '0'", + ) + .await?; + + driver + .execute( + r#" + window.__upgradeReplay = []; + const source = new EventSource('/sync/acknowledgements?after=0'); + source.addEventListener('acknowledgement', (event) => { + window.__upgradeReplay.push({ id: event.lastEventId, body: JSON.parse(event.data) }); + if (window.__upgradeReplay.length === 3) source.close(); + }); + return true; + "#, + Vec::new(), + ) + .await?; + wait_until(&driver, "return window.__upgradeReplay.length === 3").await?; + let replay = driver + .execute("return window.__upgradeReplay", Vec::new()) + .await? + .json() + .clone(); + assert_eq!( + replay, + serde_json::json!([ + { "id": "1", "body": { "commandId": "upgrade:1", "cardId": 1, "canonicalColumn": "done", "serverSequence": 1, "status": "accepted" } }, + { "id": "2", "body": { "commandId": "upgrade:2", "cardId": 2, "canonicalColumn": "done", "serverSequence": 2, "status": "accepted" } }, + { "id": "3", "body": { "commandId": "upgrade:3", "cardId": 3, "canonicalColumn": "done", "serverSequence": 3, "status": "accepted" } } + ]) + ); + assert_eq!(command_count(&driver).await?, 0); + Ok(()) + } + .await; + let quit = driver.quit().await; + result.and(quit) +} + #[tokio::test] async fn mixed_queue_removes_accepted_prefix_and_retains_rejected_tail() -> WebDriverResult<()> { // test req: sync/009 req: sync/010 @@ -398,7 +556,7 @@ async fn mixed_queue_removes_accepted_prefix_and_retains_rejected_tail() -> WebD .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onupgradeneeded = () => { const database = open.result; if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' }); @@ -460,7 +618,7 @@ async fn mixed_queue_removes_accepted_prefix_and_retains_rejected_tail() -> WebD .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onsuccess = () => { const request = open.result.transaction('commands', 'readonly').objectStore('commands').getAll(); request.onsuccess = () => done(request.result.sort((a, b) => a.causal - b.causal).map(({ id, cardId }) => ({ id, cardId }))); @@ -526,7 +684,7 @@ async fn upload_backpressure_keeps_pending_work_visible_and_recoverable() -> Web .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onupgradeneeded = () => { const database = open.result; if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' }); @@ -637,7 +795,7 @@ async fn two_tabs_coordinate_single_uploader_and_takeover_without_duplicate_appl .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onupgradeneeded = () => { const database = open.result; if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' }); @@ -808,7 +966,7 @@ async fn exhausted_offline_retries_keep_command_until_later_reconnect() -> WebDr .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onupgradeneeded = () => { const database = open.result; if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' }); @@ -949,7 +1107,7 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onupgradeneeded = () => { const database = open.result; if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' }); @@ -1007,7 +1165,7 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onsuccess = () => { const tx = open.result.transaction('meta', 'readonly'); const meta = tx.objectStore('meta'); @@ -1066,12 +1224,12 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onsuccess = () => { const tx = open.result.transaction('commands', 'readwrite'); tx.objectStore('commands').add({ - id: 'history:3', schemaVersion: 1, actor: 'history', session: 'history-session', - causal: 3, kind: 'reorder_card', cardId: '2', eventKind: 'click', key: null, + id: 'history:3', schemaVersion: 2, actor: 'history', session: 'history-session', + causal: 3, kind: 'reorder_card', cardId: '2', targetColumn: 'done', eventKind: 'click', key: null, }); tx.oncomplete = () => done({ seeded: true }); tx.onabort = () => done({ error: tx.error && tx.error.name }); @@ -1116,7 +1274,7 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onsuccess = () => { const tx = open.result.transaction(['commands', 'meta'], 'readonly'); const command = tx.objectStore('commands').get('history:3'); @@ -1287,7 +1445,7 @@ async fn command_count(driver: &WebDriver) -> WebDriverResult { .execute_async( r#" const done = arguments[arguments.length - 1]; - const open = indexedDB.open('hemx-kanban-v1', 1); + const open = indexedDB.open('hemx-kanban-v1'); open.onsuccess = () => { const request = open.result.transaction('commands', 'readonly').objectStore('commands').count(); request.onsuccess = () => done(request.result); diff --git a/hemx-wasm/tests/browser.rs b/hemx-wasm/tests/browser.rs index fd53e4d..604d30a 100644 --- a/hemx-wasm/tests/browser.rs +++ b/hemx-wasm/tests/browser.rs @@ -256,7 +256,8 @@ async fn kanban_command_persists_before_projection_and_restores_after_reload() - .json() .clone(); assert_eq!(persisted["persisted"]["order"], "1|2"); - assert_eq!(persisted["persisted"]["detail"]["schemaVersion"], 1); + assert_eq!(persisted["persisted"]["detail"]["schemaVersion"], 2); + assert_eq!(persisted["persisted"]["detail"]["targetColumn"], "done"); assert_eq!(persisted["persisted"]["detail"]["causal"], 1); let actor = persisted["persisted"]["detail"]["actor"] .as_str() @@ -307,12 +308,12 @@ async fn kanban_command_persists_before_projection_and_restores_after_reload() - .execute( r#" window.__futureCommandStored = false; - const request = indexedDB.open('hemx-kanban-v1', 1); + const request = indexedDB.open('hemx-kanban-v1'); request.onsuccess = () => { const tx = request.result.transaction('commands', 'readwrite'); tx.objectStore('commands').add({ - id: 'future:2', schemaVersion: 2, actor: 'future', session: 'future', - causal: 2, kind: 'reorder_card', cardId: '2', eventKind: 'click', key: null, + id: 'future:2', schemaVersion: 3, actor: 'future', session: 'future', + causal: 2, kind: 'reorder_card', cardId: '2', targetColumn: 'done', eventKind: 'click', key: null, }); tx.oncomplete = () => { window.__futureCommandStored = true; }; };