feat(kanban): recover from missing sync history
req: sync/007\nreq: sync/020
This commit is contained in:
slhx agent
@@ -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
|
||||
|
||||
|
||||
@@ -40,6 +40,7 @@ struct SyncStore(PathBuf);
|
||||
struct SyncState {
|
||||
next_sequence: u64,
|
||||
acknowledgements: BTreeMap<CommandId, SyncAcknowledgement>,
|
||||
retained_after: u64,
|
||||
reconnects: BTreeMap<String, u64>,
|
||||
transient_failure_limit: u8,
|
||||
transient_failures: BTreeMap<CommandId, u8>,
|
||||
@@ -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<SnapshotCard>,
|
||||
}
|
||||
|
||||
#[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::<u64>().ok())
|
||||
.unwrap_or_default();
|
||||
sync.transient_failure_limit = std::env::var("HEMX_KANBAN_SYNC_FAILURES")
|
||||
.ok()
|
||||
.and_then(|value| value.parse::<u8>().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<Arc<AppState>>) -> Json<SyncSnapshot> {
|
||||
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<Arc<AppState>>,
|
||||
@@ -502,10 +549,33 @@ async fn sync_acknowledgements(
|
||||
.keep_alive(KeepAlive::default());
|
||||
}
|
||||
}
|
||||
let events = sync
|
||||
let first_available = sync
|
||||
.acknowledgements
|
||||
.values()
|
||||
.filter(|acknowledgement| acknowledgement.server_sequence > after)
|
||||
.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())
|
||||
@@ -513,7 +583,8 @@ async fn sync_acknowledgements(
|
||||
.json_data(acknowledgement)
|
||||
.expect("serializable acknowledgement"))
|
||||
})
|
||||
.collect::<Vec<Result<Event, Infallible>>>();
|
||||
.collect::<Vec<Result<Event, Infallible>>>()
|
||||
};
|
||||
Sse::new(stream::iter(events).boxed()).keep_alive(KeepAlive::default())
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user