From db57b0c474e7452374bd487a48a55e81b2ff2a34 Mon Sep 17 00:00:00 2001 From: slhx agent Date: Mon, 13 Jul 2026 16:53:49 +0200 Subject: [PATCH] feat(kanban): recover from missing sync history req: sync/007\nreq: sync/020 --- PLAN.md | 4 +- examples/kanban/src/main.rs | 91 ++++++++++++++++++--- examples/kanban/static/sync.js | 15 ++++ examples/kanban/tests/browser_e2e.rs | 114 +++++++++++++++++++++++++++ 4 files changed, 212 insertions(+), 12 deletions(-) diff --git a/PLAN.md b/PLAN.md index 92e6c41..c5f6599 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. 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; 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. - **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. The completed slice proof must additionally cover broader partial reject/conflict decisions, missing-history snapshots, 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_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. ## Slice 5 — local-first multiplayer Kanban milestone diff --git a/examples/kanban/src/main.rs b/examples/kanban/src/main.rs index 310d5d7..a7fb8a4 100644 --- a/examples/kanban/src/main.rs +++ b/examples/kanban/src/main.rs @@ -40,6 +40,7 @@ struct SyncStore(PathBuf); struct SyncState { next_sequence: u64, acknowledgements: BTreeMap, + retained_after: u64, reconnects: BTreeMap, transient_failure_limit: u8, transient_failures: BTreeMap, @@ -76,6 +77,21 @@ struct SyncAcknowledgement { status: &'static str, } +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct SyncSnapshot { + schema_version: u8, + server_sequence: u64, + cards: Vec, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct SnapshotCard { + id: u64, + column: &'static str, +} + #[derive(Debug)] enum SyncRejection { BadRequest(&'static str), @@ -294,6 +310,10 @@ async fn main() { next_sequence: 1, ..SyncState::default() }); + sync.retained_after = std::env::var("HEMX_KANBAN_RETAINED_AFTER") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or_default(); sync.transient_failure_limit = std::env::var("HEMX_KANBAN_SYNC_FAILURES") .ok() .and_then(|value| value.parse::().ok()) @@ -321,6 +341,7 @@ async fn main() { .route("/sync.js", get(sync_js)) .route("/sync/commands", post(sync_command)) .route("/sync/acknowledgements", get(sync_acknowledgements)) + .route("/sync/snapshot", get(sync_snapshot)) .route(runtime_js_path(), get(runtime)) .with_state(state); @@ -474,6 +495,32 @@ async fn sync_command( Ok(Json(acknowledgement)) } +// req: sync/007 req: sync/020 +async fn sync_snapshot(State(state): State>) -> Json { + let board = state.board.lock().unwrap(); + let sync = state.sync.lock().unwrap(); + let cards = board + .cards + .iter() + .map(|card| SnapshotCard { + id: card.id, + column: column_name(card.column), + }) + .collect(); + Json(SyncSnapshot { + schema_version: 1, + server_sequence: sync.next_sequence.saturating_sub(1), + cards, + }) +} + +fn column_name(column: usize) -> &'static str { + ["todo", "doing", "done"] + .get(column) + .copied() + .unwrap_or("todo") +} + // req: sync/005 req: sync/006 req: sync/007 req: sync/013 async fn sync_acknowledgements( State(state): State>, @@ -502,18 +549,42 @@ async fn sync_acknowledgements( .keep_alive(KeepAlive::default()); } } - let events = sync + let first_available = sync .acknowledgements .values() - .filter(|acknowledgement| acknowledgement.server_sequence > after) - .map(|acknowledgement| { - Ok(Event::default() - .id(acknowledgement.server_sequence.to_string()) - .event("acknowledgement") - .json_data(acknowledgement) - .expect("serializable acknowledgement")) - }) - .collect::>>(); + .filter(|acknowledgement| acknowledgement.server_sequence > sync.retained_after) + .map(|acknowledgement| acknowledgement.server_sequence) + .min(); + let latest = sync.next_sequence.saturating_sub(1); + let history_missing = after < latest + && first_available.is_none_or(|first_sequence| after.saturating_add(1) < first_sequence); + let events = if history_missing { + vec![Ok(Event::default() + .id(latest.to_string()) + .event("snapshot-required") + .json_data(serde_json::json!({ + "after": after, + "firstAvailable": first_available, + "latest": latest, + "snapshotUrl": "/sync/snapshot", + })) + .expect("serializable missing history event"))] + } else { + sync.acknowledgements + .values() + .filter(|acknowledgement| { + acknowledgement.server_sequence > after + && acknowledgement.server_sequence > sync.retained_after + }) + .map(|acknowledgement| { + Ok(Event::default() + .id(acknowledgement.server_sequence.to_string()) + .event("acknowledgement") + .json_data(acknowledgement) + .expect("serializable acknowledgement")) + }) + .collect::>>() + }; Sse::new(stream::iter(events).boxed()).keep_alive(KeepAlive::default()) } diff --git a/examples/kanban/static/sync.js b/examples/kanban/static/sync.js index 0c46408..2f1d6c0 100644 --- a/examples/kanban/static/sync.js +++ b/examples/kanban/static/sync.js @@ -140,6 +140,21 @@ async function synchronize(command) { root.dispatchEvent(new CustomEvent("kanban:sync-acknowledged", { detail: canonical })); source.close(); }); + source.addEventListener("snapshot-required", async (event) => { + const missing = JSON.parse(event.data); + const response = await fetch(missing.snapshotUrl); + if (!response.ok) throw new Error(`snapshot failed with ${response.status}`); + const snapshot = await response.json(); + const queued = await pendingCommands(database); + 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 } })); + source.close(); + }); } catch (error) { setOnline(false); if (error instanceof UploadError && !error.retryable) throw error; diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index ff8592f..3b7add7 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/examples/kanban/tests/browser_e2e.rs @@ -494,6 +494,120 @@ async fn exhausted_offline_retries_keep_command_until_later_reconnect() -> WebDr result.and(quit) } +#[tokio::test] +async fn missing_history_loads_snapshot_and_preserves_pending_intent() -> WebDriverResult<()> { + // test req: sync/007 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")); + app_command + .env("HEMX_KANBAN_ADDR", &app_addr) + .env("HEMX_KANBAN_RETAINED_AFTER", "1"); + 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_server = driver + .execute_async( + r#" + const done = arguments[arguments.length - 1]; + fetch('/sync/commands?command_id=history%3A1&card_id=1', { method: 'POST' }) + .then(async (response) => done({ status: response.status, body: await response.json() })) + .catch((error) => done({ error: String(error) })); + "#, + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(seeded_server["status"], 200); + assert_eq!(seeded_server["body"]["serverSequence"], 1); + + let seeded_local = 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: 'history:2', schemaVersion: 1, actor: 'history', session: 'history-session', + causal: 2, 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_local["seeded"], true); + + driver.goto(&format!("http://{app_addr}/sync-demo")).await?; + wait_until( + &driver, + "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'rebase-required'", + ) + .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') }", + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(fallback["phase"], "rebase-required"); + assert_eq!(fallback["uploadSequence"], "2"); + assert_eq!(fallback["snapshotSequence"], "2"); + assert_eq!(fallback["snapshotSchema"], "1"); + assert_eq!(fallback["snapshotCards"], "3"); + assert_eq!(fallback["pending"], "1"); + assert_eq!(fallback["rebasePending"], "1"); + assert_eq!(fallback["canonicalColumn"], "done"); + assert_eq!( + fallback["status"], + "History is unavailable; loaded snapshot 2 and preserved 1 pending command for rebase." + ); + assert!(fallback["error"].is_null()); + assert_eq!(command_count(&driver).await?, 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]["title"], "Done"); + assert_eq!(canonical[2]["cards"], serde_json::json!(["1", "2", "3"])); + Ok(()) + } + .await; + let quit = driver.quit().await; + result.and(quit) +} + #[tokio::test] async fn canonical_acknowledgement_survives_server_restart() -> WebDriverResult<()> { // test req: sync/001 req: sync/005 req: sync/007 req: sync/008 req: sync/013