From 687c64d07851ada5fd5ea87046507efea3dee272 Mon Sep 17 00:00:00 2001 From: slhx agent Date: Mon, 13 Jul 2026 16:02:19 +0200 Subject: [PATCH] feat(kanban): acknowledge idempotent server command req: sync/001\nreq: sync/005\nreq: sync/006\nreq: sync/007\nreq: sync/008\nreq: sync/012\nreq: sync/013 --- Cargo.lock | 2 + PLAN.md | 4 +- examples/kanban/Cargo.toml | 4 +- examples/kanban/src/main.rs | 150 ++++++++++++++++++++++++++- examples/kanban/tests/browser_e2e.rs | 125 ++++++++++++++++++++++ 5 files changed, 280 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 250d4ca..f16c27c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -579,6 +579,8 @@ dependencies = [ "hemx-build", "hemx-test", "scraper", + "serde", + "serde_json", "thirtyfour 0.35.0", "tokio", ] diff --git a/PLAN.md b/PLAN.md index 70f13f1..879c552 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:** Ready — Slice 3 is complete; begin with one idempotent server command and canonical acknowledgement through a real reconnect transport before adding conflicts or multi-tab policy. +- **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. Durable server storage, automatic upload from the local command log, retry/backoff state, conflict decisions, multi-tab leadership, 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:** deterministic browser/server test covers offline → reload → reconnect, duplicate and reordered delivery, authorization revocation, rejection, conflict, missing history/fresh snapshot, two tabs, backpressure, and final convergence. +- **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. The completed slice proof must additionally cover durable server restart, automatic local-log upload and retry/backoff, partial reject/conflict decisions, missing-history snapshots, two tabs, backpressure, upgrade mid-queue, and multi-user isolation. ## Slice 5 — local-first multiplayer Kanban milestone diff --git a/examples/kanban/Cargo.toml b/examples/kanban/Cargo.toml index 620a8ab..f1591a3 100644 --- a/examples/kanban/Cargo.toml +++ b/examples/kanban/Cargo.toml @@ -6,7 +6,7 @@ publish = false [features] default = ["server"] -server = ["dep:axum", "dep:futures-util", "dep:hemx-axum", "dep:tokio"] +server = ["dep:axum", "dep:futures-util", "dep:hemx-axum", "dep:serde", "dep:serde_json", "dep:tokio"] client = ["hemx/client"] fixture = [] @@ -29,6 +29,8 @@ axum = { version = "0.8", optional = true } futures-util = { version = "0.3", optional = true } hemx = { path = "../../hemx" } hemx-axum = { path = "../../hemx-axum", optional = true } +serde = { version = "1", features = ["derive"], optional = true } +serde_json = { version = "1", optional = true } tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "time"], optional = true } [target.'cfg(not(target_arch = "wasm32"))'.dependencies] diff --git a/examples/kanban/src/main.rs b/examples/kanban/src/main.rs index 02fcab7..18905b3 100644 --- a/examples/kanban/src/main.rs +++ b/examples/kanban/src/main.rs @@ -1,7 +1,9 @@ use axum::extract::{Query, State}; +use axum::http::StatusCode; +use axum::response::sse::{Event, KeepAlive, Sse}; use axum::response::IntoResponse; -use axum::routing::get; -use axum::Router; +use axum::routing::{get, post}; +use axum::{Json, Router}; use futures_util::{stream, StreamExt}; use hemplate::Hemplate; use hemx::{Html, IntoEffect}; @@ -12,6 +14,7 @@ use hemx_axum::{ use hemx_kanban_example::ui::board::{self as board}; use hemx_kanban_example::ui::board_card as card_board; use hemx_kanban_example::ui::{self, board as board_ui}; +use serde::Serialize; use std::collections::BTreeMap; use std::convert::Infallible; use std::net::SocketAddr; @@ -23,6 +26,58 @@ const COLUMNS: [(&str, &str); 3] = [("backlog", "Backlog"), ("doing", "Doing"), #[derive(Default)] struct AppState { board: Mutex, + sync: Mutex, +} + +#[derive(Default)] +struct SyncState { + next_sequence: u64, + acknowledgements: BTreeMap, + reconnects: BTreeMap, +} + +#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)] +struct CommandId(String); + +impl CommandId { + fn parse(value: Option<&String>) -> Result { + let value = value.map(String::as_str).unwrap_or_default(); + if value.is_empty() + || value.len() > 128 + || !value.bytes().all(|byte| { + byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b':' | b'.') + }) + { + return Err(SyncRejection::BadRequest("invalid command_id")); + } + Ok(Self(value.to_owned())) + } +} + +#[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, + status: &'static str, +} + +#[derive(Debug)] +enum SyncRejection { + BadRequest(&'static str), + Conflict(&'static str), +} + +impl IntoResponse for SyncRejection { + fn into_response(self) -> axum::response::Response { + let (status, error) = match self { + Self::BadRequest(error) => (StatusCode::BAD_REQUEST, error), + Self::Conflict(error) => (StatusCode::CONFLICT, error), + }; + (status, Json(serde_json::json!({ "error": error }))).into_response() + } } #[derive(Default, Clone)] @@ -114,11 +169,17 @@ async fn main() { }, ], }), + sync: Mutex::new(SyncState { + next_sequence: 1, + ..SyncState::default() + }), }); let app = Router::new() .route("/", get(home).post(interact)) .route("/events", get(events)) + .route("/sync/commands", post(sync_command)) + .route("/sync/acknowledgements", get(sync_acknowledgements)) .route(runtime_js_path(), get(runtime)) .with_state(state); @@ -172,6 +233,91 @@ async fn events(Query(params): Query>) -> impl IntoResp sse(batches) } +// req: sync/001 req: sync/008 req: sync/012 +async fn sync_command( + State(state): State>, + Query(params): Query>, +) -> Result, SyncRejection> { + let command_id = CommandId::parse(params.get("command_id"))?; + let card_id = params + .get("card_id") + .and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0) + .ok_or(SyncRejection::BadRequest("invalid card_id"))?; + + let mut sync = state.sync.lock().unwrap(); + if let Some(existing) = sync.acknowledgements.get(&command_id) { + if existing.card_id != card_id { + return Err(SyncRejection::Conflict( + "command_id was already used for a different payload", + )); + } + return Ok(Json(existing.clone())); + } + + let mut board = state.board.lock().unwrap(); + let card = board + .cards + .iter_mut() + .find(|card| card.id == card_id) + .ok_or(SyncRejection::BadRequest("unknown card_id"))?; + card.column = 2; + let acknowledgement = SyncAcknowledgement { + command_id: command_id.0.clone(), + server_sequence: sync.next_sequence, + card_id, + canonical_column: "done", + status: "accepted", + }; + sync.next_sequence += 1; + sync.acknowledgements + .insert(command_id, acknowledgement.clone()); + Ok(Json(acknowledgement)) +} + +// req: sync/005 req: sync/006 req: sync/007 req: sync/013 +async fn sync_acknowledgements( + State(state): State>, + Query(params): Query>, +) -> Sse>> { + let after = params + .get("after") + .and_then(|value| value.parse::().ok()) + .unwrap_or_default(); + let reconnect_key = params + .get("reconnect") + .filter(|value| !value.is_empty()) + .cloned(); + + let mut sync = state.sync.lock().unwrap(); + if let Some(key) = reconnect_key { + let attempts = sync.reconnects.entry(key).or_default(); + *attempts += 1; + if *attempts == 1 { + return Sse::new( + stream::iter([Ok(Event::default() + .comment("reconnect") + .retry(Duration::from_millis(25)))]) + .boxed(), + ) + .keep_alive(KeepAlive::default()); + } + } + let events = 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::>>(); + Sse::new(stream::iter(events).boxed()).keep_alive(KeepAlive::default()) +} + fn registry(state: Arc) -> impl DispatchRegistry { interactions(ui::BUILD_FINGERPRINT) .on(board::create_card, { diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index 2fe3d94..238f187 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/examples/kanban/tests/browser_e2e.rs @@ -88,6 +88,131 @@ async fn server_first_route_does_not_load_optional_client_assets() -> WebDriverR result.and(quit) } +#[tokio::test] +async fn idempotent_server_command_is_acknowledged_after_reconnect() -> WebDriverResult<()> { + // test req: sync/001 req: sync/005 req: sync/006 req: sync/007 + // test req: sync/008 req: sync/012 req: sync/013 + let app_port = available_port(); + let app_addr = format!("127.0.0.1:{app_port}"); + let mut app = Command::new(env!("CARGO_BIN_EXE_hemx-kanban-example")); + app.env("HEMX_KANBAN_ADDR", &app_addr); + let _app = TestProcess::start(app, "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?; + driver.find(By::Css("section[data-hemx-root='kanban']")).await?; + driver + .execute( + r#" + window.__syncProof = { ready: false, opens: 0, events: [], error: null }; + (async () => { + const endpoint = '/sync/commands?command_id=actor-1%3A1&card_id=1'; + const firstResponse = await fetch(endpoint, { method: 'POST' }); + const first = await firstResponse.json(); + const duplicateResponse = await fetch(endpoint, { method: 'POST' }); + const duplicate = await duplicateResponse.json(); + const conflictResponse = await fetch('/sync/commands?command_id=actor-1%3A1&card_id=2', { method: 'POST' }); + const conflict = await conflictResponse.json(); + window.__syncProof.command = { + firstStatus: firstResponse.status, + duplicateStatus: duplicateResponse.status, + first, + duplicate, + conflictStatus: conflictResponse.status, + conflict, + }; + const source = new EventSource('/sync/acknowledgements?after=0&reconnect=browser-proof'); + source.onopen = () => { window.__syncProof.opens += 1; }; + source.addEventListener('acknowledgement', (event) => { + window.__syncProof.events.push({ id: event.lastEventId, acknowledgement: JSON.parse(event.data) }); + window.__syncProof.ready = true; + source.close(); + }); + source.onerror = () => { + if (source.readyState === EventSource.CLOSED && !window.__syncProof.ready) { + window.__syncProof.error = 'acknowledgement stream closed'; + } + }; + })().catch((error) => { window.__syncProof.error = String(error); }); + return true; + "#, + Vec::new(), + ) + .await?; + wait_until( + &driver, + "return window.__syncProof.ready === true || window.__syncProof.error !== null", + ) + .await?; + let proof = driver + .execute("return window.__syncProof", Vec::new()) + .await? + .json() + .clone(); + assert!(proof["error"].is_null(), "sync failed: {proof}"); + assert!( + proof["opens"].as_u64().is_some_and(|opens| opens >= 2), + "transport did not reconnect: {proof}" + ); + assert_eq!(proof["command"]["firstStatus"], 200); + assert_eq!(proof["command"]["duplicateStatus"], 200); + assert_eq!(proof["command"]["first"], proof["command"]["duplicate"]); + assert_eq!(proof["command"]["first"]["commandId"], "actor-1:1"); + assert_eq!(proof["command"]["first"]["serverSequence"], 1); + assert_eq!(proof["command"]["first"]["cardId"], 1); + assert_eq!(proof["command"]["first"]["canonicalColumn"], "done"); + assert_eq!(proof["command"]["first"]["status"], "accepted"); + assert_eq!(proof["command"]["conflictStatus"], 409); + assert_eq!( + proof["command"]["conflict"]["error"], + "command_id was already used for a different payload" + ); + assert_eq!(proof["events"].as_array().map(Vec::len), Some(1)); + assert_eq!(proof["events"][0]["id"], "1"); + assert_eq!( + proof["events"][0]["acknowledgement"], + proof["command"]["first"] + ); + + driver.refresh().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", "3"])); + Ok(()) + } + .await; + let quit = driver.quit().await; + result.and(quit) +} + +async fn wait_until(driver: &WebDriver, script: &str) -> WebDriverResult<()> { + for _ in 0..200 { + if driver.execute(script, Vec::new()).await?.json().as_bool() == Some(true) { + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + panic!("browser condition timed out: {script}"); +} + fn available_port() -> u16 { TcpListener::bind("127.0.0.1:0") .expect("reserve browser test port")