feat(kanban): persist canonical acknowledgements

req: sync/001\nreq: sync/005\nreq: sync/007\nreq: sync/008\nreq: sync/013

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. A dedicated opt-in sync route now reads one pending IndexedDB command, survives one injected 503 through bounded exponential backoff with randomized jitter, uploads through the idempotent command endpoint, waits for canonical acknowledgement on the reconnecting transport, and only then removes the durable command; a canonical payload conflict is not retried and remains durable with a visible reason. Durable server storage, exhausted-retry/offline recovery, broader conflict decisions, multi-tab leadership, 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, survives one injected 503 through bounded exponential backoff with randomized jitter, uploads through the idempotent command endpoint, waits for canonical acknowledgement on the reconnecting transport, and only then removes the durable command; a canonical payload conflict is not retried and remains durable with a visible reason. Exhausted-retry/offline recovery, broader 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:** `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. The completed slice proof must additionally cover durable server restart, exhausted-retry/offline recovery, 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. The completed slice proof must additionally cover exhausted-retry/offline recovery, partial reject/conflict decisions, missing-history snapshots, two tabs, backpressure, upgrade mid-queue, and multi-user isolation.
 
 ## Slice 5 — local-first multiplayer Kanban milestone
 

examples/kanban/src/main.rs

@@ -14,10 +14,13 @@ 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 serde::{Deserialize, Serialize};
 use std::collections::{BTreeMap, BTreeSet};
 use std::convert::Infallible;
+use std::fs::{self, OpenOptions};
+use std::io::Write;
 use std::net::SocketAddr;
+use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
@@ -27,8 +30,12 @@ const COLUMNS: [(&str, &str); 3] = [("backlog", "Backlog"), ("doing", "Doing"),
 struct AppState {
     board: Mutex<BoardState>,
     sync: Mutex<SyncState>,
+    sync_store: Option<SyncStore>,
 }
 
+#[derive(Clone)]
+struct SyncStore(PathBuf);
+
 #[derive(Default)]
 struct SyncState {
     next_sequence: u64,
@@ -43,7 +50,10 @@ struct CommandId(String);
 
 impl CommandId {
     fn parse(value: Option<&String>) -> Result<Self, SyncRejection> {
-        let value = value.map(String::as_str).unwrap_or_default();
+        Self::parse_str(value.map(String::as_str).unwrap_or_default())
+    }
+
+    fn parse_str(value: &str) -> Result<Self, SyncRejection> {
         if value.is_empty()
             || value.len() > 128
             || !value.bytes().all(|byte| {
@@ -71,6 +81,7 @@ enum SyncRejection {
     BadRequest(&'static str),
     Conflict(&'static str),
     Transient,
+    Storage,
 }
 
 impl IntoResponse for SyncRejection {
@@ -79,11 +90,125 @@ impl IntoResponse for SyncRejection {
             Self::BadRequest(error) => (StatusCode::BAD_REQUEST, error),
             Self::Conflict(error) => (StatusCode::CONFLICT, error),
             Self::Transient => (StatusCode::SERVICE_UNAVAILABLE, "transient sync failure"),
+            Self::Storage => (StatusCode::INTERNAL_SERVER_ERROR, "sync storage failed"),
         };
         (status, Json(serde_json::json!({ "error": error }))).into_response()
     }
 }
 
+#[derive(Deserialize, Serialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct PersistedSync {
+    schema_version: u8,
+    next_sequence: u64,
+    acknowledgements: Vec<PersistedAcknowledgement>,
+}
+
+#[derive(Deserialize, Serialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct PersistedAcknowledgement {
+    command_id: String,
+    server_sequence: u64,
+    card_id: u64,
+}
+
+impl SyncStore {
+    fn load(&self) -> Result<SyncState, String> {
+        if !self.0.exists() {
+            return Ok(SyncState {
+                next_sequence: 1,
+                ..SyncState::default()
+            });
+        }
+        let bytes =
+            fs::read(&self.0).map_err(|error| format!("read {}: {error}", self.0.display()))?;
+        let persisted: PersistedSync = serde_json::from_slice(&bytes)
+            .map_err(|error| format!("decode {}: {error}", self.0.display()))?;
+        if persisted.schema_version != 1 || persisted.next_sequence == 0 {
+            return Err(format!("unsupported sync store {}", self.0.display()));
+        }
+        let mut acknowledgements = BTreeMap::new();
+        for stored in persisted.acknowledgements {
+            let command_id = CommandId::parse_str(&stored.command_id)
+                .map_err(|_| format!("invalid command id in {}", self.0.display()))?;
+            if stored.server_sequence == 0 || stored.card_id == 0 {
+                return Err(format!("invalid acknowledgement in {}", self.0.display()));
+            }
+            let acknowledgement = SyncAcknowledgement {
+                command_id: stored.command_id,
+                server_sequence: stored.server_sequence,
+                card_id: stored.card_id,
+                canonical_column: "done",
+                status: "accepted",
+            };
+            if acknowledgements
+                .insert(command_id, acknowledgement)
+                .is_some()
+            {
+                return Err(format!("duplicate command id in {}", self.0.display()));
+            }
+        }
+        Ok(SyncState {
+            next_sequence: persisted.next_sequence,
+            acknowledgements,
+            ..SyncState::default()
+        })
+    }
+
+    fn persist(
+        &self,
+        sync: &SyncState,
+        acknowledgement: &SyncAcknowledgement,
+    ) -> Result<(), String> {
+        let mut acknowledgements = sync
+            .acknowledgements
+            .values()
+            .map(PersistedAcknowledgement::from)
+            .collect::<Vec<_>>();
+        acknowledgements.push(PersistedAcknowledgement::from(acknowledgement));
+        acknowledgements.sort_by_key(|item| item.server_sequence);
+        let persisted = PersistedSync {
+            schema_version: 1,
+            next_sequence: acknowledgement.server_sequence + 1,
+            acknowledgements,
+        };
+        let bytes = serde_json::to_vec_pretty(&persisted)
+            .map_err(|error| format!("encode {}: {error}", self.0.display()))?;
+        if let Some(parent) = self.0.parent() {
+            fs::create_dir_all(parent)
+                .map_err(|error| format!("create {}: {error}", parent.display()))?;
+        }
+        let temporary = self.0.with_extension("tmp");
+        let mut file = OpenOptions::new()
+            .create(true)
+            .truncate(true)
+            .write(true)
+            .open(&temporary)
+            .map_err(|error| format!("open {}: {error}", temporary.display()))?;
+        file.write_all(&bytes)
+            .and_then(|()| file.sync_all())
+            .map_err(|error| format!("write {}: {error}", temporary.display()))?;
+        fs::rename(&temporary, &self.0)
+            .map_err(|error| format!("replace {}: {error}", self.0.display()))?;
+        if let Some(parent) = self.0.parent() {
+            fs::File::open(parent)
+                .and_then(|directory| directory.sync_all())
+                .map_err(|error| format!("sync {}: {error}", parent.display()))?;
+        }
+        Ok(())
+    }
+}
+
+impl From<&SyncAcknowledgement> for PersistedAcknowledgement {
+    fn from(value: &SyncAcknowledgement) -> Self {
+        Self {
+            command_id: value.command_id.clone(),
+            server_sequence: value.server_sequence,
+            card_id: value.card_id,
+        }
+    }
+}
+
 #[derive(Default, Clone)]
 struct BoardState {
     next_id: u64,
@@ -157,32 +282,33 @@ struct Presence {
 
 #[tokio::main]
 async fn main() {
-    let state = Arc::new(AppState {
-        board: Mutex::new(BoardState {
-            next_id: 4,
-            cards: vec![
-                Card {
-                    id: 1,
-                    title: "Write requirements".into(),
-                    column: 0,
-                },
-                Card {
-                    id: 2,
-                    title: "Build browser example".into(),
-                    column: 1,
-                },
-                Card {
-                    id: 3,
-                    title: "Verify with HTTP".into(),
-                    column: 2,
-                },
-            ],
-        }),
-        sync: Mutex::new(SyncState {
+    let sync_store = std::env::var_os("HEMX_KANBAN_SYNC_STORE")
+        .map(PathBuf::from)
+        .map(SyncStore);
+    let mut sync = sync_store
+        .as_ref()
+        .map(SyncStore::load)
+        .transpose()
+        .unwrap_or_else(|error| panic!("cannot start with sync store: {error}"))
+        .unwrap_or_else(|| SyncState {
             next_sequence: 1,
-            fail_first_upload: std::env::var_os("HEMX_KANBAN_FAIL_FIRST_SYNC").is_some(),
             ..SyncState::default()
-        }),
+        });
+    sync.fail_first_upload = std::env::var_os("HEMX_KANBAN_FAIL_FIRST_SYNC").is_some();
+    let mut board = initial_board();
+    for acknowledgement in sync.acknowledgements.values() {
+        if let Some(card) = board
+            .cards
+            .iter_mut()
+            .find(|card| card.id == acknowledgement.card_id)
+        {
+            card.column = 2;
+        }
+    }
+    let state = Arc::new(AppState {
+        board: Mutex::new(board),
+        sync: Mutex::new(sync),
+        sync_store,
     });
 
     let app = Router::new()
@@ -203,6 +329,29 @@ async fn main() {
     axum::serve(listener, app).await.unwrap();
 }
 
+fn initial_board() -> BoardState {
+    BoardState {
+        next_id: 4,
+        cards: vec![
+            Card {
+                id: 1,
+                title: "Write requirements".into(),
+                column: 0,
+            },
+            Card {
+                id: 2,
+                title: "Build browser example".into(),
+                column: 1,
+            },
+            Card {
+                id: 3,
+                title: "Verify with HTTP".into(),
+                column: 2,
+            },
+        ],
+    }
+}
+
 // req: examples/001 req: component/003
 async fn home(State(state): State<Arc<AppState>>, request: PageRequest) -> impl IntoResponse {
     let board = state.board.lock().unwrap().clone();
@@ -289,12 +438,11 @@ async fn sync_command(
     }
 
     let mut board = state.board.lock().unwrap();
-    let card = board
+    let card_index = board
         .cards
-        .iter_mut()
-        .find(|card| card.id == card_id)
+        .iter()
+        .position(|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,
@@ -302,6 +450,13 @@ async fn sync_command(
         canonical_column: "done",
         status: "accepted",
     };
+    if let Some(store) = &state.sync_store {
+        store.persist(&sync, &acknowledgement).map_err(|error| {
+            eprintln!("sync persistence failed: {error}");
+            SyncRejection::Storage
+        })?;
+    }
+    board.cards[card_index].column = 2;
     sync.next_sequence += 1;
     sync.acknowledgements
         .insert(command_id, acknowledgement.clone());

examples/kanban/tests/browser_e2e.rs

@@ -1,4 +1,5 @@
 use hemx_test::TestProcess;
+use std::fs;
 use std::net::TcpListener;
 use std::process::Command;
 use std::time::Duration;
@@ -368,6 +369,134 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
     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
+    let app_port = available_port();
+    let app_addr = format!("127.0.0.1:{app_port}");
+    let store = std::env::temp_dir().join(format!(
+        "hemx-kanban-sync-{}-{app_port}.json",
+        std::process::id()
+    ));
+    let _ = fs::remove_file(&store);
+
+    let mut first_app_command = Command::new(env!("CARGO_BIN_EXE_hemx-kanban-example"));
+    first_app_command
+        .env("HEMX_KANBAN_ADDR", &app_addr)
+        .env("HEMX_KANBAN_SYNC_STORE", &store);
+    let first_app = TestProcess::start(
+        first_app_command,
+        "hemx-kanban-first",
+        &app_addr,
+        STARTUP_TIMEOUT,
+    )
+    .expect("start first 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 accepted = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                fetch('/sync/commands?command_id=restart-proof%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!(accepted["status"], 200);
+        assert_eq!(accepted["body"]["serverSequence"], 1);
+        assert_eq!(accepted["body"]["canonicalColumn"], "done");
+        assert!(store.is_file(), "server did not materialize sync store");
+        let persisted = fs::read_to_string(&store).expect("read sync store");
+        assert!(persisted.contains("restart-proof:1"));
+        assert!(persisted.contains("\"schemaVersion\": 1"));
+
+        drop(first_app);
+        let mut second_app_command = Command::new(env!("CARGO_BIN_EXE_hemx-kanban-example"));
+        second_app_command
+            .env("HEMX_KANBAN_ADDR", &app_addr)
+            .env("HEMX_KANBAN_SYNC_STORE", &store);
+        let second_app = TestProcess::start(
+            second_app_command,
+            "hemx-kanban-second",
+            &app_addr,
+            STARTUP_TIMEOUT,
+        )
+        .expect("restart hemx-kanban from durable sync store");
+
+        let after_restart = driver
+            .execute_async(
+                r#"
+                const done = arguments[arguments.length - 1];
+                fetch('/sync/commands?command_id=restart-proof%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!(after_restart["status"], 200);
+        assert_eq!(after_restart["body"], accepted["body"]);
+
+        driver
+            .execute(
+                r#"
+                window.__restartReplay = null;
+                const source = new EventSource('/sync/acknowledgements?after=0');
+                source.addEventListener('acknowledgement', (event) => {
+                  window.__restartReplay = { id: event.lastEventId, body: JSON.parse(event.data) };
+                  source.close();
+                });
+                return true;
+                "#,
+                Vec::new(),
+            )
+            .await?;
+        wait_until(&driver, "return window.__restartReplay !== null").await?;
+        let replay = driver
+            .execute("return window.__restartReplay", Vec::new())
+            .await?
+            .json()
+            .clone();
+        assert_eq!(replay["id"], "1");
+        assert_eq!(replay["body"], accepted["body"]);
+
+        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"]));
+        drop(second_app);
+        Ok(())
+    }
+    .await;
+    let quit = driver.quit().await;
+    let _ = fs::remove_file(&store);
+    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) {