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

feat(kanban): bound sync upload runs

Browse Source 
req: sync/017
This commit is contained in:
slhx agent
2026-07-13 17:48:47 +02:00
parent 0d475e81c8
commit 80f1ae155e
4 changed files with 169 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/kanban/static/sync.js
+58 -6
View File
@@ -8,9 +8,14 @@ const LEASE_MS = 5000;
const LEASE_POLL_MS = 100;
const LEASE_KEY = "uploaderLease";
let database;
let uploadLimit;
let retryTimer;
let leaseTimer;
let synchronizing = false;
let uploadsThisRun = 0;
let uploadedTotal = 0;
let inFlightUploads = 0;
let maxObservedInFlight = 0;
let stopped = false;
class UploadError extends Error {
@@ -168,6 +173,30 @@ async function upload(command) {
throw new Error("sync retry limit exhausted");
}
function beginUpload() {
inFlightUploads += 1;
maxObservedInFlight = Math.max(maxObservedInFlight, inFlightUploads);
root.setAttribute("data-sync-in-flight", String(inFlightUploads));
root.setAttribute("data-sync-max-observed-in-flight", String(maxObservedInFlight));
}
function finishUpload() {
inFlightUploads -= 1;
root.setAttribute("data-sync-in-flight", String(inFlightUploads));
}
async function continuePendingWork() {
const commands = await pendingCommands(database);
root.setAttribute("data-sync-pending-count", String(commands.length));
if (commands.length === 0) return;
if (uploadsThisRun >= uploadLimit) {
root.setAttribute("data-sync-manual-retry", "available");
setPhase("backpressured", `Upload limit ${uploadLimit} reached; ${commands.length} durable command${commands.length === 1 ? " remains" : "s remain"} queued. Retry now to continue.`);
return;
}
setTimeout(() => synchronize(validatePending(commands[0])).catch(failPermanently), 0);
}
async function synchronize(command) {
if (synchronizing) return;
synchronizing = true;
@@ -188,7 +217,13 @@ async function synchronize(command) {
root.removeAttribute("data-sync-manual-retry");
setOnline(navigator.onLine);
try {
const acknowledgement = await upload(command);
beginUpload();
let acknowledgement;
try {
acknowledgement = await upload(command);
} finally {
finishUpload();
}
setOnline(true);
root.setAttribute("data-sync-upload-sequence", String(acknowledgement.serverSequence));
setPhase("awaiting-ack", `Command ${command.id} uploaded; awaiting canonical acknowledgement.`);
@@ -203,8 +238,13 @@ async function synchronize(command) {
source.addEventListener("acknowledgement", async (event) => {
const canonical = JSON.parse(event.data);
if (canonical.commandId !== command.id) return;
source.close();
root.setAttribute("data-sync-pending-before-ack", String((await pendingCommands(database)).length));
await removeAcknowledged(database, command.id);
uploadsThisRun += 1;
uploadedTotal += 1;
root.setAttribute("data-sync-uploaded-this-run", String(uploadsThisRun));
root.setAttribute("data-sync-uploaded-total", String(uploadedTotal));
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);
@@ -214,7 +254,7 @@ async function synchronize(command) {
clearTimeout(leaseTimer);
await releaseUploaderLease(database);
root.setAttribute("data-sync-leader", "false");
source.close();
await continuePendingWork();
});
source.addEventListener("snapshot-required", async (event) => {
const missing = JSON.parse(event.data);
@@ -271,20 +311,32 @@ async function runLeaseLoop(command) {
async function start() {
if (!root) return;
uploadLimit = Number.parseInt(root.getAttribute("data-sync-upload-limit"), 10);
if (!Number.isSafeInteger(uploadLimit) || uploadLimit < 1) throw new Error("data-sync-upload-limit must be a positive integer");
database = await openLog();
const commands = await pendingCommands(database);
root.setAttribute("data-sync-uploaded-this-run", "0");
root.setAttribute("data-sync-uploaded-total", "0");
root.setAttribute("data-sync-in-flight", "0");
root.setAttribute("data-sync-max-observed-in-flight", "0");
root.setAttribute("data-sync-pending-count", String(commands.length));
if (commands.length === 0) {
setPhase("idle", "No pending commands.");
return;
}
const command = validatePending(commands[0]);
root.addEventListener("click", (event) => {
root.addEventListener("click", async (event) => {
if (!event.target.closest("[data-sync-retry]")) return;
synchronize(command).catch(failPermanently);
uploadsThisRun = 0;
root.setAttribute("data-sync-uploaded-this-run", "0");
root.removeAttribute("data-sync-manual-retry");
const [next] = await pendingCommands(database);
if (next) synchronize(validatePending(next)).catch(failPermanently);
});
window.addEventListener("online", () => {
if (root.getAttribute("data-sync-phase") === "offline") synchronize(command).catch(failPermanently);
window.addEventListener("online", async () => {
if (root.getAttribute("data-sync-phase") !== "offline") return;
const [next] = await pendingCommands(database);
if (next) synchronize(validatePending(next)).catch(failPermanently);
});
await runLeaseLoop(command);
}
examples/kanban/templates/sync_shell.heml
+1 -1
View File
@@ -6,7 +6,7 @@
<title>hemx Kanban sync</title>
</head>
<body>
<section data-kanban-sync data-sync-version="1" aria-labelledby="sync-title">
<section data-kanban-sync data-sync-version="1" data-sync-upload-limit="2" 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>
examples/kanban/tests/browser_e2e.rs
+108
View File
@@ -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<()> {