From d6aaca81278e44cc5f2be3aeab737794645db07f Mon Sep 17 00:00:00 2001 From: slhx agent Date: Mon, 13 Jul 2026 16:58:51 +0200 Subject: [PATCH] feat(kanban): deterministically rebase snapshot req: sync/011 --- PLAN.md | 4 +-- examples/kanban/static/sync.js | 35 ++++++++++++++++++++++-- examples/kanban/tests/browser_e2e.rs | 41 ++++++++++++++++++++++------ 3 files changed, 67 insertions(+), 13 deletions(-) diff --git a/PLAN.md b/PLAN.md index c5f6599..db4b750 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; the client loads a versioned canonical snapshot, exposes rebase-required state, and preserves the uploaded local command in IndexedDB rather than silently deleting it. Broader conflict decisions, actual rebase resolution, 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. Missing/divergent canonical state remains explicitly conflicted and keeps local intent. Broader conflict decisions, multi-tab leadership, backpressure, 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_loads_snapshot_and_preserves_pending_intent -- --exact` proves retained-history gap detection, typed snapshot fallback, versioned canonical snapshot state, visible rebase-required status, and preservation of the pending IndexedDB command. The completed slice proof must additionally cover rebase resolution, 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, and canonical board convergence. The completed slice proof must additionally cover broader partial reject/conflict decisions, two tabs, 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 2f1d6c0..490354c 100644 --- a/examples/kanban/static/sync.js +++ b/examples/kanban/static/sync.js @@ -54,6 +54,24 @@ async function removeAcknowledged(database, commandId) { await done; } +function decideRebase(snapshot, command) { + const canonical = snapshot.cards.find((card) => String(card.id) === command.cardId); + if (!canonical) return { kind: "conflicted", reason: "card-missing", canonicalColumn: "missing" }; + if (command.kind === "reorder_card" && canonical.column === "done") { + return { kind: "converged", reason: "intent-already-canonical", canonicalColumn: canonical.column }; + } + return { kind: "conflicted", reason: "canonical-state-diverged", canonicalColumn: canonical.column }; +} + +async function commitConvergedRebase(database, snapshot, command) { + const transaction = database.transaction([COMMANDS, "meta"], "readwrite"); + const done = transactionDone(transaction); + transaction.objectStore("meta").put(snapshot, "canonicalSnapshot"); + transaction.objectStore("meta").put(snapshot.serverSequence, "acknowledgementCursor"); + transaction.objectStore(COMMANDS).delete(command.id); + await done; +} + function setPhase(phase, message) { root.setAttribute("data-sync-phase", phase); root.querySelector('[role="status"]').textContent = message; @@ -146,13 +164,24 @@ 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); 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)); root.setAttribute("data-sync-rebase-pending-count", String(queued.length)); - root.setAttribute("data-sync-canonical-column", snapshot.cards.find((card) => String(card.id) === command.cardId)?.column || "missing"); - setPhase("rebase-required", `History is unavailable; loaded snapshot ${snapshot.serverSequence} and preserved ${queued.length} pending command for rebase.`); - root.dispatchEvent(new CustomEvent("kanban:sync-rebase-required", { detail: { snapshot, pending: queued } })); + 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") { + await commitConvergedRebase(database, snapshot, command); + 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.`); + root.dispatchEvent(new CustomEvent("kanban:sync-rebased", { detail: { snapshot, command, decision } })); + } else { + 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 } })); + } source.close(); }); } catch (error) { diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index 3b7add7..80f4190 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/examples/kanban/tests/browser_e2e.rs @@ -495,8 +495,8 @@ async fn exhausted_offline_retries_keep_command_until_later_reconnect() -> WebDr } #[tokio::test] -async fn missing_history_loads_snapshot_and_preserves_pending_intent() -> WebDriverResult<()> { - // test req: sync/007 req: sync/020 +async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDriverResult<()> { + // test req: sync/007 req: sync/011 req: sync/020 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")); @@ -564,31 +564,56 @@ async fn missing_history_loads_snapshot_and_preserves_pending_intent() -> WebDri driver.goto(&format!("http://{app_addr}/sync-demo")).await?; wait_until( &driver, - "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'rebase-required'", + "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'rebased'", ) .await?; let fallback = driver .execute( - "const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), uploadSequence: root.getAttribute('data-sync-upload-sequence'), snapshotSequence: root.getAttribute('data-sync-snapshot-sequence'), snapshotSchema: root.getAttribute('data-sync-snapshot-schema'), snapshotCards: root.getAttribute('data-sync-snapshot-card-count'), pending: root.getAttribute('data-sync-pending-count'), rebasePending: root.getAttribute('data-sync-rebase-pending-count'), canonicalColumn: root.getAttribute('data-sync-canonical-column'), status: root.querySelector('[role=status]').textContent, error: root.getAttribute('data-sync-error') }", + "const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), uploadSequence: root.getAttribute('data-sync-upload-sequence'), ackSequence: root.getAttribute('data-sync-ack-sequence'), snapshotSequence: root.getAttribute('data-sync-snapshot-sequence'), snapshotSchema: root.getAttribute('data-sync-snapshot-schema'), snapshotCards: root.getAttribute('data-sync-snapshot-card-count'), pending: root.getAttribute('data-sync-pending-count'), rebasePending: root.getAttribute('data-sync-rebase-pending-count'), decision: root.getAttribute('data-sync-rebase-decision'), reason: root.getAttribute('data-sync-rebase-reason'), canonicalColumn: root.getAttribute('data-sync-canonical-column'), status: root.querySelector('[role=status]').textContent, error: root.getAttribute('data-sync-error') }", Vec::new(), ) .await? .json() .clone(); - assert_eq!(fallback["phase"], "rebase-required"); + assert_eq!(fallback["phase"], "rebased"); assert_eq!(fallback["uploadSequence"], "2"); + assert_eq!(fallback["ackSequence"], "2"); assert_eq!(fallback["snapshotSequence"], "2"); assert_eq!(fallback["snapshotSchema"], "1"); assert_eq!(fallback["snapshotCards"], "3"); - assert_eq!(fallback["pending"], "1"); + assert_eq!(fallback["pending"], "0"); assert_eq!(fallback["rebasePending"], "1"); + assert_eq!(fallback["decision"], "converged"); + assert_eq!(fallback["reason"], "intent-already-canonical"); assert_eq!(fallback["canonicalColumn"], "done"); assert_eq!( fallback["status"], - "History is unavailable; loaded snapshot 2 and preserved 1 pending command for rebase." + "Canonical snapshot 2 already satisfies history:2; committed and removed the pending command." ); assert!(fallback["error"].is_null()); - assert_eq!(command_count(&driver).await?, 1); + assert_eq!(command_count(&driver).await?, 0); + let committed = driver + .execute_async( + r#" + const done = arguments[arguments.length - 1]; + const open = indexedDB.open('hemx-kanban-v1', 1); + open.onsuccess = () => { + const tx = open.result.transaction('meta', 'readonly'); + const meta = tx.objectStore('meta'); + const cursor = meta.get('acknowledgementCursor'); + const snapshot = meta.get('canonicalSnapshot'); + tx.oncomplete = () => done({ cursor: cursor.result, snapshot: snapshot.result }); + tx.onabort = () => done({ error: tx.error && tx.error.name }); + }; + "#, + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(committed["cursor"], 2); + assert_eq!(committed["snapshot"]["schemaVersion"], 1); + assert_eq!(committed["snapshot"]["serverSequence"], 2); driver.goto(&format!("http://{app_addr}/")).await?; let canonical = driver