From e4688bb6dfb73e749e7001a51acdb1e122ed0075 Mon Sep 17 00:00:00 2001 From: slhx agent Date: Mon, 13 Jul 2026 17:04:10 +0200 Subject: [PATCH] test(kanban): prove divergent rebase conflict req: sync/011 --- PLAN.md | 4 +- examples/kanban/src/main.rs | 57 ++++++++++---- examples/kanban/tests/browser_e2e.rs | 110 +++++++++++++++++++++++++++ 3 files changed, 156 insertions(+), 15 deletions(-) diff --git a/PLAN.md b/PLAN.md index db4b750..404d08f 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. Missing/divergent canonical state remains explicitly conflicted and keeps local intent. 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. 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_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. +- **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. ## Slice 5 — local-first multiplayer Kanban milestone diff --git a/examples/kanban/src/main.rs b/examples/kanban/src/main.rs index a7fb8a4..d465c47 100644 --- a/examples/kanban/src/main.rs +++ b/examples/kanban/src/main.rs @@ -67,13 +67,40 @@ impl CommandId { } } +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "lowercase")] +enum CanonicalColumn { + Todo, + Doing, + Done, +} + +impl CanonicalColumn { + fn parse(value: Option<&String>) -> Result { + match value.map(String::as_str).unwrap_or("done") { + "todo" => Ok(Self::Todo), + "doing" => Ok(Self::Doing), + "done" => Ok(Self::Done), + _ => Err(SyncRejection::BadRequest("invalid column")), + } + } + + fn index(self) -> usize { + match self { + Self::Todo => 0, + Self::Doing => 1, + Self::Done => 2, + } + } +} + #[derive(Clone, Debug, Eq, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] struct SyncAcknowledgement { command_id: String, server_sequence: u64, card_id: u64, - canonical_column: &'static str, + canonical_column: CanonicalColumn, status: &'static str, } @@ -89,7 +116,7 @@ struct SyncSnapshot { #[serde(rename_all = "camelCase")] struct SnapshotCard { id: u64, - column: &'static str, + column: CanonicalColumn, } #[derive(Debug)] @@ -126,6 +153,7 @@ struct PersistedAcknowledgement { command_id: String, server_sequence: u64, card_id: u64, + canonical_column: CanonicalColumn, } impl SyncStore { @@ -154,7 +182,7 @@ impl SyncStore { command_id: stored.command_id, server_sequence: stored.server_sequence, card_id: stored.card_id, - canonical_column: "done", + canonical_column: stored.canonical_column, status: "accepted", }; if acknowledgements @@ -221,6 +249,7 @@ impl From<&SyncAcknowledgement> for PersistedAcknowledgement { command_id: value.command_id.clone(), server_sequence: value.server_sequence, card_id: value.card_id, + canonical_column: value.canonical_column, } } } @@ -325,7 +354,7 @@ async fn main() { .iter_mut() .find(|card| card.id == acknowledgement.card_id) { - card.column = 2; + card.column = acknowledgement.canonical_column.index(); } } let state = Arc::new(AppState { @@ -447,10 +476,11 @@ async fn sync_command( .and_then(|value| value.parse::().ok()) .filter(|value| *value > 0) .ok_or(SyncRejection::BadRequest("invalid card_id"))?; + let canonical_column = CanonicalColumn::parse(params.get("column"))?; let mut sync = state.sync.lock().unwrap(); if let Some(existing) = sync.acknowledgements.get(&command_id) { - if existing.card_id != card_id { + if existing.card_id != card_id || existing.canonical_column != canonical_column { return Err(SyncRejection::Conflict( "command_id was already used for a different payload", )); @@ -479,7 +509,7 @@ async fn sync_command( command_id: command_id.0.clone(), server_sequence: sync.next_sequence, card_id, - canonical_column: "done", + canonical_column, status: "accepted", }; if let Some(store) = &state.sync_store { @@ -488,7 +518,7 @@ async fn sync_command( SyncRejection::Storage })?; } - board.cards[card_index].column = 2; + board.cards[card_index].column = canonical_column.index(); sync.next_sequence += 1; sync.acknowledgements .insert(command_id, acknowledgement.clone()); @@ -504,7 +534,7 @@ async fn sync_snapshot(State(state): State>) -> Json .iter() .map(|card| SnapshotCard { id: card.id, - column: column_name(card.column), + column: canonical_column(card.column), }) .collect(); Json(SyncSnapshot { @@ -514,11 +544,12 @@ async fn sync_snapshot(State(state): State>) -> Json }) } -fn column_name(column: usize) -> &'static str { - ["todo", "doing", "done"] - .get(column) - .copied() - .unwrap_or("todo") +fn canonical_column(column: usize) -> CanonicalColumn { + match column { + 1 => CanonicalColumn::Doing, + 2 => CanonicalColumn::Done, + _ => CanonicalColumn::Todo, + } } // req: sync/005 req: sync/006 req: sync/007 req: sync/013 diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index 80f4190..3e124d5 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/examples/kanban/tests/browser_e2e.rs @@ -626,6 +626,116 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr .clone(); assert_eq!(canonical[2]["title"], "Done"); assert_eq!(canonical[2]["cards"], serde_json::json!(["1", "2", "3"])); + + let divergent_server = driver + .execute_async( + r#" + const done = arguments[arguments.length - 1]; + (async () => { + const local = await fetch('/sync/commands?command_id=history%3A3&card_id=2&column=done', { method: 'POST' }); + const remote = await fetch('/sync/commands?command_id=remote%3A4&card_id=2&column=doing', { method: 'POST' }); + done({ + local: { status: local.status, body: await local.json() }, + remote: { status: remote.status, body: await remote.json() }, + }); + })().catch((error) => done({ error: String(error) })); + "#, + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(divergent_server["local"]["status"], 200); + assert_eq!(divergent_server["local"]["body"]["serverSequence"], 3); + assert_eq!(divergent_server["remote"]["status"], 200); + assert_eq!(divergent_server["remote"]["body"]["serverSequence"], 4); + assert_eq!(divergent_server["remote"]["body"]["canonicalColumn"], "doing"); + + let seeded_conflict = 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('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, + }); + tx.oncomplete = () => done({ seeded: true }); + tx.onabort = () => done({ error: tx.error && tx.error.name }); + }; + "#, + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(seeded_conflict["seeded"], true); + + driver.goto(&format!("http://{app_addr}/sync-demo")).await?; + wait_until( + &driver, + "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'conflicted'", + ) + .await?; + let conflicted = 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'), 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!(conflicted["phase"], "conflicted"); + assert_eq!(conflicted["uploadSequence"], "3"); + assert_eq!(conflicted["snapshotSequence"], "4"); + assert_eq!(conflicted["pending"], "1"); + assert_eq!(conflicted["rebasePending"], "1"); + assert_eq!(conflicted["decision"], "conflicted"); + assert_eq!(conflicted["reason"], "canonical-state-diverged"); + assert_eq!(conflicted["canonicalColumn"], "doing"); + assert_eq!( + conflicted["status"], + "Canonical snapshot 4 conflicts with history:3 (canonical-state-diverged); the pending command remains queued." + ); + assert!(conflicted["error"].is_null()); + assert_eq!(command_count(&driver).await?, 1); + let preserved = 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(['commands', 'meta'], 'readonly'); + const command = tx.objectStore('commands').get('history:3'); + const cursor = tx.objectStore('meta').get('acknowledgementCursor'); + const snapshot = tx.objectStore('meta').get('canonicalSnapshot'); + tx.oncomplete = () => done({ command: command.result, cursor: cursor.result, snapshot: snapshot.result }); + tx.onabort = () => done({ error: tx.error && tx.error.name }); + }; + "#, + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(preserved["command"]["id"], "history:3"); + assert_eq!(preserved["cursor"], 2); + assert_eq!(preserved["snapshot"]["serverSequence"], 2); + + driver.goto(&format!("http://{app_addr}/")).await?; + let divergent_board = 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!(divergent_board[1]["title"], "Doing"); + assert_eq!(divergent_board[1]["cards"], serde_json::json!(["2"])); + assert_eq!(divergent_board[2]["cards"], serde_json::json!(["1", "3"])); Ok(()) } .await;