Waiting for pending commands.
diff --git a/examples/kanban/tests/browser_e2e.rs b/examples/kanban/tests/browser_e2e.rs index 8367b12..134ab6d 100644 --- a/examples/kanban/tests/browser_e2e.rs +++ b/examples/kanban/tests/browser_e2e.rs @@ -369,6 +369,114 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack( result.and(quit) } +#[tokio::test] +async fn upload_backpressure_keeps_pending_work_visible_and_recoverable() -> WebDriverResult<()> { + // test req: sync/017 + 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); + 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 = 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'); + const commands = tx.objectStore('commands'); + for (let causal = 1; causal <= 3; causal += 1) { + commands.add({ + id: `pressure:${causal}`, schemaVersion: 1, actor: 'pressure', session: 'pressure-session', + causal, kind: 'reorder_card', cardId: String(causal), 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["seeded"], true, "failed to seed pending work: {seeded}"); + + driver.goto(&format!("http://{app_addr}/sync-demo")).await?; + wait_until( + &driver, + "return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'backpressured'", + ) + .await?; + let bounded = driver + .execute( + "const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), pending: root.getAttribute('data-sync-pending-count'), limit: root.getAttribute('data-sync-upload-limit'), uploaded: root.getAttribute('data-sync-uploaded-this-run'), total: root.getAttribute('data-sync-uploaded-total'), maxInFlight: root.getAttribute('data-sync-max-observed-in-flight'), sequence: root.getAttribute('data-sync-ack-sequence'), manual: root.getAttribute('data-sync-manual-retry'), status: root.querySelector('[role=status]').textContent }", + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(bounded["phase"], "backpressured"); + assert_eq!(bounded["pending"], "1"); + assert_eq!(bounded["limit"], "2"); + assert_eq!(bounded["uploaded"], "2"); + assert_eq!(bounded["total"], "2"); + assert_eq!(bounded["maxInFlight"], "1"); + assert_eq!(bounded["sequence"], "2"); + assert_eq!(bounded["manual"], "available"); + assert_eq!( + bounded["status"], + "Upload limit 2 reached; 1 durable command remains queued. Retry now to continue." + ); + assert_eq!(command_count(&driver).await?, 1); + + driver.find(By::Css("[data-sync-retry]")).await?.click().await?; + wait_until( + &driver, + "const root = document.querySelector('[data-kanban-sync]'); return root?.getAttribute('data-sync-phase') === 'acknowledged' && root?.getAttribute('data-sync-pending-count') === '0'", + ) + .await?; + let recovered = driver + .execute( + "const root = document.querySelector('[data-kanban-sync]'); return { pending: root.getAttribute('data-sync-pending-count'), uploaded: root.getAttribute('data-sync-uploaded-this-run'), total: root.getAttribute('data-sync-uploaded-total'), maxInFlight: root.getAttribute('data-sync-max-observed-in-flight'), sequence: root.getAttribute('data-sync-ack-sequence'), status: root.querySelector('[role=status]').textContent }", + Vec::new(), + ) + .await? + .json() + .clone(); + assert_eq!(recovered["pending"], "0"); + assert_eq!(recovered["uploaded"], "1"); + assert_eq!(recovered["total"], "3"); + assert_eq!(recovered["maxInFlight"], "1"); + assert_eq!(recovered["sequence"], "3"); + assert_eq!(recovered["status"], "Command pressure:3 acknowledged in done."); + assert_eq!(command_count(&driver).await?, 0); + Ok(()) + } + .await; + let quit = driver.quit().await; + result.and(quit) +} + #[tokio::test] async fn two_tabs_coordinate_single_uploader_and_takeover_without_duplicate_application( ) -> WebDriverResult<()> {