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

feat(kanban): migrate queued commands transactionally

Browse Source 
req: sync/014
This commit is contained in:
slhx agent
2026-07-13 18:06:47 +02:00
parent 3ace17f9f1
commit a17de85b4f
5 changed files with 251 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/kanban/static/command-log.js
+31 -7
View File
@@ -1,7 +1,10 @@
const DATABASE = "hemx-kanban-v1";
const DATABASE_VERSION = 2;
const COMMANDS = "commands";
const META = "meta";
const COMMAND_SCHEMA = 1;
const COMMAND_SCHEMA = 2;
const LEGACY_COMMAND_SCHEMA = 1;
const MIGRATION_KEY = "commandSchemaMigration";
const EXPORT_SCHEMA = 1;
const MAX_REPLAY_COMMANDS = 64;
const REPLAY_BUDGET_MS = 100;
@@ -23,13 +26,31 @@ function completed(transaction) {
});
}
function migrateCommandLog(request, oldVersion) {
const database = request.result;
if (!database.objectStoreNames.contains(COMMANDS)) database.createObjectStore(COMMANDS, { keyPath: "id" });
if (!database.objectStoreNames.contains(META)) database.createObjectStore(META);
if (oldVersion === 0 || oldVersion >= DATABASE_VERSION) return;
const transaction = request.transaction;
const commands = transaction.objectStore(COMMANDS);
const meta = transaction.objectStore(META);
const all = commands.getAll();
all.addEventListener("success", () => {
const legacy = all.result;
if (legacy.some((command) => command.schemaVersion !== LEGACY_COMMAND_SCHEMA)) {
transaction.abort();
return;
}
for (const command of legacy) {
commands.put({ ...command, schemaVersion: COMMAND_SCHEMA, targetColumn: "done" });
}
meta.put({ from: LEGACY_COMMAND_SCHEMA, to: COMMAND_SCHEMA, migrated: legacy.length }, MIGRATION_KEY);
}, { once: true });
}
function openCommandLog() {
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);
});
const request = indexedDB.open(DATABASE, DATABASE_VERSION);
request.addEventListener("upgradeneeded", (event) => migrateCommandLog(request, event.oldVersion));
return result(request);
}
@@ -86,6 +107,7 @@ async function appendReorder(database, wire) {
causal,
kind: "reorder_card",
cardId: String(wire[2] || "1"),
targetColumn: "done",
eventKind: String(wire[1] || "click"),
key: wire[4] ? String(wire[4]) : null,
};
@@ -145,6 +167,7 @@ function validate(command) {
if (command.id !== `${command.actor}:${command.causal}`) invalidCommand(command, "id");
if (command.kind !== "reorder_card") invalidCommand(command, "kind");
if (typeof command.cardId !== "string" || !command.cardId) invalidCommand(command, "cardId");
if (command.targetColumn !== "done") invalidCommand(command, "targetColumn");
if (typeof command.eventKind !== "string" || !command.eventKind) invalidCommand(command, "eventKind");
if (command.key !== null && typeof command.key !== "string") invalidCommand(command, "key");
return command;
@@ -291,6 +314,7 @@ async function start() {
actor: command.actor,
session: command.session,
causal: command.causal,
targetColumn: command.targetColumn,
},
}));
try {
examples/kanban/static/sync.js
+38 -8
View File
@@ -1,5 +1,9 @@
const DATABASE = "hemx-kanban-v1";
const DATABASE_VERSION = 2;
const COMMANDS = "commands";
const COMMAND_SCHEMA = 2;
const LEGACY_COMMAND_SCHEMA = 1;
const MIGRATION_KEY = "commandSchemaMigration";
const MAX_ATTEMPTS = 3;
const BACKOFF_MS = [25, 50];
const root = document.querySelector("[data-kanban-sync]");
@@ -43,13 +47,31 @@ function transactionDone(transaction) {
});
}
function migrateCommandLog(request, oldVersion) {
const database = request.result;
if (!database.objectStoreNames.contains(COMMANDS)) database.createObjectStore(COMMANDS, { keyPath: "id" });
if (!database.objectStoreNames.contains("meta")) database.createObjectStore("meta");
if (oldVersion === 0 || oldVersion >= DATABASE_VERSION) return;
const transaction = request.transaction;
const commands = transaction.objectStore(COMMANDS);
const meta = transaction.objectStore("meta");
const all = commands.getAll();
all.addEventListener("success", () => {
const legacy = all.result;
if (legacy.some((command) => command.schemaVersion !== LEGACY_COMMAND_SCHEMA)) {
transaction.abort();
return;
}
for (const command of legacy) {
commands.put({ ...command, schemaVersion: COMMAND_SCHEMA, targetColumn: "done" });
}
meta.put({ from: LEGACY_COMMAND_SCHEMA, to: COMMAND_SCHEMA, migrated: legacy.length }, MIGRATION_KEY);
}, { 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");
});
const request = indexedDB.open(DATABASE, DATABASE_VERSION);
request.addEventListener("upgradeneeded", (event) => migrateCommandLog(request, event.oldVersion));
return requestResult(request);
}
@@ -124,7 +146,7 @@ function setPhase(phase, 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) {
if (!command || command.schemaVersion !== COMMAND_SCHEMA || command.kind !== "reorder_card" || typeof command.id !== "string" || !command.id || typeof command.cardId !== "string" || !command.cardId || command.targetColumn !== "done") {
throw new Error("invalid pending command");
}
return command;
@@ -155,7 +177,7 @@ async function upload(command) {
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 query = new URLSearchParams({ command_id: command.id, card_id: command.cardId, column: command.targetColumn });
const response = await fetch(`/sync/commands?${query}`, { method: "POST" });
if (response.status === 503 && attempt < MAX_ATTEMPTS) {
const base = BACKOFF_MS[attempt - 1];
@@ -342,6 +364,14 @@ async function start() {
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 migration = await requestResult(database.transaction("meta", "readonly").objectStore("meta").get(MIGRATION_KEY));
root.setAttribute("data-sync-database-version", String(database.version));
root.setAttribute("data-sync-command-schema", String(COMMAND_SCHEMA));
if (migration) {
root.setAttribute("data-sync-migration-from", String(migration.from));
root.setAttribute("data-sync-migration-to", String(migration.to));
root.setAttribute("data-sync-migrated-count", String(migration.migrated));
}
const commands = await pendingCommands(database);
root.setAttribute("data-sync-uploaded-this-run", "0");
root.setAttribute("data-sync-uploaded-total", "0");
examples/kanban/tests/browser_e2e.rs
+175 -17
View File
@@ -232,7 +232,7 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onupgradeneeded = () => {
const database = open.result;
if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' });
@@ -297,7 +297,7 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onsuccess = () => {
const request = open.result.transaction('commands', 'readonly').objectStore('commands').count();
request.onsuccess = () => done(request.result);
@@ -315,12 +315,12 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
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,
id: 'sync-actor:1', schemaVersion: 2, actor: 'sync-actor', session: 'sync-session',
causal: 2, kind: 'reorder_card', cardId: '2', targetColumn: 'done', eventKind: 'click', key: null,
});
tx.oncomplete = () => done({ seeded: true });
tx.onabort = () => done({ error: tx.error && tx.error.name });
@@ -372,6 +372,164 @@ async fn pending_local_command_uploads_with_bounded_retry_and_is_removed_on_ack(
result.and(quit)
}
#[tokio::test]
async fn schema_upgrade_preserves_queued_order_and_local_intent() -> WebDriverResult<()> {
// test req: sync/004 req: sync/010 req: sync/014
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;
database.createObjectStore('commands', { keyPath: 'id' });
database.createObjectStore('meta');
};
open.onsuccess = () => {
const tx = open.result.transaction(['commands', 'meta'], 'readwrite');
const commands = tx.objectStore('commands');
for (const [causal, cardId, eventKind, key] of [[1, '1', 'click', null], [2, '2', 'keydown', 'Enter'], [3, '3', 'click', null]]) {
commands.add({
id: `upgrade:${causal}`, schemaVersion: 1, actor: 'upgrade', session: 'legacy-session',
causal, kind: 'reorder_card', cardId, eventKind, key,
});
}
tx.objectStore('meta').put(3, 'causal');
tx.oncomplete = () => done({ version: open.result.version });
tx.onabort = () => done({ error: tx.error && tx.error.name });
};
"#,
Vec::new(),
)
.await?
.json()
.clone();
assert_eq!(seeded["version"], 1, "failed to seed legacy queue: {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 migrated = driver
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const root = document.querySelector('[data-kanban-sync]');
const open = indexedDB.open('hemx-kanban-v1');
open.onsuccess = () => {
const database = open.result;
const tx = database.transaction(['commands', 'meta'], 'readonly');
const commands = tx.objectStore('commands').getAll();
const receipt = tx.objectStore('meta').get('commandSchemaMigration');
tx.oncomplete = () => done({
databaseVersion: database.version,
commandSchema: root.getAttribute('data-sync-command-schema'),
migrationFrom: root.getAttribute('data-sync-migration-from'),
migrationTo: root.getAttribute('data-sync-migration-to'),
migratedCount: root.getAttribute('data-sync-migrated-count'),
pending: root.getAttribute('data-sync-pending-count'),
receipt: receipt.result,
commands: commands.result.sort((a, b) => a.causal - b.causal).map(({ id, schemaVersion, cardId, targetColumn, eventKind, key }) => ({ id, schemaVersion, cardId, targetColumn, eventKind, key })),
});
tx.onabort = () => done({ error: tx.error && tx.error.name });
};
"#,
Vec::new(),
)
.await?
.json()
.clone();
assert_eq!(migrated["databaseVersion"], 2);
assert_eq!(migrated["commandSchema"], "2");
assert_eq!(migrated["migrationFrom"], "1");
assert_eq!(migrated["migrationTo"], "2");
assert_eq!(migrated["migratedCount"], "3");
assert_eq!(migrated["pending"], "3");
assert_eq!(
migrated["receipt"],
serde_json::json!({ "from": 1, "to": 2, "migrated": 3 })
);
assert_eq!(
migrated["commands"],
serde_json::json!([
{ "id": "upgrade:1", "schemaVersion": 2, "cardId": "1", "targetColumn": "done", "eventKind": "click", "key": null },
{ "id": "upgrade:2", "schemaVersion": 2, "cardId": "2", "targetColumn": "done", "eventKind": "keydown", "key": "Enter" },
{ "id": "upgrade:3", "schemaVersion": 2, "cardId": "3", "targetColumn": "done", "eventKind": "click", "key": null }
])
);
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') === 'backpressured' && root?.getAttribute('data-sync-pending-count') === '1'",
)
.await?;
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?;
driver
.execute(
r#"
window.__upgradeReplay = [];
const source = new EventSource('/sync/acknowledgements?after=0');
source.addEventListener('acknowledgement', (event) => {
window.__upgradeReplay.push({ id: event.lastEventId, body: JSON.parse(event.data) });
if (window.__upgradeReplay.length === 3) source.close();
});
return true;
"#,
Vec::new(),
)
.await?;
wait_until(&driver, "return window.__upgradeReplay.length === 3").await?;
let replay = driver
.execute("return window.__upgradeReplay", Vec::new())
.await?
.json()
.clone();
assert_eq!(
replay,
serde_json::json!([
{ "id": "1", "body": { "commandId": "upgrade:1", "cardId": 1, "canonicalColumn": "done", "serverSequence": 1, "status": "accepted" } },
{ "id": "2", "body": { "commandId": "upgrade:2", "cardId": 2, "canonicalColumn": "done", "serverSequence": 2, "status": "accepted" } },
{ "id": "3", "body": { "commandId": "upgrade:3", "cardId": 3, "canonicalColumn": "done", "serverSequence": 3, "status": "accepted" } }
])
);
assert_eq!(command_count(&driver).await?, 0);
Ok(())
}
.await;
let quit = driver.quit().await;
result.and(quit)
}
#[tokio::test]
async fn mixed_queue_removes_accepted_prefix_and_retains_rejected_tail() -> WebDriverResult<()> {
// test req: sync/009 req: sync/010
@@ -398,7 +556,7 @@ async fn mixed_queue_removes_accepted_prefix_and_retains_rejected_tail() -> WebD
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onupgradeneeded = () => {
const database = open.result;
if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' });
@@ -460,7 +618,7 @@ async fn mixed_queue_removes_accepted_prefix_and_retains_rejected_tail() -> WebD
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onsuccess = () => {
const request = open.result.transaction('commands', 'readonly').objectStore('commands').getAll();
request.onsuccess = () => done(request.result.sort((a, b) => a.causal - b.causal).map(({ id, cardId }) => ({ id, cardId })));
@@ -526,7 +684,7 @@ async fn upload_backpressure_keeps_pending_work_visible_and_recoverable() -> Web
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onupgradeneeded = () => {
const database = open.result;
if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' });
@@ -637,7 +795,7 @@ async fn two_tabs_coordinate_single_uploader_and_takeover_without_duplicate_appl
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onupgradeneeded = () => {
const database = open.result;
if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' });
@@ -808,7 +966,7 @@ async fn exhausted_offline_retries_keep_command_until_later_reconnect() -> WebDr
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onupgradeneeded = () => {
const database = open.result;
if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' });
@@ -949,7 +1107,7 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onupgradeneeded = () => {
const database = open.result;
if (!database.objectStoreNames.contains('commands')) database.createObjectStore('commands', { keyPath: 'id' });
@@ -1007,7 +1165,7 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onsuccess = () => {
const tx = open.result.transaction('meta', 'readonly');
const meta = tx.objectStore('meta');
@@ -1066,12 +1224,12 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onsuccess = () => {
const tx = open.result.transaction('commands', 'readwrite');
tx.objectStore('commands').add({
id: 'history:3', schemaVersion: 1, actor: 'history', session: 'history-session',
causal: 3, kind: 'reorder_card', cardId: '2', eventKind: 'click', key: null,
id: 'history:3', schemaVersion: 2, actor: 'history', session: 'history-session',
causal: 3, kind: 'reorder_card', cardId: '2', targetColumn: 'done', eventKind: 'click', key: null,
});
tx.oncomplete = () => done({ seeded: true });
tx.onabort = () => done({ error: tx.error && tx.error.name });
@@ -1116,7 +1274,7 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onsuccess = () => {
const tx = open.result.transaction(['commands', 'meta'], 'readonly');
const command = tx.objectStore('commands').get('history:3');
@@ -1287,7 +1445,7 @@ async fn command_count(driver: &WebDriver) -> WebDriverResult<u64> {
.execute_async(
r#"
const done = arguments[arguments.length - 1];
const open = indexedDB.open('hemx-kanban-v1', 1);
const open = indexedDB.open('hemx-kanban-v1');
open.onsuccess = () => {
const request = open.result.transaction('commands', 'readonly').objectStore('commands').count();
request.onsuccess = () => done(request.result);