fix(desktop): prevent false sync conflicts
This commit is contained in:
1 parent
0fe66082b3
commit
facf68614e
5 files changed
+349
-67
No files matched your search
@@ -10,6 +10,7 @@ import {
|
||||
} from "@edgeever/shared";
|
||||
import { api, ApiRequestError } from "@/lib/api";
|
||||
import { isDesktopResourceRuntime } from "@/lib/desktop-resources";
|
||||
import { notifyMemoIdRemapped, notifyMemoSyncAcknowledged } from "@/lib/sync-events";
|
||||
|
||||
type StagedResourceRewrite = { memoId: string; placeholder: string; url: string };
|
||||
let lastSyncFailed = false;
|
||||
@@ -152,12 +153,6 @@ const request = async <M extends keyof DesktopRpcResponses>(method: M, params: D
|
||||
return bridge.sidecarRequest<DesktopRpcResponses[M]>(method, params);
|
||||
};
|
||||
|
||||
const applyRemoteMemo = async (memo: NonNullable<DesktopRpcResponses["memo.get"]>["memo"]) => {
|
||||
await request("sync.apply", {
|
||||
changes: [{ entityType: "memo", operation: "upsert", entityId: memo.id, memo, notebook: null }],
|
||||
});
|
||||
};
|
||||
|
||||
const applyRemoteNotebook = async (notebook: DesktopRpcResponses["notebook.list"]["notebooks"][number]) => {
|
||||
await request("sync.apply", {
|
||||
changes: [{ entityType: "notebook", operation: "upsert", entityId: notebook.id, notebook, memo: null }],
|
||||
@@ -188,9 +183,8 @@ const acknowledge = async (
|
||||
remoteNotebook?: DesktopRpcResponses["notebook.list"]["notebooks"][number],
|
||||
remoteTemplate?: DesktopRpcResponses["template.list"]["templates"][number],
|
||||
) => {
|
||||
if (remoteMemo) await applyRemoteMemo(remoteMemo);
|
||||
if (remoteNotebook) await applyRemoteNotebook(remoteNotebook);
|
||||
await request("sync.outbox.ack", { id: item.id, remoteMemo, remoteNotebook, remoteTemplate });
|
||||
await request("sync.outbox.ack", { id: item.id, version: item.version, remoteMemo, remoteNotebook, remoteTemplate });
|
||||
};
|
||||
|
||||
const syncOutboxItem = async (item: DesktopOutboxItem, stagedRewrites: StagedResourceRewrite[]) => {
|
||||
@@ -380,15 +374,21 @@ const syncOutbox = async (stagedRewrites: StagedResourceRewrite[], onlyKinds?: S
|
||||
const result = await syncOutboxItem(item, stagedRewrites);
|
||||
if (item.kind === "memo.create" && result && typeof result === "object" && "id" in result && typeof result.id === "string") {
|
||||
memoIdMappings.set(item.entityId, result.id);
|
||||
// Remap the live editor and advance its cloud base as soon as the
|
||||
// create is acknowledged. Waiting for the workspace-wide sync to end
|
||||
// leaves a window where the next autosave still sends revision 0.
|
||||
notifyMemoIdRemapped(new Map([[item.entityId, result.id]]));
|
||||
}
|
||||
if (result && typeof result === "object" && "id" in result && typeof result.id === "string" && "contentJson" in result) {
|
||||
syncedMemos.set(result.id, result as DesktopRpcResponses["memo.get"]["memo"]);
|
||||
const syncedMemo = result as DesktopRpcResponses["memo.get"]["memo"];
|
||||
syncedMemos.set(result.id, syncedMemo);
|
||||
notifyMemoSyncAcknowledged(syncedMemo);
|
||||
}
|
||||
synced += 1;
|
||||
} catch (error) {
|
||||
const conflict = error instanceof ApiRequestError
|
||||
&& (error.code === "revision_conflict" || error.code === "content_conflict" || error.code === "edit_session_conflict");
|
||||
await request("sync.outbox.fail", { id: item.id, error: error instanceof Error ? error.message : String(error), conflict });
|
||||
await request("sync.outbox.fail", { id: item.id, version: item.version, error: error instanceof Error ? error.message : String(error), conflict });
|
||||
if (conflict) conflicted += 1;
|
||||
else failed += 1;
|
||||
}
|
||||
@@ -521,7 +521,9 @@ export const syncDesktopData = () => {
|
||||
if (stagedResources.failed === 0 && outbox.failed === 0 && outbox.conflicted === 0 && creates.conflicted === 0) {
|
||||
await removeSyncedStagedResources(stagedResources.stagedIds);
|
||||
}
|
||||
if (outbox.conflicted === 0 && creates.conflicted === 0 && (typeof navigator === "undefined" || navigator.onLine)) await pullRemoteChanges();
|
||||
const remaining = await request("sync.status", {});
|
||||
if (remaining.pending === 0 && remaining.syncing === 0 && remaining.error === 0 && remaining.conflict === 0
|
||||
&& (typeof navigator === "undefined" || navigator.onLine)) await pullRemoteChanges();
|
||||
// Catch resources staged while the network sync itself was running.
|
||||
await remapStagedResourceMemoIds(creates.memoIdMappings);
|
||||
lastSyncFailed = false;
|
||||
@@ -552,7 +554,7 @@ export const getDesktopSyncSummary = async () => {
|
||||
};
|
||||
|
||||
export const discardDesktopConflicts = async () => {
|
||||
const response = await request("sync.outbox.list", { limit: 200 });
|
||||
const response = await request("sync.outbox.list", { limit: 200, includeConflicts: true });
|
||||
const conflicts = response.items.filter((item) => item.status === "conflict");
|
||||
let discarded = 0;
|
||||
|
||||
@@ -565,7 +567,7 @@ export const discardDesktopConflicts = async () => {
|
||||
await request("sync.apply", {
|
||||
changes: [{ entityType: "memo", operation: "upsert", entityId: remote.memo.id, memo: remote.memo, notebook: null }],
|
||||
});
|
||||
await request("sync.outbox.ack", { id: item.id, remoteMemo: remote.memo });
|
||||
await request("sync.outbox.ack", { id: item.id, version: item.version, remoteMemo: remote.memo });
|
||||
} else {
|
||||
await request("sync.outbox.discard", { id: item.id });
|
||||
}
|
||||
@@ -585,13 +587,13 @@ export const discardDesktopMemoConflict = async (memoId: string) => {
|
||||
changes: [{ entityType: "memo", operation: "upsert", entityId: remote.memo.id, memo: remote.memo, notebook: null }],
|
||||
});
|
||||
|
||||
const response = await request("sync.outbox.list", { limit: 200 });
|
||||
const response = await request("sync.outbox.list", { limit: 200, includeConflicts: true });
|
||||
for (const item of response.items) {
|
||||
if (item.entityId !== memoId || item.status !== "conflict") {
|
||||
continue;
|
||||
}
|
||||
if (item.kind === "memo.update") {
|
||||
await request("sync.outbox.ack", { id: item.id, remoteMemo: remote.memo });
|
||||
await request("sync.outbox.ack", { id: item.id, version: item.version, remoteMemo: remote.memo });
|
||||
} else {
|
||||
await request("sync.outbox.discard", { id: item.id });
|
||||
}
|
||||
|
||||
@@ -97,6 +97,43 @@ fn apply_migrations(connection: &Connection, migrations: &Path) -> rusqlite::Res
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn ensure_sidecar_schema(connection: &Connection) -> rusqlite::Result<()> {
|
||||
connection.execute_batch(
|
||||
"CREATE TABLE IF NOT EXISTS _edgeever_sidecar_meta (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS _edgeever_sidecar_outbox (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
kind TEXT NOT NULL,
|
||||
entity_id TEXT NOT NULL,
|
||||
payload_json TEXT NOT NULL CHECK (json_valid(payload_json)),
|
||||
status TEXT NOT NULL DEFAULT 'pending',
|
||||
attempt_count INTEGER NOT NULL DEFAULT 0,
|
||||
last_error TEXT,
|
||||
version INTEGER NOT NULL DEFAULT 1,
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
|
||||
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_sidecar_outbox_status ON _edgeever_sidecar_outbox(status, id);",
|
||||
)?;
|
||||
|
||||
let has_version = connection
|
||||
.prepare("PRAGMA table_info(_edgeever_sidecar_outbox)")?
|
||||
.query_map([], |row| row.get::<_, String>(1))?
|
||||
.collect::<Result<Vec<_>, _>>()?
|
||||
.iter()
|
||||
.any(|name| name == "version");
|
||||
if !has_version {
|
||||
connection.execute(
|
||||
"ALTER TABLE _edgeever_sidecar_outbox ADD COLUMN version INTEGER NOT NULL DEFAULT 1",
|
||||
[],
|
||||
)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn copy_directory(source: &Path, destination: &Path) -> io::Result<()> {
|
||||
if !source.exists() {
|
||||
return Ok(());
|
||||
@@ -246,6 +283,7 @@ pub(crate) fn restore_database(
|
||||
drop(source);
|
||||
fs::remove_file(&source_copy).map_err(|error| error.to_string())?;
|
||||
apply_migrations(database, migrations).map_err(|error| error.to_string())?;
|
||||
ensure_sidecar_schema(database).map_err(|error| error.to_string())?;
|
||||
let resource_source = if resource_restore_source.is_dir() {
|
||||
resource_restore_source.as_path()
|
||||
} else {
|
||||
@@ -271,25 +309,7 @@ pub(crate) fn open_database(root: &Path, migrations: &Path) -> rusqlite::Result<
|
||||
let existed = root.join("edgeever.sqlite").exists();
|
||||
let connection = Connection::open(root.join("edgeever.sqlite"))?;
|
||||
connection.pragma_update(None, "journal_mode", "WAL")?;
|
||||
connection.execute_batch(
|
||||
"CREATE TABLE IF NOT EXISTS _edgeever_sidecar_meta (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS _edgeever_sidecar_outbox (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
kind TEXT NOT NULL,
|
||||
entity_id TEXT NOT NULL,
|
||||
payload_json TEXT NOT NULL CHECK (json_valid(payload_json)),
|
||||
status TEXT NOT NULL DEFAULT 'pending',
|
||||
attempt_count INTEGER NOT NULL DEFAULT 0,
|
||||
last_error TEXT,
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
|
||||
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_sidecar_outbox_status ON _edgeever_sidecar_outbox(status, id);",
|
||||
)?;
|
||||
ensure_sidecar_schema(&connection)?;
|
||||
if existed {
|
||||
backup_database(&connection, root)?;
|
||||
}
|
||||
|
||||
@@ -33,15 +33,15 @@ fn enqueue_change(
|
||||
payload: &Value,
|
||||
) -> Result<(), String> {
|
||||
if kind == "memo.update" {
|
||||
let existing: Option<(i64, String)> = database
|
||||
let existing: Option<(i64, String, String)> = database
|
||||
.query_row(
|
||||
"SELECT id, payload_json FROM _edgeever_sidecar_outbox WHERE kind = 'memo.update' AND entity_id = ?1 AND status IN ('pending', 'error') ORDER BY id LIMIT 1",
|
||||
"SELECT id, payload_json, status FROM _edgeever_sidecar_outbox WHERE kind = 'memo.update' AND entity_id = ?1 AND status IN ('pending', 'error', 'conflict') ORDER BY id LIMIT 1",
|
||||
[entity_id],
|
||||
|row| Ok((row.get(0)?, row.get(1)?)),
|
||||
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
|
||||
)
|
||||
.optional()
|
||||
.map_err(|e| e.to_string())?;
|
||||
if let Some((outbox_id, existing_payload)) = existing {
|
||||
if let Some((outbox_id, existing_payload, existing_status)) = existing {
|
||||
let previous: Value =
|
||||
serde_json::from_str(&existing_payload).unwrap_or_else(|_| json!({}));
|
||||
let mut merged = payload.clone();
|
||||
@@ -57,10 +57,19 @@ fn enqueue_change(
|
||||
}
|
||||
}
|
||||
}
|
||||
database.execute(
|
||||
"UPDATE _edgeever_sidecar_outbox SET payload_json = ?1, status = 'pending', last_error = NULL, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?2",
|
||||
rusqlite::params![merged.to_string(), outbox_id],
|
||||
).map(|_| ()).map_err(|e| e.to_string())?;
|
||||
database
|
||||
.execute(
|
||||
"UPDATE _edgeever_sidecar_outbox
|
||||
SET payload_json = ?1,
|
||||
status = ?2,
|
||||
last_error = CASE WHEN ?2 = 'conflict' THEN last_error ELSE NULL END,
|
||||
version = version + 1,
|
||||
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')
|
||||
WHERE id = ?3",
|
||||
rusqlite::params![merged.to_string(), existing_status, outbox_id],
|
||||
)
|
||||
.map(|_| ())
|
||||
.map_err(|e| e.to_string())?;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
@@ -70,6 +79,43 @@ fn enqueue_change(
|
||||
).map(|_| ()).map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
fn memo_remap_base_key(memo_id: &str) -> String {
|
||||
format!("memo.remap-base:{memo_id}")
|
||||
}
|
||||
|
||||
fn resolve_remapped_memo_base(
|
||||
database: &Connection,
|
||||
memo_id: &str,
|
||||
expected_revision: i64,
|
||||
expected_content_hash: &str,
|
||||
) -> (i64, String, bool) {
|
||||
let key = memo_remap_base_key(memo_id);
|
||||
let Some(marker) =
|
||||
meta_value(database, &key).and_then(|value| serde_json::from_str::<Value>(&value).ok())
|
||||
else {
|
||||
return (expected_revision, expected_content_hash.to_owned(), false);
|
||||
};
|
||||
let matches_temporary_base = marker.get("temporaryRevision").and_then(Value::as_i64)
|
||||
== Some(expected_revision)
|
||||
&& marker.get("temporaryContentHash").and_then(Value::as_str)
|
||||
== Some(expected_content_hash);
|
||||
if !matches_temporary_base {
|
||||
return (expected_revision, expected_content_hash.to_owned(), false);
|
||||
}
|
||||
(
|
||||
marker
|
||||
.get("remoteRevision")
|
||||
.and_then(Value::as_i64)
|
||||
.unwrap_or(expected_revision),
|
||||
marker
|
||||
.get("remoteContentHash")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or(expected_content_hash)
|
||||
.to_owned(),
|
||||
true,
|
||||
)
|
||||
}
|
||||
|
||||
fn meta_value(database: &Connection, key: &str) -> Option<String> {
|
||||
database
|
||||
.query_row(
|
||||
@@ -981,6 +1027,27 @@ fn create_memo(database: &Connection, params: &Value) -> Result<Value, String> {
|
||||
fn update_memo(database: &Connection, params: &Value) -> Result<Value, String> {
|
||||
let memo_id = string_param(params, "memoId")?;
|
||||
let previous = memo_value(database, &memo_id, true)?;
|
||||
let requested_revision = params
|
||||
.get("expectedRevision")
|
||||
.and_then(Value::as_i64)
|
||||
.unwrap_or(0);
|
||||
let requested_content_hash = params
|
||||
.get("expectedContentHash")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("");
|
||||
// A freshly-created desktop note changes from its local id/revision 0 to
|
||||
// the server id/revision while the editor is still live. If an autosave
|
||||
// was already scheduled, translate that one known local base to the exact
|
||||
// create acknowledgement instead of misclassifying our own create as a
|
||||
// remote edit. The marker retains both sides of the mapping, so a genuine
|
||||
// later remote revision still conflicts normally.
|
||||
let (expected_revision, expected_content_hash, consumed_remap_base) =
|
||||
resolve_remapped_memo_base(
|
||||
database,
|
||||
&memo_id,
|
||||
requested_revision,
|
||||
requested_content_hash,
|
||||
);
|
||||
let title = params
|
||||
.get("title")
|
||||
.and_then(Value::as_str)
|
||||
@@ -1028,12 +1095,20 @@ fn update_memo(database: &Connection, params: &Value) -> Result<Value, String> {
|
||||
"memo.update",
|
||||
&memo_id,
|
||||
&json!({
|
||||
"memoId": memo_id, "expectedRevision": params.get("expectedRevision").and_then(Value::as_i64).unwrap_or(0),
|
||||
"expectedContentHash": params.get("expectedContentHash").and_then(Value::as_str).unwrap_or(""),
|
||||
"memoId": memo_id, "expectedRevision": expected_revision,
|
||||
"expectedContentHash": expected_content_hash,
|
||||
"title": params.get("title").and_then(Value::as_str).unwrap_or(""), "contentJson": content_json,
|
||||
"contentMarkdown": markdown, "tags": tags
|
||||
}),
|
||||
)?;
|
||||
if consumed_remap_base {
|
||||
database
|
||||
.execute(
|
||||
"DELETE FROM _edgeever_sidecar_meta WHERE key = ?1",
|
||||
[memo_remap_base_key(&memo_id)],
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
}
|
||||
memo_value(database, &memo_id, true)
|
||||
}
|
||||
|
||||
@@ -1183,7 +1258,8 @@ fn reset_sync_mirror(database: &Connection) -> Result<(), String> {
|
||||
DELETE FROM notebooks
|
||||
WHERE id NOT IN (SELECT id FROM _edgeever_bootstrap_preserved_notebooks);
|
||||
DELETE FROM _edgeever_sidecar_meta
|
||||
WHERE key IN ('sync.cursor', 'sync.identity', 'sync.last_synced_at');
|
||||
WHERE key IN ('sync.cursor', 'sync.identity', 'sync.last_synced_at')
|
||||
OR key LIKE 'memo.remap-base:%';
|
||||
INSERT INTO _edgeever_sidecar_meta (key, value, updated_at)
|
||||
VALUES ('sync.bootstrap.reset_pending', '1', strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at;",
|
||||
@@ -1250,12 +1326,16 @@ fn sync_outbox_list(database: &Connection, params: &Value) -> Result<Value, Stri
|
||||
.and_then(Value::as_i64)
|
||||
.unwrap_or(50)
|
||||
.clamp(1, 200);
|
||||
let mut statement = database.prepare(
|
||||
"SELECT id, kind, entity_id, payload_json, attempt_count, status, last_error FROM _edgeever_sidecar_outbox WHERE status IN ('pending', 'error', 'conflict') ORDER BY id LIMIT ?1",
|
||||
).map_err(|e| e.to_string())?;
|
||||
let include_conflicts = bool_param(params, "includeConflicts", false);
|
||||
let sql = if include_conflicts {
|
||||
"SELECT id, kind, entity_id, payload_json, attempt_count, status, last_error, version FROM _edgeever_sidecar_outbox WHERE status IN ('pending', 'error', 'conflict') ORDER BY id LIMIT ?1"
|
||||
} else {
|
||||
"SELECT id, kind, entity_id, payload_json, attempt_count, status, last_error, version FROM _edgeever_sidecar_outbox WHERE status IN ('pending', 'error') ORDER BY id LIMIT ?1"
|
||||
};
|
||||
let mut statement = database.prepare(sql).map_err(|e| e.to_string())?;
|
||||
let rows = statement.query_map([limit], |row| {
|
||||
let payload: String = row.get(3)?;
|
||||
Ok(json!({ "id": row.get::<_, i64>(0)?, "kind": row.get::<_, String>(1)?, "entityId": row.get::<_, String>(2)?, "payload": serde_json::from_str::<Value>(&payload).unwrap_or_else(|_| json!({})), "attemptCount": row.get::<_, i64>(4)?, "status": row.get::<_, String>(5)?, "lastError": row.get::<_, Option<String>>(6)? }))
|
||||
Ok(json!({ "id": row.get::<_, i64>(0)?, "kind": row.get::<_, String>(1)?, "entityId": row.get::<_, String>(2)?, "payload": serde_json::from_str::<Value>(&payload).unwrap_or_else(|_| json!({})), "attemptCount": row.get::<_, i64>(4)?, "status": row.get::<_, String>(5)?, "lastError": row.get::<_, Option<String>>(6)?, "version": row.get::<_, i64>(7)? }))
|
||||
}).map_err(|e| e.to_string())?;
|
||||
let items: Result<Vec<_>, _> = rows.collect();
|
||||
Ok(json!({ "items": items.map_err(|e| e.to_string())? }))
|
||||
@@ -1266,16 +1346,95 @@ fn sync_outbox_ack(database: &Connection, params: &Value) -> Result<Value, Strin
|
||||
.get("id")
|
||||
.and_then(Value::as_i64)
|
||||
.ok_or_else(|| "Missing outbox id".to_owned())?;
|
||||
let (kind, entity_id): (String, String) = database
|
||||
let (kind, entity_id, current_version): (String, String, i64) = database
|
||||
.query_row(
|
||||
"SELECT kind, entity_id FROM _edgeever_sidecar_outbox WHERE id = ?1",
|
||||
"SELECT kind, entity_id, version FROM _edgeever_sidecar_outbox WHERE id = ?1",
|
||||
[id],
|
||||
|row| Ok((row.get(0)?, row.get(1)?)),
|
||||
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
let requested_version = params.get("version").and_then(Value::as_i64);
|
||||
let superseded = requested_version.is_some_and(|version| version != current_version);
|
||||
let remote_memo = params.get("remoteMemo").cloned();
|
||||
let remote_notebook = params.get("remoteNotebook").cloned();
|
||||
let remote_template = params.get("remoteTemplate").cloned();
|
||||
let temporary_memo_base = if kind == "memo.create" {
|
||||
memo_value(database, &entity_id, true).ok().map(|memo| {
|
||||
(
|
||||
memo.get("revision").and_then(Value::as_i64).unwrap_or(0),
|
||||
memo.get("contentHash")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("")
|
||||
.to_owned(),
|
||||
)
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
if kind == "memo.update" && superseded {
|
||||
if let Some(remote) = remote_memo.as_ref() {
|
||||
let remote_revision = remote.get("revision").and_then(Value::as_i64).unwrap_or(0);
|
||||
let remote_hash = remote
|
||||
.get("contentHash")
|
||||
.and_then(Value::as_str)
|
||||
.unwrap_or("");
|
||||
let payload_text: String = database
|
||||
.query_row(
|
||||
"SELECT payload_json FROM _edgeever_sidecar_outbox WHERE id = ?1",
|
||||
[id],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
let mut payload: Value =
|
||||
serde_json::from_str(&payload_text).unwrap_or_else(|_| json!({}));
|
||||
payload["expectedRevision"] = json!(remote_revision);
|
||||
payload["expectedContentHash"] = json!(remote_hash);
|
||||
let tx = database
|
||||
.unchecked_transaction()
|
||||
.map_err(|e| e.to_string())?;
|
||||
// Preserve the successor draft's content while advancing only its
|
||||
// acknowledged cloud base. The next flush will send that draft on
|
||||
// top of the response that just succeeded.
|
||||
tx.execute(
|
||||
"UPDATE memo_contents SET revision = ?1 WHERE memo_id = ?2",
|
||||
rusqlite::params![remote_revision, entity_id],
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
tx.execute(
|
||||
"UPDATE _edgeever_sidecar_outbox
|
||||
SET payload_json = ?1, status = 'pending', last_error = NULL,
|
||||
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')
|
||||
WHERE id = ?2",
|
||||
rusqlite::params![payload.to_string(), id],
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
tx.commit().map_err(|e| e.to_string())?;
|
||||
}
|
||||
return Ok(json!({
|
||||
"ok": true,
|
||||
"superseded": true,
|
||||
"memo": remote_memo,
|
||||
"notebook": remote_notebook,
|
||||
"template": remote_template
|
||||
}));
|
||||
}
|
||||
|
||||
if let Some(remote) = remote_memo.as_ref() {
|
||||
let remote_id = string_param(remote, "id")?;
|
||||
apply_sync_changes(
|
||||
database,
|
||||
&json!({
|
||||
"changes": [{
|
||||
"entityType": "memo",
|
||||
"operation": "upsert",
|
||||
"entityId": remote_id,
|
||||
"memo": remote,
|
||||
"notebook": null
|
||||
}]
|
||||
}),
|
||||
)?;
|
||||
}
|
||||
if kind == "memo.create" || kind == "memo.merge" {
|
||||
if let Some(remote) = remote_memo.as_ref() {
|
||||
let remote_id = string_param(remote, "id")?;
|
||||
@@ -1287,8 +1446,8 @@ fn sync_outbox_ack(database: &Connection, params: &Value) -> Result<Value, Strin
|
||||
)
|
||||
.map_err(|e| e.to_string())?;
|
||||
if remote_exists {
|
||||
// `acknowledge` caches the remote memo before acknowledging the
|
||||
// outbox item, so the remote id normally already exists here.
|
||||
// The acknowledgement caches the remote memo before reconciling
|
||||
// the local placeholder, so the remote id exists here.
|
||||
// A merged local memo can still own resources moved from its
|
||||
// sources. Deleting it directly is rejected by the resources
|
||||
// foreign key and leaves both the local placeholder and remote
|
||||
@@ -1355,7 +1514,24 @@ fn sync_outbox_ack(database: &Connection, params: &Value) -> Result<Value, Strin
|
||||
} else {
|
||||
outbox_entity
|
||||
};
|
||||
database.execute("UPDATE _edgeever_sidecar_outbox SET entity_id = ?1, payload_json = ?2 WHERE id = ?3", rusqlite::params![next_entity, payload.to_string(), outbox_id]).map_err(|e| e.to_string())?;
|
||||
database.execute("UPDATE _edgeever_sidecar_outbox SET entity_id = ?1, payload_json = ?2, version = version + 1 WHERE id = ?3", rusqlite::params![next_entity, payload.to_string(), outbox_id]).map_err(|e| e.to_string())?;
|
||||
}
|
||||
if kind == "memo.create" {
|
||||
if let Some((temporary_revision, temporary_content_hash)) =
|
||||
temporary_memo_base.as_ref()
|
||||
{
|
||||
set_meta(
|
||||
database,
|
||||
&memo_remap_base_key(&remote_id),
|
||||
&json!({
|
||||
"temporaryRevision": temporary_revision,
|
||||
"temporaryContentHash": temporary_content_hash,
|
||||
"remoteRevision": remote.get("revision").and_then(Value::as_i64).unwrap_or(0),
|
||||
"remoteContentHash": remote.get("contentHash").and_then(Value::as_str).unwrap_or("")
|
||||
})
|
||||
.to_string(),
|
||||
)?;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1404,7 +1580,7 @@ fn sync_outbox_ack(database: &Connection, params: &Value) -> Result<Value, Strin
|
||||
} else {
|
||||
outbox_entity
|
||||
};
|
||||
database.execute("UPDATE _edgeever_sidecar_outbox SET entity_id = ?1, payload_json = ?2 WHERE id = ?3", rusqlite::params![next_entity, payload.to_string(), outbox_id]).map_err(|e| e.to_string())?;
|
||||
database.execute("UPDATE _edgeever_sidecar_outbox SET entity_id = ?1, payload_json = ?2, version = version + 1 WHERE id = ?3", rusqlite::params![next_entity, payload.to_string(), outbox_id]).map_err(|e| e.to_string())?;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1453,15 +1629,24 @@ fn sync_outbox_ack(database: &Connection, params: &Value) -> Result<Value, Strin
|
||||
} else {
|
||||
outbox_entity
|
||||
};
|
||||
database.execute("UPDATE _edgeever_sidecar_outbox SET entity_id = ?1, payload_json = ?2 WHERE id = ?3", rusqlite::params![next_entity, payload.to_string(), outbox_id]).map_err(|e| e.to_string())?;
|
||||
database.execute("UPDATE _edgeever_sidecar_outbox SET entity_id = ?1, payload_json = ?2, version = version + 1 WHERE id = ?3", rusqlite::params![next_entity, payload.to_string(), outbox_id]).map_err(|e| e.to_string())?;
|
||||
}
|
||||
}
|
||||
}
|
||||
database
|
||||
.execute("DELETE FROM _edgeever_sidecar_outbox WHERE id = ?1", [id])
|
||||
.map_err(|e| e.to_string())?;
|
||||
let deleted = if let Some(version) = requested_version {
|
||||
database
|
||||
.execute(
|
||||
"DELETE FROM _edgeever_sidecar_outbox WHERE id = ?1 AND version = ?2",
|
||||
rusqlite::params![id, version],
|
||||
)
|
||||
.map_err(|e| e.to_string())?
|
||||
} else {
|
||||
database
|
||||
.execute("DELETE FROM _edgeever_sidecar_outbox WHERE id = ?1", [id])
|
||||
.map_err(|e| e.to_string())?
|
||||
};
|
||||
Ok(
|
||||
json!({ "ok": true, "memo": remote_memo, "notebook": remote_notebook, "template": remote_template }),
|
||||
json!({ "ok": true, "superseded": deleted == 0, "memo": remote_memo, "notebook": remote_notebook, "template": remote_template }),
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1479,8 +1664,12 @@ fn sync_outbox_fail(database: &Connection, params: &Value) -> Result<Value, Stri
|
||||
} else {
|
||||
"error"
|
||||
};
|
||||
database.execute("UPDATE _edgeever_sidecar_outbox SET status = ?1, attempt_count = attempt_count + 1, last_error = ?2, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?3", rusqlite::params![status, error, id]).map_err(|e| e.to_string())?;
|
||||
Ok(json!({ "ok": true }))
|
||||
let updated = if let Some(version) = params.get("version").and_then(Value::as_i64) {
|
||||
database.execute("UPDATE _edgeever_sidecar_outbox SET status = ?1, attempt_count = attempt_count + 1, last_error = ?2, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?3 AND version = ?4", rusqlite::params![status, error, id, version]).map_err(|e| e.to_string())?
|
||||
} else {
|
||||
database.execute("UPDATE _edgeever_sidecar_outbox SET status = ?1, attempt_count = attempt_count + 1, last_error = ?2, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?3", rusqlite::params![status, error, id]).map_err(|e| e.to_string())?
|
||||
};
|
||||
Ok(json!({ "ok": true, "superseded": updated == 0 }))
|
||||
}
|
||||
|
||||
fn sync_outbox_discard(database: &Connection, params: &Value) -> Result<Value, String> {
|
||||
|
||||
@@ -153,9 +153,9 @@ export type DesktopRpcParams = {
|
||||
"memo.revision.cache": { revision: import("./types").MemoRevision };
|
||||
"sync.status": Record<string, never>;
|
||||
"sync.bootstrap.prepare": { reset?: boolean };
|
||||
"sync.outbox.list": { limit?: number };
|
||||
"sync.outbox.ack": { id: number; remoteMemo?: MemoDetail; remoteNotebook?: Notebook; remoteTemplate?: MemoTemplate };
|
||||
"sync.outbox.fail": { id: number; error: string; conflict?: boolean };
|
||||
"sync.outbox.list": { limit?: number; includeConflicts?: boolean };
|
||||
"sync.outbox.ack": { id: number; version?: number; remoteMemo?: MemoDetail; remoteNotebook?: Notebook; remoteTemplate?: MemoTemplate };
|
||||
"sync.outbox.fail": { id: number; version?: number; error: string; conflict?: boolean };
|
||||
"sync.outbox.discard": { id: number };
|
||||
"sync.apply": { changes: Array<{ entityType: "memo" | "notebook"; operation: "upsert" | "delete"; memo?: MemoDetail | null; notebook?: Notebook | null; entityId: string }> };
|
||||
"sync.cursor.set": { cursor: number; syncIdentity: string };
|
||||
@@ -172,6 +172,7 @@ export type DesktopOutboxItem = {
|
||||
entityId: string;
|
||||
payload: Record<string, unknown>;
|
||||
attemptCount: number;
|
||||
version: number;
|
||||
status?: "pending" | "syncing" | "conflict" | "error";
|
||||
lastError?: string | null;
|
||||
};
|
||||
@@ -108,6 +108,75 @@ const coalescedUpdates = (await request("sync.outbox.list", { limit: 200 })).ite
|
||||
assert.equal(coalescedUpdates.length, 1, "offline autosaves should coalesce into one sidecar outbox item");
|
||||
assert.equal(coalescedUpdates[0].payload.contentMarkdown, "latest autosave");
|
||||
assert.equal((await request("memo.get", { memoId: coalesced.memo.id })).memo.revision, 0, "local autosaves must not advance the acknowledged cloud revision");
|
||||
assert.equal(coalescedUpdates[0].version, 2, "coalescing a newer autosave should advance the outbox version");
|
||||
|
||||
const remapRace = await request("memo.create", { notebookId: inbox.id, title: "Remap race", contentMarkdown: "created locally", tags: [] });
|
||||
const remapCreate = (await request("sync.outbox.list", { limit: 200 })).items.find((item) => item.kind === "memo.create" && item.entityId === remapRace.memo.id);
|
||||
assert.ok(remapCreate, "new desktop memo should have a create outbox item");
|
||||
const remoteCreatedMemo = {
|
||||
...remapRace.memo,
|
||||
id: "memo_remote_remap_race",
|
||||
revision: 1,
|
||||
updatedAt: "2026-08-26T02:15:21.000Z",
|
||||
};
|
||||
await request("sync.outbox.ack", { id: remapCreate.id, version: remapCreate.version, remoteMemo: remoteCreatedMemo });
|
||||
await request("memo.update", {
|
||||
...updatePayload("first save after create acknowledgement"),
|
||||
memoId: remoteCreatedMemo.id,
|
||||
expectedRevision: 0,
|
||||
expectedContentHash: remapRace.memo.contentHash,
|
||||
});
|
||||
const remappedUpdate = (await request("sync.outbox.list", { limit: 200 })).items.find((item) => item.kind === "memo.update" && item.entityId === remoteCreatedMemo.id);
|
||||
assert.ok(remappedUpdate, "editing immediately after id remapping should queue a remote-id update");
|
||||
assert.equal(remappedUpdate.payload.expectedRevision, remoteCreatedMemo.revision, "a scheduled revision-0 autosave should rebase to its own create acknowledgement");
|
||||
assert.equal(remappedUpdate.payload.expectedContentHash, remoteCreatedMemo.contentHash);
|
||||
|
||||
await request("memo.update", {
|
||||
...updatePayload("successor typed while the first request was in flight"),
|
||||
memoId: remoteCreatedMemo.id,
|
||||
expectedRevision: remoteCreatedMemo.revision,
|
||||
expectedContentHash: remoteCreatedMemo.contentHash,
|
||||
});
|
||||
const successorBeforeAck = (await request("sync.outbox.list", { limit: 200 })).items.find((item) => item.id === remappedUpdate.id);
|
||||
assert.equal(successorBeforeAck.version, remappedUpdate.version + 1);
|
||||
const firstRemoteUpdate = {
|
||||
...remoteCreatedMemo,
|
||||
revision: 2,
|
||||
contentJson: remappedUpdate.payload.contentJson,
|
||||
contentMarkdown: remappedUpdate.payload.contentMarkdown,
|
||||
contentText: remappedUpdate.payload.contentMarkdown,
|
||||
contentHash: "remote-hash-after-first-save",
|
||||
updatedAt: "2026-08-26T02:15:22.000Z",
|
||||
};
|
||||
const supersededAck = await request("sync.outbox.ack", {
|
||||
id: remappedUpdate.id,
|
||||
version: remappedUpdate.version,
|
||||
remoteMemo: firstRemoteUpdate,
|
||||
});
|
||||
assert.equal(supersededAck.superseded, true, "an acknowledgement must not delete a newer coalesced autosave");
|
||||
const successorAfterAck = (await request("sync.outbox.list", { limit: 200 })).items.find((item) => item.id === remappedUpdate.id);
|
||||
assert.ok(successorAfterAck, "the successor autosave should remain queued");
|
||||
assert.equal(successorAfterAck.status, "pending");
|
||||
assert.equal(successorAfterAck.payload.contentMarkdown, "successor typed while the first request was in flight");
|
||||
assert.equal(successorAfterAck.payload.expectedRevision, firstRemoteUpdate.revision);
|
||||
assert.equal(successorAfterAck.payload.expectedContentHash, firstRemoteUpdate.contentHash);
|
||||
const preservedSuccessor = await request("memo.get", { memoId: remoteCreatedMemo.id });
|
||||
assert.equal(preservedSuccessor.memo.contentMarkdown, "successor typed while the first request was in flight", "acknowledging an older request must preserve the live local draft");
|
||||
assert.equal(preservedSuccessor.memo.revision, firstRemoteUpdate.revision, "the preserved successor should advance to the acknowledged cloud revision");
|
||||
const secondRemoteUpdate = {
|
||||
...preservedSuccessor.memo,
|
||||
revision: 3,
|
||||
contentHash: "remote-hash-after-successor",
|
||||
updatedAt: "2026-08-26T02:15:23.000Z",
|
||||
};
|
||||
const finalAck = await request("sync.outbox.ack", {
|
||||
id: successorAfterAck.id,
|
||||
version: successorAfterAck.version,
|
||||
remoteMemo: secondRemoteUpdate,
|
||||
});
|
||||
assert.equal(finalAck.superseded, false);
|
||||
assert.ok(!(await request("sync.outbox.list", { limit: 200 })).items.some((item) => item.id === successorAfterAck.id), "the exact successor acknowledgement should clear the outbox item");
|
||||
assert.equal((await request("memo.get", { memoId: remoteCreatedMemo.id })).memo.revision, secondRemoteUpdate.revision);
|
||||
|
||||
const batchNotebook = (await request("notebook.create", { name: "Batch destination" })).notebook;
|
||||
const batchFirst = (await request("memo.create", { notebookId: inbox.id, title: "Batch one", contentMarkdown: "batch one", tags: ["batch-tag"] })).memo;
|
||||
@@ -226,7 +295,8 @@ const conflictMemo = await request("memo.create", { notebookId: inbox.id, title:
|
||||
const conflictCandidate = (await request("sync.outbox.list", { limit: 200 })).items.find((item) => item.entityId === conflictMemo.memo.id);
|
||||
assert.ok(conflictCandidate, "conflict test should create an outbox item");
|
||||
await request("sync.outbox.fail", { id: conflictCandidate.id, error: "test conflict", conflict: true });
|
||||
assert.ok((await request("sync.outbox.list", { limit: 200 })).items.some((item) => item.id === conflictCandidate.id && item.status === "conflict"));
|
||||
assert.ok(!(await request("sync.outbox.list", { limit: 200 })).items.some((item) => item.id === conflictCandidate.id), "conflicts must not be returned to the automatic retry loop");
|
||||
assert.ok((await request("sync.outbox.list", { limit: 200, includeConflicts: true })).items.some((item) => item.id === conflictCandidate.id && item.status === "conflict"), "conflicts should remain available to explicit recovery actions");
|
||||
await request("sync.outbox.discard", { id: conflictCandidate.id });
|
||||
assert.equal((await request("sync.status")).conflict, 0);
|
||||
|
||||
|
||||
Reference in new issue
Block a user