From 2bc68c57777b3461d5519f518b1e7b25392cbfe1 Mon Sep 17 00:00:00 2001 From: slhx agent Date: Mon, 13 Jul 2026 21:48:00 +0200 Subject: [PATCH] fix(sync): bound acknowledgement stream req: operations/004 --- PLAN.md | 2 +- examples/kanban/src/main.rs | 59 +++++++++++++-- examples/kanban/static/sync.js | 18 ++++- examples/kanban/tests/browser_e2e.rs | 107 +++++++++++++++++++++++++++ 4 files changed, 178 insertions(+), 8 deletions(-) diff --git a/PLAN.md b/PLAN.md index 9f85ba6..f8d23fe 100644 --- a/PLAN.md +++ b/PLAN.md @@ -46,7 +46,7 @@ encryption, retention, backup, and deployment policy remain host concerns. - **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-wasm --test browser flat_patch_persists_offline_then_uploads_with_same_operation_identity -- --exact` proves `sync/002` and `sync/003` through generated real WASM: `SyncEffect::send_patch` travels beside ordinary DOM effects, the strict flat scalar schema is accepted, the offline interaction performs zero patch requests while remaining durable, and reconnect uploads with identical opaque operation/idempotency identity before matching acknowledgement removes it. `cargo test -p hemx-wasm --test browser kanban_public_api_offline_sync_journey_converges_without_duplicate_replay -- --exact` proves `sync/023` through the public generated client handler and real WASM: offline durable mutation, unavailable-origin reload projection, reconnect upload, canonical convergence, duplicate idempotency without a second event, conflicting duplicate rejection, invalid-command rejection, and an empty local queue. `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_and_user_conflict_resolution_preserve_suffix -- --exact` proves retained-history gap detection, typed/versioned snapshot fallback, deterministic already-canonical convergence, atomic snapshot/cursor commit, then a divergent canonical update producing explicit conflict with a two-command queue; the user’s canonical-state action removes only the conflicted head, retains and resumes the suffix, and reaches canonical convergence with an empty queue. `cargo test -p hemx-kanban-example --test browser_e2e keep_local_retry_preserves_conflicted_command_and_suffix_order -- --exact` proves the keep-local action derives one resolution identity; a permanent rejection preserves order and re-enables the choice; three later transport failures still preserve the original conflicted head plus suffix and expose manual recovery; final acceptance removes only the head and converges the untouched suffix without duplication or loss. `cargo test -p hemx-kanban-example --test browser_e2e redacted_sync_diagnostics_are_bounded_and_leak_no_sensitive_material -- --exact` proves a 15-second-old queue is bucketed rather than timestamped, cursor and acknowledgement latency advance after recovery, diagnostics stay under a fixed small readout, and command id/payload/actor/session/cookie material never reaches the diagnostics surface; the conflict and keep-local tests additionally assert conflict/rejection counters at their actual failure boundaries. `cargo test -p hemx-kanban-example --test browser_e2e two_tabs_coordinate_single_uploader_and_takeover_without_duplicate_application -- --exact` proves one retry-exhausted leader/one explicit standby, zero follower upload before takeover, lease takeover after the leader closes, one canonical sequence/event, one queue removal, and one board application. `cargo test -p hemx-kanban-example --test browser_e2e upload_backpressure_keeps_pending_work_visible_and_recoverable -- --exact` proves one in-flight upload, a two-acknowledgement activation limit, one retained durable command with visible recovery state, and explicit retry draining the final command without loss. `cargo test -p hemx-kanban-example --test browser_e2e mixed_queue_removes_accepted_prefix_and_retains_rejected_tail -- --exact` proves an accepted prefix is canonically applied and removed once before a permanent rejection stops processing, exposes its typed HTTP/server cause, disables blind retry, and leaves both the rejected command and untouched suffix durable. `cargo test -p hemx-kanban-example --test browser_e2e schema_upgrade_preserves_queued_order_and_local_intent -- --exact` proves a three-command v1 queue migrates atomically to the explicit-target v2 schema, remains byte-for-intent ordered after interrupted upload, then receives canonical sequences 1–3 in original order and drains without loss. `cargo test -p hemx-kanban-example --test browser_e2e account_partition_hides_replay_and_export_until_owner_returns -- --exact` proves a beta editor and alpha viewer enumerate zero commands and issue no replay for an alpha owner queue, signed-out startup cannot open a partition, no foreign id/export surface leaks, and only the returning alpha owner can export the intact command then resume it exactly once. `cargo test -p hemx-kanban-example --test browser_e2e canonical_snapshot_and_history_are_tenant_scoped -- --exact` proves alpha and beta sessions receive only their own canonical cards and acknowledgement events with independent sequence 1 cursors, while signed-out snapshot/history reads receive 401 and expose no stream. `cargo test -p hemx-kanban-example --test browser_e2e identical_sync_inputs_reconcile_deterministically -- --exact` proves `sync/022` through a real browser and live sync endpoints: one accepted server result and canonical snapshot plus an ordered command sequence produce byte-equivalent structured reconciliation records on independent cloned inputs, do not mutate inputs, and select the explicit server-authoritative model. `cargo test -p hemx-kanban-example --test browser_e2e sync_requests_timeout_and_cancel_on_pagehide -- --exact` proves the sync request boundary enforces a named finite timeout and reports `TimeoutError`, while page teardown cancels a pending request with `AbortError`, closes the acknowledgement stream, and clears retry/lease timers. Per-requirement audit: every `sync/001-023` row now has concrete browser proof; existing live-browser proofs also cover `operations/001-002`, `operations/005`, `security/002`, `security/004`, `performance/004`, and `performance/005`, while this slice closes the sync request/stream leg of `operations/003`. `cargo test -p hemx-kanban-example --test browser_e2e adversarial_wire_inputs_are_rejected_before_partial_application -- --exact` proves malformed, oversized, truncated, trailing, unknown-version, and invalid-kind batch/state/command inputs fail without UI mutation; `cargo test -p hemx-wasm --test browser flat_patch_persists_offline_then_uploads_with_same_operation_identity -- --exact` additionally proves unknown-version and oversized event/state inputs are rejected at the generated real-WASM handler boundary before application code. The runtime now bounds total payloads, fields, vectors, varints, UTF-8, enums, and option discriminants before allocation or dispatch, and the durable queue validates every loaded command before replay, closing `security/005`. `cargo test -p hemx-axum --test response interaction_boundary_honors_media_type_and_host_body_limit -- --exact` proves the mutation extractor rejects missing/unexpected media types with 415 and an Axum host-configured oversized body with 413 before handler dispatch, while a valid bounded form dispatches exactly once; the extractor documentation names `DefaultBodyLimit`, and CSRF remains the existing documented application/middleware policy, closing `security/003`. Slice 4 remains open on `operations/004` bounded liveness/slow-consumer proof and the non-sync handler/background-replay legs of `operations/003`. The next coherent slice is `operations/004`: bound and prove heartbeat, reconnect/backoff, slow-consumer buffering, and cancellation behavior for the acknowledgement stream. +- **Proof:** `cargo test -p hemx-wasm --test browser flat_patch_persists_offline_then_uploads_with_same_operation_identity -- --exact` proves `sync/002` and `sync/003` through generated real WASM: `SyncEffect::send_patch` travels beside ordinary DOM effects, the strict flat scalar schema is accepted, the offline interaction performs zero patch requests while remaining durable, and reconnect uploads with identical opaque operation/idempotency identity before matching acknowledgement removes it. `cargo test -p hemx-wasm --test browser kanban_public_api_offline_sync_journey_converges_without_duplicate_replay -- --exact` proves `sync/023` through the public generated client handler and real WASM: offline durable mutation, unavailable-origin reload projection, reconnect upload, canonical convergence, duplicate idempotency without a second event, conflicting duplicate rejection, invalid-command rejection, and an empty local queue. `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_and_user_conflict_resolution_preserve_suffix -- --exact` proves retained-history gap detection, typed/versioned snapshot fallback, deterministic already-canonical convergence, atomic snapshot/cursor commit, then a divergent canonical update producing explicit conflict with a two-command queue; the user’s canonical-state action removes only the conflicted head, retains and resumes the suffix, and reaches canonical convergence with an empty queue. `cargo test -p hemx-kanban-example --test browser_e2e keep_local_retry_preserves_conflicted_command_and_suffix_order -- --exact` proves the keep-local action derives one resolution identity; a permanent rejection preserves order and re-enables the choice; three later transport failures still preserve the original conflicted head plus suffix and expose manual recovery; final acceptance removes only the head and converges the untouched suffix without duplication or loss. `cargo test -p hemx-kanban-example --test browser_e2e redacted_sync_diagnostics_are_bounded_and_leak_no_sensitive_material -- --exact` proves a 15-second-old queue is bucketed rather than timestamped, cursor and acknowledgement latency advance after recovery, diagnostics stay under a fixed small readout, and command id/payload/actor/session/cookie material never reaches the diagnostics surface; the conflict and keep-local tests additionally assert conflict/rejection counters at their actual failure boundaries. `cargo test -p hemx-kanban-example --test browser_e2e two_tabs_coordinate_single_uploader_and_takeover_without_duplicate_application -- --exact` proves one retry-exhausted leader/one explicit standby, zero follower upload before takeover, lease takeover after the leader closes, one canonical sequence/event, one queue removal, and one board application. `cargo test -p hemx-kanban-example --test browser_e2e upload_backpressure_keeps_pending_work_visible_and_recoverable -- --exact` proves one in-flight upload, a two-acknowledgement activation limit, one retained durable command with visible recovery state, and explicit retry draining the final command without loss. `cargo test -p hemx-kanban-example --test browser_e2e mixed_queue_removes_accepted_prefix_and_retains_rejected_tail -- --exact` proves an accepted prefix is canonically applied and removed once before a permanent rejection stops processing, exposes its typed HTTP/server cause, disables blind retry, and leaves both the rejected command and untouched suffix durable. `cargo test -p hemx-kanban-example --test browser_e2e schema_upgrade_preserves_queued_order_and_local_intent -- --exact` proves a three-command v1 queue migrates atomically to the explicit-target v2 schema, remains byte-for-intent ordered after interrupted upload, then receives canonical sequences 1–3 in original order and drains without loss. `cargo test -p hemx-kanban-example --test browser_e2e account_partition_hides_replay_and_export_until_owner_returns -- --exact` proves a beta editor and alpha viewer enumerate zero commands and issue no replay for an alpha owner queue, signed-out startup cannot open a partition, no foreign id/export surface leaks, and only the returning alpha owner can export the intact command then resume it exactly once. `cargo test -p hemx-kanban-example --test browser_e2e canonical_snapshot_and_history_are_tenant_scoped -- --exact` proves alpha and beta sessions receive only their own canonical cards and acknowledgement events with independent sequence 1 cursors, while signed-out snapshot/history reads receive 401 and expose no stream. `cargo test -p hemx-kanban-example --test browser_e2e identical_sync_inputs_reconcile_deterministically -- --exact` proves `sync/022` through a real browser and live sync endpoints: one accepted server result and canonical snapshot plus an ordered command sequence produce byte-equivalent structured reconciliation records on independent cloned inputs, do not mutate inputs, and select the explicit server-authoritative model. `cargo test -p hemx-kanban-example --test browser_e2e sync_requests_timeout_and_cancel_on_pagehide -- --exact` proves the sync request boundary enforces a named finite timeout and reports `TimeoutError`, while page teardown cancels a pending request with `AbortError`, closes the acknowledgement stream, and clears retry/lease timers. Per-requirement audit: every `sync/001-023` row now has concrete browser proof; existing live-browser proofs also cover `operations/001-002`, `operations/005`, `security/002`, `security/004`, `performance/004`, and `performance/005`, while this slice closes the sync request/stream leg of `operations/003`. `cargo test -p hemx-kanban-example --test browser_e2e adversarial_wire_inputs_are_rejected_before_partial_application -- --exact` proves malformed, oversized, truncated, trailing, unknown-version, and invalid-kind batch/state/command inputs fail without UI mutation; `cargo test -p hemx-wasm --test browser flat_patch_persists_offline_then_uploads_with_same_operation_identity -- --exact` additionally proves unknown-version and oversized event/state inputs are rejected at the generated real-WASM handler boundary before application code. The runtime now bounds total payloads, fields, vectors, varints, UTF-8, enums, and option discriminants before allocation or dispatch, and the durable queue validates every loaded command before replay, closing `security/005`. `cargo test -p hemx-axum --test response interaction_boundary_honors_media_type_and_host_body_limit -- --exact` proves the mutation extractor rejects missing/unexpected media types with 415 and an Axum host-configured oversized body with 413 before handler dispatch, while a valid bounded form dispatches exactly once; the extractor documentation names `DefaultBodyLimit`, and CSRF remains the existing documented application/middleware policy, closing `security/003`. `cargo test -p hemx-kanban-example --test browser_e2e acknowledgement_stream_bounds_reconnect_buffering_heartbeat_and_cancellation -- --exact` closes `operations/004`: the live stream exposes a finite 15-second production heartbeat (25 ms under proof configuration), exactly three bounded reconnect delays (100/250/500 ms), converts 65 pending events over its 64-event transport budget into one canonical snapshot recovery event, then emits heartbeats after recovery; both proof streams are explicitly closed, and the page teardown path closes its owned stream and reports cancellation. Slice 4 remains open only on the non-sync handler/background-replay legs of `operations/003`. The next coherent slice is `operations/003`: apply and prove finite timeout/cancellation policy for ordinary server handlers and startup/background replay, not just sync requests. ## Slice 5 — local-first multiplayer Kanban milestone diff --git a/examples/kanban/src/main.rs b/examples/kanban/src/main.rs index 093ad52..9b5da37 100644 --- a/examples/kanban/src/main.rs +++ b/examples/kanban/src/main.rs @@ -25,6 +25,13 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; const COLUMNS: [(&str, &str); 3] = [("backlog", "Backlog"), ("doing", "Doing"), ("done", "Done")]; +const ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT: usize = 64; +const ACKNOWLEDGEMENT_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15); +const ACKNOWLEDGEMENT_RECONNECT_BACKOFF: [Duration; 3] = [ + Duration::from_millis(100), + Duration::from_millis(250), + Duration::from_millis(500), +]; #[derive(Default)] struct AppState { @@ -32,6 +39,7 @@ struct AppState { sync: Mutex, sync_store: Option, sync_sessions: SyncSessionTokens, + acknowledgement_heartbeat_interval: Duration, } #[derive(Default)] @@ -437,6 +445,12 @@ async fn main() { sync: Mutex::new(sync), sync_store, sync_sessions: SyncSessionTokens::from_env(), + acknowledgement_heartbeat_interval: std::env::var("HEMX_KANBAN_ACK_HEARTBEAT_MS") + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|milliseconds| *milliseconds > 0) + .map(Duration::from_millis) + .unwrap_or(ACKNOWLEDGEMENT_HEARTBEAT_INTERVAL), }); let app = Router::new() @@ -789,14 +803,17 @@ async fn sync_acknowledgements( if let Some(key) = reconnect_key { let attempts = sync.reconnects.entry(key).or_default(); *attempts += 1; - if *attempts == 1 { + if let Some(backoff) = usize::try_from(*attempts - 1) + .ok() + .and_then(|index| ACKNOWLEDGEMENT_RECONNECT_BACKOFF.get(index)) + { return Ok(Sse::new( stream::iter([Ok(Event::default() - .comment("reconnect") - .retry(Duration::from_millis(25)))]) + .comment(format!("reconnect-attempt-{attempts}")) + .retry(*backoff))]) .boxed(), ) - .keep_alive(KeepAlive::default())); + .keep_alive(KeepAlive::new().interval(state.acknowledgement_heartbeat_interval))); } } let first_available = sync @@ -817,7 +834,17 @@ async fn sync_acknowledgements( .unwrap_or_default(); let history_missing = after < latest && first_available.is_none_or(|first_sequence| after.saturating_add(1) < first_sequence); - let events = if history_missing { + let pending_count = sync + .acknowledgements + .values() + .filter(|acknowledgement| { + visible_acknowledgement(principal, acknowledgement) + && acknowledgement.server_sequence > after + && acknowledgement.server_sequence > sync.retained_after + }) + .count(); + let slow_consumer = pending_count > ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT; + let events = if history_missing || slow_consumer { vec![Ok(Event::default() .id(latest.to_string()) .event("snapshot-required") @@ -826,6 +853,9 @@ async fn sync_acknowledgements( "firstAvailable": first_available, "latest": latest, "snapshotUrl": "/sync/snapshot", + "reason": if slow_consumer { "slow-consumer" } else { "missing-history" }, + "pendingCount": pending_count, + "bufferLimit": ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT, })) .expect("serializable missing history event"))] } else { @@ -845,7 +875,24 @@ async fn sync_acknowledgements( }) .collect::>>() }; - Ok(Sse::new(stream::iter(events).boxed()).keep_alive(KeepAlive::default())) + drop(sync); + let heartbeat_interval = state.acknowledgement_heartbeat_interval; + let heartbeat = stream::unfold(heartbeat_interval, |interval| async move { + tokio::time::sleep(interval).await; + Some(( + Ok(Event::default() + .event("heartbeat") + .data("{\"status\":\"ok\"}")), + interval, + )) + }); + Ok( + Sse::new(stream::iter(events).chain(heartbeat).boxed()).keep_alive( + KeepAlive::new() + .interval(heartbeat_interval) + .text("heartbeat"), + ), + ) } fn registry(state: Arc) -> impl DispatchRegistry { diff --git a/examples/kanban/static/sync.js b/examples/kanban/static/sync.js index 2fca4f1..1c9d214 100644 --- a/examples/kanban/static/sync.js +++ b/examples/kanban/static/sync.js @@ -8,6 +8,7 @@ const MIGRATION_KEY = "commandSchemaMigration"; const MAX_ATTEMPTS = 3; const BACKOFF_MS = [25, 50]; const REQUEST_TIMEOUT_MS = 1_000; +const ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT = 64; const root = document.querySelector("[data-kanban-sync]"); const TAB_ID = sessionStorage.getItem("hemx-kanban-sync-tab-id") || crypto.randomUUID(); const LEASE_MS = 5000; @@ -481,6 +482,17 @@ async function synchronize(command) { source.addEventListener("open", () => { opens += 1; root.setAttribute("data-sync-transport-opens", String(opens)); + root.setAttribute("data-sync-stream-state", "open"); + }); + source.addEventListener("heartbeat", () => { + const heartbeats = Number(root.getAttribute("data-sync-heartbeats") || "0") + 1; + root.setAttribute("data-sync-heartbeats", String(heartbeats)); + root.setAttribute("data-sync-stream-state", "healthy"); + }); + source.addEventListener("error", () => { + const reconnects = Number(root.getAttribute("data-sync-reconnects") || "0") + 1; + root.setAttribute("data-sync-reconnects", String(reconnects)); + root.setAttribute("data-sync-stream-state", "reconnecting"); }); source.addEventListener("acknowledgement", async (event) => { const canonical = JSON.parse(event.data); @@ -631,6 +643,7 @@ async function runLeaseLoop(command) { async function start() { if (!root) return; root.setAttribute("data-sync-request-timeout-ms", String(REQUEST_TIMEOUT_MS)); + root.setAttribute("data-sync-stream-buffer-limit", String(ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT)); const contextResponse = await fetchWithTimeout("/sync/context", { credentials: "same-origin", cache: "no-store" }); if (!contextResponse.ok) throw new Error(`account context failed with ${contextResponse.status}`); const context = await contextResponse.json(); @@ -701,7 +714,10 @@ window.addEventListener("pagehide", () => { stopped = true; clearTimeout(leaseTimer); clearTimeout(retryTimer); - acknowledgementSource?.close(); + if (acknowledgementSource) { + acknowledgementSource.close(); + root.setAttribute("data-sync-stream-state", "cancelled"); + } acknowledgementSource = undefined; for (const controller of activeRequests) { controller.abort(new DOMException("sync cancelled because page is hidden", "AbortError")); diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index 6f90ae3..c618278 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/examples/kanban/tests/browser_e2e.rs @@ -2196,6 +2196,113 @@ async fn adversarial_wire_inputs_are_rejected_before_partial_application() -> We result.and(quit) } +#[tokio::test] +async fn acknowledgement_stream_bounds_reconnect_buffering_heartbeat_and_cancellation( +) -> WebDriverResult<()> { + // test req: operations/004 + 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_ACK_HEARTBEAT_MS", "25"); + let _app = TestProcess::start(app_command, "hemx-kanban", &app_addr, STARTUP_TIMEOUT) + .expect("start 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}/sync-demo")).await?; + wait_until( + &driver, + "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'idle'", + ) + .await?; + let proof = driver + .execute_async( + r#" + const done = arguments[arguments.length - 1]; + (async () => { + for (let index = 1; index <= 65; index += 1) { + const query = new URLSearchParams({ + command_id: `buffer:${index}`, + card_id: '1', + column: 'done', + }); + const response = await fetch(`/sync/commands?${query}`, { method: 'POST' }); + if (!response.ok) throw new Error(`command ${index} failed with ${response.status}`); + } + + const observe = (url, eventName, timeoutMs = 5000) => new Promise((resolve, reject) => { + const started = performance.now(); + const source = new EventSource(url); + let opens = 0; + let errors = 0; + const timeout = setTimeout(() => { + source.close(); + reject(new Error(`${eventName} timed out after ${timeoutMs} ms`)); + }, timeoutMs); + source.addEventListener('open', () => { opens += 1; }); + source.addEventListener('error', () => { errors += 1; }); + source.addEventListener(eventName, (event) => { + clearTimeout(timeout); + const data = JSON.parse(event.data); + source.close(); + resolve({ + data, + opens, + errors, + elapsedMs: performance.now() - started, + cancelled: source.readyState === EventSource.CLOSED, + }); + }); + }); + + const slowConsumer = await observe( + '/sync/acknowledgements?after=0&reconnect=slow-consumer-proof', + 'snapshot-required', + ); + const heartbeat = await observe( + '/sync/acknowledgements?after=65&reconnect=heartbeat-proof', + 'heartbeat', + ); + done({ slowConsumer, heartbeat }); + })().catch((error) => done({ error: String(error), stack: error?.stack })); + "#, + Vec::new(), + ) + .await? + .json() + .clone(); + assert!(proof["error"].is_null(), "stream bounds failed: {proof}"); + assert_eq!(proof["slowConsumer"]["data"]["reason"], "slow-consumer"); + assert_eq!(proof["slowConsumer"]["data"]["pendingCount"], 65); + assert_eq!(proof["slowConsumer"]["data"]["bufferLimit"], 64); + assert_eq!(proof["slowConsumer"]["opens"], 4); + assert!(proof["slowConsumer"]["errors"].as_u64().is_some_and(|errors| errors >= 3)); + assert!(proof["slowConsumer"]["elapsedMs"] + .as_f64() + .is_some_and(|elapsed| (700.0..5_000.0).contains(&elapsed))); + assert_eq!(proof["slowConsumer"]["cancelled"], true); + assert_eq!(proof["heartbeat"]["data"]["status"], "ok"); + assert_eq!(proof["heartbeat"]["opens"], 4); + assert!(proof["heartbeat"]["errors"].as_u64().is_some_and(|errors| errors >= 3)); + assert_eq!(proof["heartbeat"]["cancelled"], true); + Ok(()) + } + .await; + let quit = driver.quit().await; + result.and(quit) +} + #[tokio::test] async fn sync_requests_timeout_and_cancel_on_pagehide() -> WebDriverResult<()> { // test req: operations/003