From 071711dafc85624042b764b84e8e9abf3a03ba8b Mon Sep 17 00:00:00 2001 From: ZacharyZcR Date: Thu, 15 Oct 2026 14:59:59 +0800 Subject: [PATCH] fix(sync): don't delete a local row when the server can't send its copy (#1650) * fix(sync): don't delete a local row when the server can't send its copy A push conflict on a live record whose server row failed to serialize came back as row: null, deleted: false, and the desktop treated that as a delete. The server now rejects that op as unavailable, the desktop never deletes on a conflict without a row, and serialization failures are logged. * fix(sync): retry temporarily unavailable records --- src/backend/sync/client/engine.ts | 17 ++- src/backend/sync/server/push.ts | 13 ++- src/backend/sync/store.ts | 9 +- src/backend/tests/sync/two-way.test.ts | 139 ++++++++++++++++++++++++- 4 files changed, 172 insertions(+), 6 deletions(-) diff --git a/src/backend/sync/client/engine.ts b/src/backend/sync/client/engine.ts index f4152ebf3..8ccd562b3 100644 --- a/src/backend/sync/client/engine.ts +++ b/src/backend/sync/client/engine.ts @@ -351,7 +351,10 @@ export class SyncEngine { const key = recordKey(entity.type, row.syncId); const record = records.get(key); if (record && !record.deleted && record.hash === row.hash) continue; - if (record?.errorHash === row.hash) continue; + // Unavailable server rows can recover without another local edit. + if (record?.errorHash === row.hash && record.error !== "unavailable") { + continue; + } ops.push({ entityType: entity.type, syncId: row.syncId, @@ -367,7 +370,12 @@ export class SyncEngine { if (record.revision === 0) continue; if (rows.has(record.syncId) || failed.has(record.syncId)) continue; if (entity.covers && !entity.covers(record.syncId)) continue; - if (record.errorHash === DELETED_HASH) continue; + if ( + record.errorHash === DELETED_HASH && + record.error !== "unavailable" + ) { + continue; + } ops.push({ entityType: entity.type, syncId: record.syncId, @@ -433,6 +441,9 @@ export class SyncEngine { return; } + // A live row the server could not send says nothing; keep ours to retry. + if (!result.deleted && !result.row) return; + // Conflict: the server moved on since this device last saw the row. Its // version wins here, and a local edit it replaced is kept for the user. if (op.op === "upsert" && op.row) { @@ -445,7 +456,7 @@ export class SyncEngine { ); } const resolvers = createResolvers(userId); - if (result.deleted || !result.row) { + if (result.deleted) { await deleteStoredRow(entity, userId, op.syncId); await putRecord(userId, op.entityType, op.syncId, { revision: result.revision, diff --git a/src/backend/sync/server/push.ts b/src/backend/sync/server/push.ts index 6cf4b41ee..d192e3dfc 100644 --- a/src/backend/sync/server/push.ts +++ b/src/backend/sync/server/push.ts @@ -134,12 +134,23 @@ export async function applyPush( const record = await getRecord(userId, op.entityType, op.syncId); const currentRevision = record?.revision ?? 0; if (op.baseRevision !== currentRevision) { + const row = record?.deleted ? null : (wire.get(key) ?? null); + // A live record with no row failed to serialize; a conflict without a + // row would read as a delete on the desktop. + if (record && !record.deleted && !row) { + results.set(key, { + ...base, + status: "rejected", + reason: "unavailable", + }); + continue; + } results.set(key, { ...base, status: "conflict", revision: currentRevision, deleted: !!record?.deleted, - row: record?.deleted ? null : (wire.get(key) ?? null), + row, }); continue; } diff --git a/src/backend/sync/store.ts b/src/backend/sync/store.ts index 430a3258e..2d6426e18 100644 --- a/src/backend/sync/store.ts +++ b/src/backend/sync/store.ts @@ -13,6 +13,7 @@ import { createCurrentRepositoryContext } from "../database/repositories/factory import { DataCrypto } from "../utils/data-crypto.js"; import { SystemCrypto } from "../utils/system-crypto.js"; import { DatabaseSaveTrigger } from "../utils/database-save-trigger.js"; +import { syncLogger } from "../utils/logger.js"; import { getEntity, type RegisteredSyncEntity, @@ -229,7 +230,13 @@ export async function loadWireRows( ); const wire = toWire(entity, serialized, syncId); rows.set(syncId, { syncId, wire, hash: hashWire(entity, wire, key) }); - } catch { + } catch (error) { + syncLogger.warn("Could not serialize a sync row", { + operation: "sync_serialize", + entityType: entity.type, + syncId, + error: error instanceof Error ? error.message : String(error), + }); failed.add(syncId); } } diff --git a/src/backend/tests/sync/two-way.test.ts b/src/backend/tests/sync/two-way.test.ts index 2a8325b0f..5524eb464 100644 --- a/src/backend/tests/sync/two-way.test.ts +++ b/src/backend/tests/sync/two-way.test.ts @@ -153,11 +153,14 @@ const { default: syncRouter } = await import("../../sync/server/routes.js"); const { SyncEngine } = await import("../../sync/client/engine.js"); const { createLink, getLink, updateLink } = await import("../../sync/client/link-store.js"); -const { markChanged, resetFeed } = await import("../../sync/server/feed.js"); +const { markChanged, reconcileUser, resetFeed } = + await import("../../sync/server/feed.js"); const { resetSeqAllocator, listConflicts, getConflict, deleteConflict } = await import("../../sync/records.js"); const { createResolvers, writeWireRow } = await import("../../sync/store.js"); const { getEntity } = await import("../../plugins/sync-registry.js"); +const { FieldCrypto } = await import("../../utils/field-crypto.js"); +const { syncLogger } = await import("../../utils/logger.js"); const app = express(); app.use(express.json({ limit: "5mb" })); @@ -487,6 +490,140 @@ describe("after linking", () => { }); }); +describe("a server row that cannot be read", () => { + async function recordOf(where: Side, syncId: string) { + const [record] = await on(where, () => + db(where) + .select() + .from(schema.syncRecords) + .where( + and( + eq(schema.syncRecords.entityType, "hosts"), + eq(schema.syncRecords.syncId, syncId), + ), + ), + ); + return record; + } + + it.each(["upsert", "delete"] as const)( + "retries a local %s after the server row becomes readable", + async (op) => { + const [desk] = desktops; + await on(server, () => addHost(server, { name: "a" })); + const engine = await linkDesktop(desk); + await pass(desk, engine); + + await on(desk, () => + (op === "delete" + ? db(desk).delete(schema.hosts) + : db(desk).update(schema.hosts).set({ ip: "desk-edit" }) + ).where(eq(schema.hosts.syncId, "host-a")), + ); + await on(server, async () => { + await db(server) + .update(schema.hosts) + .set({ ip: "server-edit" }) + .where(eq(schema.hosts.syncId, "host-a")); + await reconcileUser(SERVER_USER, { force: true }); + // Then its secret stops decrypting, as under a different key. + const [host] = await hostsOf(server); + await db(server) + .update(schema.hosts) + .set({ + password: FieldCrypto.encryptField( + "s3cret", + Buffer.alloc(32, 9), + String(host.id), + "password", + ), + }) + .where(eq(schema.hosts.id, host.id)); + }); + + const warn = vi.spyOn(syncLogger, "warn"); + try { + await pass(desk, engine); + expect(warn).toHaveBeenCalledWith( + expect.any(String), + expect.objectContaining({ + entityType: "hosts", + syncId: "host-a", + error: expect.any(String), + }), + ); + } finally { + warn.mockRestore(); + } + + const [local] = await on(desk, () => hostsOf(desk)); + expect(local?.ip).toBe(op === "upsert" ? "desk-edit" : undefined); + expect(await on(desk, () => listConflicts(desk.userId))).toEqual([]); + expect((await recordOf(desk, "host-a")).error).toBe("unavailable"); + + // Restore the same server content without another local edit or a new + // server revision. The failed operation must still be retried. + await on(server, () => + db(server) + .update(schema.hosts) + .set({ password: null }) + .where(eq(schema.hosts.syncId, "host-a")), + ); + await on(server, () => reconcileUser(SERVER_USER, { force: true })); + await pass(desk, engine); + const [after] = await on(desk, () => hostsOf(desk)); + expect(after?.ip).toBe("server-edit"); + expect((await recordOf(desk, "host-a")).error).toBeNull(); + const conflicts = await on(desk, () => listConflicts(desk.userId)); + expect(conflicts.map((c) => JSON.parse(c.localRow).ip)).toEqual( + op === "upsert" ? ["desk-edit"] : [], + ); + }, + ); + + it("keeps the local row when a conflict carries no row", async () => { + const [desk] = desktops; + await on(server, () => addHost(server, { name: "a" })); + const engine = await linkDesktop(desk); + await pass(desk, engine); + + await on(desk, async () => { + const link = (await getLink())!; + const settle = ( + engine as unknown as { + settle: (...args: unknown[]) => Promise; + } + ).settle.bind(engine); + await settle( + link, + { + entityType: "hosts", + syncId: "host-a", + baseRevision: 1, + op: "upsert", + row: { syncId: "host-a", name: "a" }, + }, + { + entityType: "hosts", + syncId: "host-a", + status: "conflict", + revision: 2, + deleted: false, + row: null, + }, + "hash", + ); + }); + + expect((await on(desk, () => hostsOf(desk))).map((h) => h.name)).toEqual([ + "a", + ]); + const record = await recordOf(desk, "host-a"); + expect(record.deleted).toBe(false); + expect(record.revision).toBe(1); + }); +}); + describe("hosts shared with the account", () => { it("arrive on the desktop with their text columns as stored", async () => { const [desk] = desktops;