feat(kanban): upload pending command with retry
req: sync/004\nreq: sync/009\nreq: sync/010\nreq: sync/016\nreq: sync/017
This commit is contained in:
slhx agent
@@ -9,13 +9,13 @@ use hemplate::Hemplate;
|
||||
use hemx::{Html, IntoEffect};
|
||||
use hemx_axum::{
|
||||
interactions, runtime_js, runtime_js_path, sse, DispatchRegistry, DispatchRejection,
|
||||
EffectResponse, InteractionRequest, PageRequest,
|
||||
EffectResponse, InteractionRequest, PageRequest, PageResponse,
|
||||
};
|
||||
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::collections::{BTreeMap, BTreeSet};
|
||||
use std::convert::Infallible;
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::{Arc, Mutex};
|
||||
@@ -34,6 +34,8 @@ struct SyncState {
|
||||
next_sequence: u64,
|
||||
acknowledgements: BTreeMap<CommandId, SyncAcknowledgement>,
|
||||
reconnects: BTreeMap<String, u64>,
|
||||
fail_first_upload: bool,
|
||||
transient_failures: BTreeSet<CommandId>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
|
||||
@@ -68,6 +70,7 @@ struct SyncAcknowledgement {
|
||||
enum SyncRejection {
|
||||
BadRequest(&'static str),
|
||||
Conflict(&'static str),
|
||||
Transient,
|
||||
}
|
||||
|
||||
impl IntoResponse for SyncRejection {
|
||||
@@ -75,6 +78,7 @@ impl IntoResponse for SyncRejection {
|
||||
let (status, error) = match self {
|
||||
Self::BadRequest(error) => (StatusCode::BAD_REQUEST, error),
|
||||
Self::Conflict(error) => (StatusCode::CONFLICT, error),
|
||||
Self::Transient => (StatusCode::SERVICE_UNAVAILABLE, "transient sync failure"),
|
||||
};
|
||||
(status, Json(serde_json::json!({ "error": error }))).into_response()
|
||||
}
|
||||
@@ -99,6 +103,11 @@ struct AppShell {
|
||||
body: Html,
|
||||
}
|
||||
|
||||
#[derive(Hemplate)]
|
||||
struct SyncShell {
|
||||
runtime_src: &'static str,
|
||||
}
|
||||
|
||||
#[derive(Hemplate)]
|
||||
#[hemplate = "partials"]
|
||||
struct BoardColumns {
|
||||
@@ -171,6 +180,7 @@ async fn main() {
|
||||
}),
|
||||
sync: Mutex::new(SyncState {
|
||||
next_sequence: 1,
|
||||
fail_first_upload: std::env::var_os("HEMX_KANBAN_FAIL_FIRST_SYNC").is_some(),
|
||||
..SyncState::default()
|
||||
}),
|
||||
});
|
||||
@@ -178,6 +188,8 @@ async fn main() {
|
||||
let app = Router::new()
|
||||
.route("/", get(home).post(interact))
|
||||
.route("/events", get(events))
|
||||
.route("/sync-demo", get(sync_demo))
|
||||
.route("/sync.js", get(sync_js))
|
||||
.route("/sync/commands", post(sync_command))
|
||||
.route("/sync/acknowledgements", get(sync_acknowledgements))
|
||||
.route(runtime_js_path(), get(runtime))
|
||||
@@ -204,6 +216,24 @@ async fn runtime() -> impl IntoResponse {
|
||||
runtime_js()
|
||||
}
|
||||
|
||||
async fn sync_demo() -> impl IntoResponse {
|
||||
PageResponse::full(
|
||||
ui::page(&SyncShell {
|
||||
runtime_src: runtime_js_path(),
|
||||
})
|
||||
.into_string(),
|
||||
)
|
||||
.title("hemx Kanban sync")
|
||||
.fingerprint(ui::BUILD_FINGERPRINT)
|
||||
}
|
||||
|
||||
async fn sync_js() -> impl IntoResponse {
|
||||
(
|
||||
[("content-type", "text/javascript; charset=utf-8")],
|
||||
include_str!("../static/sync.js"),
|
||||
)
|
||||
}
|
||||
|
||||
async fn interact(
|
||||
State(state): State<Arc<AppState>>,
|
||||
request: InteractionRequest,
|
||||
@@ -254,6 +284,9 @@ async fn sync_command(
|
||||
}
|
||||
return Ok(Json(existing.clone()));
|
||||
}
|
||||
if sync.fail_first_upload && sync.transient_failures.insert(command_id.clone()) {
|
||||
return Err(SyncRejection::Transient);
|
||||
}
|
||||
|
||||
let mut board = state.board.lock().unwrap();
|
||||
let card = board
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
const DATABASE = "hemx-kanban-v1";
|
||||
const COMMANDS = "commands";
|
||||
const MAX_ATTEMPTS = 3;
|
||||
const BACKOFF_MS = [25, 50];
|
||||
const root = document.querySelector("[data-kanban-sync]");
|
||||
|
||||
class UploadError extends Error {
|
||||
constructor(status, retryable) {
|
||||
super(`sync upload failed with ${status}`);
|
||||
this.name = "UploadError";
|
||||
this.retryable = retryable;
|
||||
}
|
||||
}
|
||||
|
||||
function requestResult(request) {
|
||||
return new Promise((resolve, reject) => {
|
||||
request.addEventListener("success", () => resolve(request.result), { once: true });
|
||||
request.addEventListener("error", () => reject(request.error || new Error("IndexedDB request failed")), { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
function transactionDone(transaction) {
|
||||
return new Promise((resolve, reject) => {
|
||||
transaction.addEventListener("complete", resolve, { once: true });
|
||||
transaction.addEventListener("abort", () => reject(transaction.error || new Error("IndexedDB transaction aborted")), { once: true });
|
||||
transaction.addEventListener("error", () => reject(transaction.error || new Error("IndexedDB transaction failed")), { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
async function openLog() {
|
||||
const request = indexedDB.open(DATABASE, 1);
|
||||
request.addEventListener("upgradeneeded", () => {
|
||||
const database = request.result;
|
||||
if (!database.objectStoreNames.contains(COMMANDS)) database.createObjectStore(COMMANDS, { keyPath: "id" });
|
||||
if (!database.objectStoreNames.contains("meta")) database.createObjectStore("meta");
|
||||
});
|
||||
return requestResult(request);
|
||||
}
|
||||
|
||||
async function pendingCommands(database) {
|
||||
const transaction = database.transaction(COMMANDS, "readonly");
|
||||
const done = transactionDone(transaction);
|
||||
const commands = await requestResult(transaction.objectStore(COMMANDS).getAll());
|
||||
await done;
|
||||
return commands.sort((left, right) => left.causal - right.causal);
|
||||
}
|
||||
|
||||
async function removeAcknowledged(database, commandId) {
|
||||
const transaction = database.transaction(COMMANDS, "readwrite");
|
||||
const done = transactionDone(transaction);
|
||||
transaction.objectStore(COMMANDS).delete(commandId);
|
||||
await done;
|
||||
}
|
||||
|
||||
function setPhase(phase, message) {
|
||||
root.setAttribute("data-sync-phase", phase);
|
||||
root.querySelector('[role="status"]').textContent = message;
|
||||
}
|
||||
|
||||
function validatePending(command) {
|
||||
if (!command || command.schemaVersion !== 1 || command.kind !== "reorder_card" || typeof command.id !== "string" || !command.id || typeof command.cardId !== "string" || !command.cardId) {
|
||||
throw new Error("invalid pending command");
|
||||
}
|
||||
return command;
|
||||
}
|
||||
|
||||
async function upload(command) {
|
||||
root.setAttribute("data-sync-max-attempts", String(MAX_ATTEMPTS));
|
||||
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt += 1) {
|
||||
root.setAttribute("data-sync-attempts", String(attempt));
|
||||
setPhase(attempt === 1 ? "uploading" : "retrying", `Uploading ${command.id} (attempt ${attempt} of ${MAX_ATTEMPTS}).`);
|
||||
try {
|
||||
const query = new URLSearchParams({ command_id: command.id, card_id: command.cardId });
|
||||
const response = await fetch(`/sync/commands?${query}`, { method: "POST" });
|
||||
if (response.status === 503 && attempt < MAX_ATTEMPTS) {
|
||||
const base = BACKOFF_MS[attempt - 1];
|
||||
const delay = base + Math.floor(Math.random() * base);
|
||||
root.setAttribute("data-sync-last-backoff-base-ms", String(base));
|
||||
root.setAttribute("data-sync-last-backoff-ms", String(delay));
|
||||
root.dispatchEvent(new CustomEvent("kanban:sync-retry", { detail: { attempt, base, delay } }));
|
||||
await new Promise((resolve) => setTimeout(resolve, delay));
|
||||
continue;
|
||||
}
|
||||
if (!response.ok) throw new UploadError(response.status, response.status >= 500);
|
||||
return response.json();
|
||||
} catch (error) {
|
||||
if (error instanceof UploadError && !error.retryable) throw error;
|
||||
if (attempt === MAX_ATTEMPTS) throw error;
|
||||
const base = BACKOFF_MS[attempt - 1];
|
||||
const delay = base + Math.floor(Math.random() * base);
|
||||
root.setAttribute("data-sync-last-backoff-base-ms", String(base));
|
||||
root.setAttribute("data-sync-last-backoff-ms", String(delay));
|
||||
root.dispatchEvent(new CustomEvent("kanban:sync-retry", { detail: { attempt, base, delay } }));
|
||||
await new Promise((resolve) => setTimeout(resolve, delay));
|
||||
}
|
||||
}
|
||||
throw new Error("sync retry limit exhausted");
|
||||
}
|
||||
|
||||
async function start() {
|
||||
if (!root) return;
|
||||
const database = await openLog();
|
||||
const commands = await pendingCommands(database);
|
||||
root.setAttribute("data-sync-pending-count", String(commands.length));
|
||||
if (commands.length === 0) {
|
||||
setPhase("idle", "No pending commands.");
|
||||
return;
|
||||
}
|
||||
const command = validatePending(commands[0]);
|
||||
const acknowledgement = await upload(command);
|
||||
root.setAttribute("data-sync-upload-sequence", String(acknowledgement.serverSequence));
|
||||
setPhase("awaiting-ack", `Command ${command.id} uploaded; awaiting canonical acknowledgement.`);
|
||||
|
||||
const reconnect = command.session || command.actor || "kanban";
|
||||
const source = new EventSource(`/sync/acknowledgements?after=0&reconnect=${encodeURIComponent(reconnect)}`);
|
||||
let opens = 0;
|
||||
source.addEventListener("open", () => {
|
||||
opens += 1;
|
||||
root.setAttribute("data-sync-transport-opens", String(opens));
|
||||
});
|
||||
source.addEventListener("acknowledgement", async (event) => {
|
||||
const canonical = JSON.parse(event.data);
|
||||
if (canonical.commandId !== command.id) return;
|
||||
root.setAttribute("data-sync-pending-before-ack", String((await pendingCommands(database)).length));
|
||||
await removeAcknowledged(database, command.id);
|
||||
root.setAttribute("data-sync-pending-count", String((await pendingCommands(database)).length));
|
||||
root.setAttribute("data-sync-ack-sequence", String(canonical.serverSequence));
|
||||
root.setAttribute("data-sync-canonical-column", canonical.canonicalColumn);
|
||||
setPhase("acknowledged", `Command ${command.id} acknowledged in ${canonical.canonicalColumn}.`);
|
||||
root.dispatchEvent(new CustomEvent("kanban:sync-acknowledged", { detail: canonical }));
|
||||
source.close();
|
||||
});
|
||||
}
|
||||
|
||||
start().catch((error) => {
|
||||
if (!root) return;
|
||||
root.setAttribute("data-sync-error", error instanceof Error ? error.message : String(error));
|
||||
setPhase("failed", "Sync failed; the durable command remains queued.");
|
||||
});
|
||||
@@ -0,0 +1,16 @@
|
||||
<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>hemx Kanban sync</title>
|
||||
</head>
|
||||
<body>
|
||||
<section data-kanban-sync data-sync-version="1" aria-labelledby="sync-title">
|
||||
<h2 id="sync-title">Sync status</h2>
|
||||
<p role="status" aria-live="polite">Waiting for pending commands.</p>
|
||||
</section>
|
||||
<script +src="self.runtime_src" defer></script>
|
||||
<script src="/sync.js" defer></script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -203,6 +203,171 @@ async fn idempotent_server_command_is_acknowledged_after_reconnect() -> WebDrive
|
||||
result.and(quit)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
|
||||
) -> WebDriverResult<()> {
|
||||
// test req: sync/004 req: sync/009 req: sync/010 req: sync/016 req: sync/017
|
||||
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)
|
||||
.env("HEMX_KANBAN_FAIL_FIRST_SYNC", "1");
|
||||
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?;
|
||||
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.onerror = () => done({ error: open.error && open.error.name });
|
||||
open.onsuccess = () => {
|
||||
const tx = open.result.transaction('commands', 'readwrite');
|
||||
tx.objectStore('commands').add({
|
||||
id: 'sync-actor:1', schemaVersion: 1, actor: 'sync-actor', session: 'sync-session',
|
||||
causal: 1, kind: 'reorder_card', cardId: '1', 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 durable command: {seeded}");
|
||||
|
||||
driver.goto(&format!("http://{app_addr}/sync-demo")).await?;
|
||||
wait_until(
|
||||
&driver,
|
||||
"return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'acknowledged' || document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'failed'",
|
||||
)
|
||||
.await?;
|
||||
let proof = driver
|
||||
.execute(
|
||||
"const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), pending: root.getAttribute('data-sync-pending-count'), pendingBeforeAck: root.getAttribute('data-sync-pending-before-ack'), attempts: root.getAttribute('data-sync-attempts'), maxAttempts: root.getAttribute('data-sync-max-attempts'), backoffBase: root.getAttribute('data-sync-last-backoff-base-ms'), backoff: root.getAttribute('data-sync-last-backoff-ms'), opens: root.getAttribute('data-sync-transport-opens'), uploadSequence: root.getAttribute('data-sync-upload-sequence'), ackSequence: root.getAttribute('data-sync-ack-sequence'), 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!(proof["phase"], "acknowledged", "sync failed: {proof}");
|
||||
assert_eq!(proof["pending"], "0");
|
||||
assert_eq!(proof["pendingBeforeAck"], "1");
|
||||
assert_eq!(proof["attempts"], "2");
|
||||
assert_eq!(proof["maxAttempts"], "3");
|
||||
assert_eq!(proof["backoffBase"], "25");
|
||||
assert!(
|
||||
proof["backoff"]
|
||||
.as_str()
|
||||
.and_then(|value| value.parse::<u64>().ok())
|
||||
.is_some_and(|delay| (25..50).contains(&delay)),
|
||||
"retry jitter left its bounded interval: {proof}"
|
||||
);
|
||||
assert!(
|
||||
proof["opens"].as_str().and_then(|value| value.parse::<u64>().ok()).is_some_and(|opens| opens >= 2),
|
||||
"transport did not reconnect: {proof}"
|
||||
);
|
||||
assert_eq!(proof["uploadSequence"], "1");
|
||||
assert_eq!(proof["ackSequence"], "1");
|
||||
assert_eq!(proof["canonicalColumn"], "done");
|
||||
assert_eq!(proof["status"], "Command sync-actor:1 acknowledged in done.");
|
||||
assert!(proof["error"].is_null());
|
||||
|
||||
let queue_count = driver
|
||||
.execute_async(
|
||||
r#"
|
||||
const done = arguments[arguments.length - 1];
|
||||
const open = indexedDB.open('hemx-kanban-v1', 1);
|
||||
open.onsuccess = () => {
|
||||
const request = open.result.transaction('commands', 'readonly').objectStore('commands').count();
|
||||
request.onsuccess = () => done(request.result);
|
||||
request.onerror = () => done({ error: request.error && request.error.name });
|
||||
};
|
||||
"#,
|
||||
Vec::new(),
|
||||
)
|
||||
.await?
|
||||
.json()
|
||||
.clone();
|
||||
assert_eq!(queue_count, 0);
|
||||
|
||||
let rejected_seed = 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('commands', 'readwrite');
|
||||
tx.objectStore('commands').add({
|
||||
id: 'sync-actor:1', schemaVersion: 1, actor: 'sync-actor', session: 'sync-session',
|
||||
causal: 2, kind: 'reorder_card', cardId: '2', 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!(rejected_seed["seeded"], true, "failed to seed rejected command");
|
||||
driver.goto(&format!("http://{app_addr}/sync-demo")).await?;
|
||||
wait_until(
|
||||
&driver,
|
||||
"return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'failed'",
|
||||
)
|
||||
.await?;
|
||||
let rejected = driver
|
||||
.execute(
|
||||
"const root = document.querySelector('[data-kanban-sync]'); return { pending: root.getAttribute('data-sync-pending-count'), attempts: root.getAttribute('data-sync-attempts'), error: root.getAttribute('data-sync-error'), status: root.querySelector('[role=status]').textContent }",
|
||||
Vec::new(),
|
||||
)
|
||||
.await?
|
||||
.json()
|
||||
.clone();
|
||||
assert_eq!(rejected["pending"], "1");
|
||||
assert_eq!(rejected["attempts"], "1");
|
||||
assert_eq!(rejected["error"], "sync upload failed with 409");
|
||||
assert_eq!(rejected["status"], "Sync failed; the durable command remains queued.");
|
||||
|
||||
driver.goto(&format!("http://{app_addr}/")).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) {
|
||||
@@ -210,7 +375,15 @@ async fn wait_until(driver: &WebDriver, script: &str) -> WebDriverResult<()> {
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(25)).await;
|
||||
}
|
||||
panic!("browser condition timed out: {script}");
|
||||
let snapshot = driver
|
||||
.execute(
|
||||
"const sync = document.querySelector('[data-kanban-sync]'); return { url: location.href, body: document.body.textContent, sync: sync ? Object.fromEntries([...sync.attributes].map((attribute) => [attribute.name, attribute.value])) : null }",
|
||||
Vec::new(),
|
||||
)
|
||||
.await?
|
||||
.json()
|
||||
.clone();
|
||||
panic!("browser condition timed out: {script}; snapshot: {snapshot}");
|
||||
}
|
||||
|
||||
fn available_port() -> u16 {
|
||||
|
||||
Reference in New Issue
Block a user