tmk241/hemx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(kanban): deterministically rebase snapshot

Browse Source 
req: sync/011
This commit is contained in:
slhx agent
2026-07-13 16:58:51 +02:00
parent db57b0c474
commit d6aaca8127
3 changed files with 67 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
PLAN.md
+2 -2
View File
@@ -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
examples/kanban/static/sync.js
+32 -3
View File
@@ -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) {
examples/kanban/tests/browser_e2e.rs
+33 -8
View File
@@ -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