tmk241/hemx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(kanban): recover after retry exhaustion

Browse Source 
req: sync/004\nreq: sync/010\nreq: sync/011\nreq: sync/014\nreq: sync/016
This commit is contained in:
slhx agent
2026-07-13 16:47:52 +02:00
parent 0a7c380187
commit 5ea747d968
5 changed files with 229 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/kanban/src/main.rs
+17 -6
View File
@@ -15,7 +15,7 @@ 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::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::collections::BTreeMap;
use std::convert::Infallible;
use std::fs::{self, OpenOptions};
use std::io::Write;
@@ -41,8 +41,8 @@ struct SyncState {
next_sequence: u64,
acknowledgements: BTreeMap<CommandId, SyncAcknowledgement>,
reconnects: BTreeMap<String, u64>,
fail_first_upload: bool,
transient_failures: BTreeSet<CommandId>,
transient_failure_limit: u8,
transient_failures: BTreeMap<CommandId, u8>,
}
#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
@@ -294,7 +294,10 @@ async fn main() {
next_sequence: 1,
..SyncState::default()
});
sync.fail_first_upload = std::env::var_os("HEMX_KANBAN_FAIL_FIRST_SYNC").is_some();
sync.transient_failure_limit = std::env::var("HEMX_KANBAN_SYNC_FAILURES")
.ok()
.and_then(|value| value.parse::<u8>().ok())
.unwrap_or_else(|| u8::from(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
@@ -433,8 +436,16 @@ 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 transient_failure_limit = sync.transient_failure_limit;
if transient_failure_limit > 0 {
let failures = sync
.transient_failures
.entry(command_id.clone())
.or_default();
if *failures < transient_failure_limit {
*failures += 1;
return Err(SyncRejection::Transient);
}
}
let mut board = state.board.lock().unwrap();
examples/kanban/static/sync.js
+64 -25
View File
@@ -3,6 +3,8 @@ const COMMANDS = "commands";
const MAX_ATTEMPTS = 3;
const BACKOFF_MS = [25, 50];
const root = document.querySelector("[data-kanban-sync]");
let database;
let retryTimer;
class UploadError extends Error {
constructor(status, retryable) {
@@ -64,6 +66,18 @@ function validatePending(command) {
return command;
}
function setOnline(online) {
root.setAttribute("data-sync-connection", online ? "online" : "offline");
}
function scheduleManualRetry(command, error) {
clearTimeout(retryTimer);
root.setAttribute("data-sync-error", error instanceof Error ? error.message : String(error));
root.setAttribute("data-sync-manual-retry", "available");
setPhase("offline", "Sync is offline after bounded retries; the durable command remains queued. Retry now when ready.");
root.dispatchEvent(new CustomEvent("kanban:sync-exhausted", { detail: { commandId: command.id, attempts: MAX_ATTEMPTS } }));
}
async function upload(command) {
root.setAttribute("data-sync-max-attempts", String(MAX_ATTEMPTS));
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt += 1) {
@@ -97,9 +111,45 @@ async function upload(command) {
throw new Error("sync retry limit exhausted");
}
async function synchronize(command) {
root.removeAttribute("data-sync-error");
root.removeAttribute("data-sync-manual-retry");
setOnline(navigator.onLine);
try {
const acknowledgement = await upload(command);
setOnline(true);
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();
});
} catch (error) {
setOnline(false);
if (error instanceof UploadError && !error.retryable) throw error;
scheduleManualRetry(command, error);
}
}
async function start() {
if (!root) return;
const database = await openLog();
database = await openLog();
const commands = await pendingCommands(database);
root.setAttribute("data-sync-pending-count", String(commands.length));
if (commands.length === 0) {
@@ -107,33 +157,22 @@ async function start() {
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.`);
root.addEventListener("click", (event) => {
if (!event.target.closest("[data-sync-retry]")) return;
synchronize(command).catch(failPermanently);
});
window.addEventListener("online", () => {
if (root.getAttribute("data-sync-phase") === "offline") synchronize(command).catch(failPermanently);
});
await synchronize(command);
}
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();
});
function failPermanently(error) {
root.setAttribute("data-sync-error", error instanceof Error ? error.message : String(error));
setPhase("failed", "Sync failed; the durable command remains queued.");
}
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.");
failPermanently(error);
});
examples/kanban/templates/sync_shell.heml
+1
View File
@@ -9,6 +9,7 @@
<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>
<button type="button" data-sync-retry>Retry sync now</button>
</section>
<script +src="self.runtime_src" defer></script>
<script src="/sync.js" defer></script>
examples/kanban/tests/browser_e2e.rs
+145
View File
@@ -369,6 +369,131 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
result.and(quit)
}
#[tokio::test]
async fn exhausted_offline_retries_keep_command_until_later_reconnect() -> WebDriverResult<()> {
// test req: sync/004 req: sync/010 req: sync/011 req: sync/014 req: sync/016
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_SYNC_FAILURES", "3");
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');
tx.objectStore('commands').add({
id: 'offline-actor:1', schemaVersion: 1, actor: 'offline-actor', session: 'offline-session',
causal: 1, 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!(seeded["seeded"], true, "failed to seed offline command: {seeded}");
driver.goto(&format!("http://{app_addr}/sync-demo")).await?;
wait_until(
&driver,
"return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'offline'",
)
.await?;
let exhausted = driver
.execute(
"const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), connection: root.getAttribute('data-sync-connection'), pending: root.getAttribute('data-sync-pending-count'), attempts: root.getAttribute('data-sync-attempts'), maxAttempts: root.getAttribute('data-sync-max-attempts'), manual: root.getAttribute('data-sync-manual-retry'), error: root.getAttribute('data-sync-error'), status: root.querySelector('[role=status]').textContent, retry: root.querySelector('[data-sync-retry]').textContent }",
Vec::new(),
)
.await?
.json()
.clone();
assert_eq!(exhausted["phase"], "offline");
assert_eq!(exhausted["connection"], "offline");
assert_eq!(exhausted["pending"], "1");
assert_eq!(exhausted["attempts"], "3");
assert_eq!(exhausted["maxAttempts"], "3");
assert_eq!(exhausted["manual"], "available");
assert_eq!(exhausted["error"], "sync upload failed with 503");
assert_eq!(
exhausted["status"],
"Sync is offline after bounded retries; the durable command remains queued. Retry now when ready."
);
assert_eq!(exhausted["retry"], "Retry sync now");
let queued = command_count(&driver).await?;
assert_eq!(queued, 1);
driver
.find(By::Css("[data-sync-retry]"))
.await?
.click()
.await?;
wait_until(
&driver,
"return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'acknowledged'",
)
.await?;
let converged = driver
.execute(
"const root = document.querySelector('[data-kanban-sync]'); return { connection: root.getAttribute('data-sync-connection'), pending: root.getAttribute('data-sync-pending-count'), attempts: root.getAttribute('data-sync-attempts'), sequence: root.getAttribute('data-sync-ack-sequence'), column: 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!(converged["connection"], "online");
assert_eq!(converged["pending"], "0");
assert_eq!(converged["attempts"], "1");
assert_eq!(converged["sequence"], "1");
assert_eq!(converged["column"], "done");
assert_eq!(converged["status"], "Command offline-actor:1 acknowledged in done.");
assert!(converged["error"].is_null());
assert_eq!(command_count(&driver).await?, 0);
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!(["2", "3"]));
Ok(())
}
.await;
let quit = driver.quit().await;
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
@@ -497,6 +622,26 @@ async fn canonical_acknowledgement_survives_server_restart() -> WebDriverResult<
result.and(quit)
}
async fn command_count(driver: &WebDriver) -> WebDriverResult<u64> {
let 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();
Ok(count.as_u64().expect("durable command count"))
}
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) {