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

feat(kanban): resolve divergent queue head

Browse Source 
req: sync/010

req: sync/011
This commit is contained in:
slhx agent
2026-07-13 19:31:40 +02:00
parent 1ec8a1368d
commit 54301c6aa5
4 changed files with 95 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/kanban/static/sync.js
+39 -2
View File
@@ -21,6 +21,7 @@ let uploadsThisRun = 0;
let uploadedTotal = 0;
let inFlightUploads = 0;
let maxObservedInFlight = 0;
let activeConflict;
let stopped = false;
class UploadError extends Error {
@@ -92,7 +93,7 @@ async function pendingCommands(database) {
return commands.sort((left, right) => left.causal - right.causal);
}
async function removeAcknowledged(database, commandId) {
async function removePendingCommand(database, commandId) {
const transaction = database.transaction(COMMANDS, "readwrite");
const done = transactionDone(transaction);
transaction.objectStore(COMMANDS).delete(commandId);
@@ -178,6 +179,30 @@ function setExportAvailable(available) {
root.querySelector("[data-sync-export]").disabled = !available;
}
function setConflictResolutionAvailable(available) {
root.querySelector("[data-sync-use-canonical]").disabled = !available;
}
async function useCanonicalState() {
if (!activeConflict) return;
const { command, snapshot } = activeConflict;
setConflictResolutionAvailable(false);
await removePendingCommand(database, command.id);
const remaining = await pendingCommands(database);
root.setAttribute("data-sync-conflict-resolution", "used-canonical-state");
root.setAttribute("data-sync-resolved-command-id", command.id);
root.setAttribute("data-sync-pending-count", String(remaining.length));
setExportAvailable(remaining.length > 0);
setPhase("conflict-resolved", `Used canonical snapshot ${snapshot.serverSequence}; removed ${command.id} and retained ${remaining.length} queued command${remaining.length === 1 ? "" : "s"}.`);
activeConflict = undefined;
uploadsThisRun = 0;
synchronizing = false;
clearTimeout(leaseTimer);
await releaseUploaderLease(database);
root.setAttribute("data-sync-leader", "false");
await continuePendingWork();
}
async function exportPendingWork() {
const commands = await pendingCommands(database);
if (commands.length === 0) return;
@@ -305,7 +330,7 @@ async function synchronize(command) {
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);
await removePendingCommand(database, command.id);
uploadsThisRun += 1;
uploadedTotal += 1;
root.setAttribute("data-sync-uploaded-this-run", String(uploadsThisRun));
@@ -339,17 +364,24 @@ async function synchronize(command) {
root.setAttribute("data-sync-rebase-reason", decision.reason);
root.setAttribute("data-sync-canonical-column", decision.canonicalColumn);
if (decision.kind === "converged") {
activeConflict = undefined;
setConflictResolutionAvailable(false);
await commitConvergedRebase(database, snapshot, command);
root.setAttribute("data-sync-pending-count", String((await pendingCommands(database)).length));
root.setAttribute("data-sync-ack-sequence", String(snapshot.serverSequence));
setPhase("rebased", `Canonical snapshot ${snapshot.serverSequence} already satisfies ${command.id}; committed and removed the pending command.`);
root.dispatchEvent(new CustomEvent("kanban:sync-rebased", { detail: { snapshot, command, decision } }));
} else {
activeConflict = { command, snapshot, decision };
setConflictResolutionAvailable(true);
setPhase("conflicted", `Canonical snapshot ${snapshot.serverSequence} conflicts with ${command.id} (${decision.reason}); the pending command remains queued.`);
root.dispatchEvent(new CustomEvent("kanban:sync-conflicted", { detail: { snapshot, command, decision } }));
}
synchronizing = false;
source.close();
clearTimeout(leaseTimer);
await releaseUploaderLease(database);
root.setAttribute("data-sync-leader", "false");
});
} catch (error) {
synchronizing = false;
@@ -429,6 +461,7 @@ async function start() {
root.setAttribute("data-sync-max-observed-in-flight", "0");
root.setAttribute("data-sync-pending-count", String(commands.length));
setExportAvailable(commands.length > 0);
setConflictResolutionAvailable(false);
setManualRetryAvailable(false);
if (commands.length === 0) {
setPhase("idle", "No pending commands.");
@@ -436,6 +469,10 @@ async function start() {
}
const command = validatePending(commands[0]);
root.addEventListener("click", async (event) => {
if (event.target.closest("[data-sync-use-canonical]")) {
await useCanonicalState();
return;
}
if (event.target.closest("[data-sync-export]")) {
await exportPendingWork();
return;
examples/kanban/templates/sync_shell.heml
+1
View File
@@ -11,6 +11,7 @@
<p role="status" aria-live="polite">Waiting for pending commands.</p>
<button type="button" data-sync-retry>Retry sync now</button>
<button type="button" data-sync-export disabled>Export this account's queue</button>
<button type="button" data-sync-use-canonical disabled>Use canonical state and continue</button>
</section>
<script +src="self.runtime_src" defer></script>
<script src="/sync.js" defer></script>
examples/kanban/tests/browser_e2e.rs
+53 -10
View File
@@ -1413,8 +1413,9 @@ async fn exhausted_offline_retries_keep_command_until_later_reconnect() -> WebDr
}
#[tokio::test]
async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDriverResult<()> {
// test req: sync/007 req: sync/011 req: sync/020
async fn missing_history_rebase_and_user_conflict_resolution_preserve_suffix() -> WebDriverResult<()>
{
// test req: sync/007 req: sync/010 req: sync/011 req: sync/020
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"));
@@ -1576,10 +1577,15 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
const open = indexedDB.open('hemx-kanban-v1');
open.onsuccess = () => {
const tx = open.result.transaction('commands', 'readwrite');
tx.objectStore('commands').add({
const commands = tx.objectStore('commands');
commands.add({
id: 'history:3', schemaVersion: 2, accountPartition: 'demo:demo', actor: 'history', session: 'history-session',
causal: 3, kind: 'reorder_card', cardId: '2', targetColumn: 'done', eventKind: 'click', key: null,
});
commands.add({
id: 'history:4', schemaVersion: 2, accountPartition: 'demo:demo', actor: 'history', session: 'history-session',
causal: 4, kind: 'reorder_card', cardId: '1', targetColumn: 'done', eventKind: 'click', key: null,
});
tx.oncomplete = () => done({ seeded: true });
tx.onabort = () => done({ error: tx.error && tx.error.name });
};
@@ -1599,7 +1605,7 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
.await?;
let conflicted = driver
.execute(
"const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), uploadSequence: root.getAttribute('data-sync-upload-sequence'), snapshotSequence: root.getAttribute('data-sync-snapshot-sequence'), pending: root.getAttribute('data-sync-pending-count'), rebasePending: root.getAttribute('data-sync-rebase-pending-count'), decision: root.getAttribute('data-sync-rebase-decision'), reason: root.getAttribute('data-sync-rebase-reason'), canonicalColumn: root.getAttribute('data-sync-canonical-column'), status: root.querySelector('[role=status]').textContent, error: root.getAttribute('data-sync-error') }",
"const root = document.querySelector('[data-kanban-sync]'); return { phase: root.getAttribute('data-sync-phase'), uploadSequence: root.getAttribute('data-sync-upload-sequence'), snapshotSequence: root.getAttribute('data-sync-snapshot-sequence'), pending: root.getAttribute('data-sync-pending-count'), rebasePending: root.getAttribute('data-sync-rebase-pending-count'), decision: root.getAttribute('data-sync-rebase-decision'), reason: root.getAttribute('data-sync-rebase-reason'), canonicalColumn: root.getAttribute('data-sync-canonical-column'), resolutionDisabled: root.querySelector('[data-sync-use-canonical]').disabled, status: root.querySelector('[role=status]').textContent, error: root.getAttribute('data-sync-error') }",
Vec::new(),
)
.await?
@@ -1608,17 +1614,18 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
assert_eq!(conflicted["phase"], "conflicted", "unexpected conflict state: {conflicted}");
assert_eq!(conflicted["uploadSequence"], "3");
assert_eq!(conflicted["snapshotSequence"], "4");
assert_eq!(conflicted["pending"], "1");
assert_eq!(conflicted["rebasePending"], "1");
assert_eq!(conflicted["pending"], "2");
assert_eq!(conflicted["rebasePending"], "2");
assert_eq!(conflicted["decision"], "conflicted");
assert_eq!(conflicted["reason"], "canonical-state-diverged");
assert_eq!(conflicted["canonicalColumn"], "doing");
assert_eq!(conflicted["resolutionDisabled"], false);
assert_eq!(
conflicted["status"],
"Canonical snapshot 4 conflicts with history:3 (canonical-state-diverged); the pending command remains queued."
);
assert!(conflicted["error"].is_null());
assert_eq!(command_count(&driver).await?, 1);
assert_eq!(command_count(&driver).await?, 2);
let preserved = driver
.execute_async(
r#"
@@ -1626,10 +1633,10 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
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');
const commands = tx.objectStore('commands').getAll();
const cursor = tx.objectStore('meta').get('acknowledgementCursor');
const snapshot = tx.objectStore('meta').get('canonicalSnapshot');
tx.oncomplete = () => done({ command: command.result, cursor: cursor.result, snapshot: snapshot.result });
tx.oncomplete = () => done({ commands: commands.result.sort((left, right) => left.causal - right.causal), cursor: cursor.result, snapshot: snapshot.result });
tx.onabort = () => done({ error: tx.error && tx.error.name });
};
"#,
@@ -1638,7 +1645,8 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
.await?
.json()
.clone();
assert_eq!(preserved["command"]["id"], "history:3");
assert_eq!(preserved["commands"][0]["id"], "history:3");
assert_eq!(preserved["commands"][1]["id"], "history:4");
assert_eq!(preserved["cursor"], 2);
assert_eq!(preserved["snapshot"]["serverSequence"], 2);
@@ -1654,6 +1662,41 @@ async fn missing_history_rebase_converges_without_losing_local_intent() -> WebDr
assert_eq!(divergent_board[1]["title"], "Doing");
assert_eq!(divergent_board[1]["cards"], serde_json::json!(["2"]));
assert_eq!(divergent_board[2]["cards"], serde_json::json!(["1", "3"]));
driver.goto(&format!("http://{app_addr}/sync-demo")).await?;
wait_until(
&driver,
"return document.querySelector('[data-kanban-sync]')?.getAttribute('data-sync-phase') === 'conflicted'",
)
.await?;
driver
.find(By::Css("[data-sync-use-canonical]"))
.await?
.click()
.await?;
wait_until(
&driver,
"const root = document.querySelector('[data-kanban-sync]'); return root?.getAttribute('data-sync-phase') === 'rebased' && root?.getAttribute('data-sync-pending-count') === '0'",
)
.await?;
let resolved = driver
.execute(
"const root = document.querySelector('[data-kanban-sync]'); return { resolution: root.getAttribute('data-sync-conflict-resolution'), resolvedCommand: root.getAttribute('data-sync-resolved-command-id'), pending: root.getAttribute('data-sync-pending-count'), ackSequence: root.getAttribute('data-sync-ack-sequence'), canonicalColumn: root.getAttribute('data-sync-canonical-column'), status: root.querySelector('[role=status]').textContent }",
Vec::new(),
)
.await?
.json()
.clone();
assert_eq!(resolved["resolution"], "used-canonical-state");
assert_eq!(resolved["resolvedCommand"], "history:3");
assert_eq!(resolved["pending"], "0");
assert_eq!(resolved["ackSequence"], "5");
assert_eq!(resolved["canonicalColumn"], "done");
assert_eq!(
resolved["status"],
"Canonical snapshot 5 already satisfies history:4; committed and removed the pending command."
);
assert_eq!(command_count(&driver).await?, 0);
Ok(())
}
.await;