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

fix(sync): bound request lifetime

Browse Source 
req: operations/003
This commit is contained in:
slhx agent
2026-07-13 21:16:07 +02:00
parent 6d502d8ff5
commit 2a8aa4c07f
3 changed files with 118 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/kanban/static/sync.js
+40 -3
View File
@@ -7,6 +7,7 @@ const LEGACY_COMMAND_SCHEMA = 1;
const MIGRATION_KEY = "commandSchemaMigration";
const MAX_ATTEMPTS = 3;
const BACKOFF_MS = [25, 50];
const REQUEST_TIMEOUT_MS = 1_000;
const root = document.querySelector("[data-kanban-sync]");
const TAB_ID = sessionStorage.getItem("hemx-kanban-sync-tab-id") || crypto.randomUUID();
const LEASE_MS = 5000;
@@ -16,6 +17,8 @@ let accountPartition;
let uploadLimit;
let retryTimer;
let leaseTimer;
let acknowledgementSource;
const activeRequests = new Set();
let synchronizing = false;
let uploadsThisRun = 0;
let uploadedTotal = 0;
@@ -39,6 +42,30 @@ class UploadError extends Error {
}
}
// req: operations/003
export async function fetchWithTimeout(
input,
init = {},
fetchImplementation = fetch,
timeoutMs = REQUEST_TIMEOUT_MS,
) {
if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 1) {
throw new TypeError("sync request timeout must be a positive integer");
}
const controller = new AbortController();
const timeout = setTimeout(
() => controller.abort(new DOMException(`sync request timed out after ${timeoutMs} ms`, "TimeoutError")),
timeoutMs,
);
activeRequests.add(controller);
try {
return await fetchImplementation(input, { ...init, signal: controller.signal });
} finally {
clearTimeout(timeout);
activeRequests.delete(controller);
}
}
function requestResult(request) {
return new Promise((resolve, reject) => {
request.addEventListener("success", () => resolve(request.result), { once: true });
@@ -322,7 +349,7 @@ async function upload(command) {
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, column: command.targetColumn });
const response = await fetch(`/sync/commands?${query}`, { method: "POST" });
const response = await fetchWithTimeout(`/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);
@@ -413,6 +440,7 @@ async function synchronize(command) {
const reconnect = command.session || command.actor || "kanban";
const source = new EventSource(`/sync/acknowledgements?after=0&reconnect=${encodeURIComponent(reconnect)}`);
acknowledgementSource = source;
let opens = 0;
source.addEventListener("open", () => {
opens += 1;
@@ -422,6 +450,7 @@ async function synchronize(command) {
const canonical = JSON.parse(event.data);
if (canonical.commandId !== command.id) return;
source.close();
if (acknowledgementSource === source) acknowledgementSource = undefined;
root.setAttribute("data-sync-pending-before-ack", String((await pendingCommands(database)).length));
const queueCommandId = command.queueCommandId || command.id;
await removePendingCommand(database, queueCommandId);
@@ -454,7 +483,7 @@ async function synchronize(command) {
});
source.addEventListener("snapshot-required", async (event) => {
const missing = JSON.parse(event.data);
const response = await fetch(missing.snapshotUrl);
const response = await fetchWithTimeout(missing.snapshotUrl);
if (!response.ok) throw new Error(`snapshot failed with ${response.status}`);
const snapshot = await response.json();
const queued = await pendingCommands(database);
@@ -497,6 +526,7 @@ async function synchronize(command) {
}
synchronizing = false;
source.close();
if (acknowledgementSource === source) acknowledgementSource = undefined;
clearTimeout(leaseTimer);
await releaseUploaderLease(database);
root.setAttribute("data-sync-leader", "false");
@@ -564,7 +594,8 @@ async function runLeaseLoop(command) {
async function start() {
if (!root) return;
const contextResponse = await fetch("/sync/context", { credentials: "same-origin", cache: "no-store" });
root.setAttribute("data-sync-request-timeout-ms", String(REQUEST_TIMEOUT_MS));
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();
if (!context || typeof context.accountPartition !== "string" || !context.accountPartition) {
@@ -633,6 +664,12 @@ async function start() {
window.addEventListener("pagehide", () => {
stopped = true;
clearTimeout(leaseTimer);
clearTimeout(retryTimer);
acknowledgementSource?.close();
acknowledgementSource = undefined;
for (const controller of activeRequests) {
controller.abort(new DOMException("sync cancelled because page is hidden", "AbortError"));
}
if (database) releaseUploaderLease(database).catch(() => {});
});
examples/kanban/tests/browser_e2e.rs
+77 -2
View File
@@ -599,7 +599,7 @@ async fn account_partition_hides_replay_and_export_until_owner_returns() -> WebD
#[tokio::test]
async fn canonical_snapshot_and_history_are_tenant_scoped() -> WebDriverResult<()> {
// test req: sync/007 req: security/004 req: auth/005
// test req: sync/007 req: security/004 req: auth/005 req: performance/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"));
@@ -1708,7 +1708,7 @@ async fn missing_history_rebase_and_user_conflict_resolution_preserve_suffix() -
#[tokio::test]
async fn redacted_sync_diagnostics_are_bounded_and_leak_no_sensitive_material(
) -> WebDriverResult<()> {
// test req: operations/001 req: security/002 req: sync/016 req: sync/021
// test req: operations/001 req: operations/005 req: security/002 req: sync/016 req: sync/021
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"));
@@ -2077,6 +2077,81 @@ async fn keep_local_retry_preserves_conflicted_command_and_suffix_order() -> Web
result.and(quit)
}
#[tokio::test]
async fn sync_requests_timeout_and_cancel_on_pagehide() -> WebDriverResult<()> {
// test req: operations/003
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 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 () => {
const { fetchWithTimeout } = await import('/sync.js');
const pendingFetch = (_input, init) => new Promise((_resolve, reject) => {
init.signal.addEventListener('abort', () => reject(init.signal.reason), { once: true });
});
const started = performance.now();
let timeout;
try {
await fetchWithTimeout('/never-timeout', {}, pendingFetch, 40);
} catch (error) {
timeout = { name: error.name, message: error.message, elapsedMs: performance.now() - started };
}
const cancellationPromise = fetchWithTimeout('/never-pagehide', {}, pendingFetch, 10_000)
.then(() => ({ resolved: true }))
.catch((error) => ({ name: error.name, message: error.message }));
window.dispatchEvent(new PageTransitionEvent('pagehide'));
done({ timeout, cancellation: await cancellationPromise });
})().catch((error) => done({ error: String(error), stack: error?.stack }));
"#,
Vec::new(),
)
.await?
.json()
.clone();
assert!(proof["error"].is_null(), "bounded request failed: {proof}");
assert_eq!(proof["timeout"]["name"], "TimeoutError", "{proof}");
assert!(proof["timeout"]["message"]
.as_str()
.is_some_and(|message| message.contains("40 ms")));
assert!(proof["timeout"]["elapsedMs"]
.as_f64()
.is_some_and(|elapsed| (35.0..1_000.0).contains(&elapsed)));
assert_eq!(proof["cancellation"]["name"], "AbortError", "{proof}");
assert_eq!(
proof["cancellation"]["message"],
"sync cancelled because page is hidden"
);
Ok(())
}
.await;
let quit = driver.quit().await;
result.and(quit)
}
#[tokio::test]
async fn identical_sync_inputs_reconcile_deterministically() -> WebDriverResult<()> {
// test req: sync/022