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

@@ -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,

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;
     }

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);
     }
   }

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<void>;
+        }
+      ).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;