Reject same-clock sync push races

cloudflare/src/sync_push.ts

@@ -61,7 +61,22 @@ const SYNC_OBJECT_UPSERT_QUERY = `
     device_id = excluded.device_id,
     updated_at = excluded.updated_at,
     deleted_at = excluded.deleted_at
-  WHERE excluded.logical_clock >= sync_objects.logical_clock
+  WHERE excluded.logical_clock > sync_objects.logical_clock
+    OR (
+      excluded.logical_clock = sync_objects.logical_clock
+      AND sync_objects.object_type = excluded.object_type
+      AND (
+        (sync_objects.payload_r2_key IS NULL AND excluded.payload_r2_key IS NULL)
+        OR sync_objects.payload_r2_key = excluded.payload_r2_key
+      )
+      AND sync_objects.payload_hash = excluded.payload_hash
+      AND sync_objects.schema_rev = excluded.schema_rev
+      AND sync_objects.device_id = excluded.device_id
+      AND (
+        (sync_objects.deleted_at IS NULL AND excluded.deleted_at IS NULL)
+        OR (sync_objects.deleted_at IS NOT NULL AND excluded.deleted_at IS NOT NULL)
+      )
+    )
 `;
 const SYNC_CHANGE_INSERT_QUERY = `
   INSERT INTO sync_change_log (
@@ -241,6 +256,17 @@ function assertSavedObjectMatchesPush(
   if (object.logical_clock > push.logicalClock) {
     throw new SyncPushConflictError("logical_clock_stale");
   }
+  if (
+    object.logical_clock === push.logicalClock &&
+    (object.object_type !== push.objectType ||
+      object.operation !== push.operation ||
+      object.payload_hash !== push.payloadHash ||
+      object.schema_rev !== push.schemaRev ||
+      object.device_id !== deviceId ||
+      object.payload_r2_key !== push.payload.r2Key)
+  ) {
+    throw new SyncPushConflictError("logical_clock_conflict");
+  }
   if (
     object.object_id !== push.objectId ||
     object.object_type !== push.objectType ||

cloudflare/tests/sync_push_routes.test.ts

@@ -212,6 +212,31 @@ describe("sync push routes", () => {
     assert.deepEqual(d1.batches, []);
   });
 
+  it("rejects same-clock object write races after D1 persistence", async () => {
+    const payload = bytes("encrypted tab payload");
+    const payloadHash = sha256(payload);
+    const tokenHash = await authTokenHash(ACCESS_TOKEN);
+    const d1 = testD1Database({
+      firstRows: [
+        { device_id: DEVICE_ID },
+        null,
+        syncObjectRow({ payload_hash: "e".repeat(64), logical_clock: 42 }),
+      ],
+    });
+
+    const response = await handleRequest(
+      syncPushRequest(syncPushBody({ payload_hash: payloadHash, payload: inlinePayload(payload) })),
+      testEnv({
+        d1,
+        kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]],
+      }),
+    );
+
+    assert.equal(response.status, 409);
+    assert.deepEqual(await response.json(), { error: "sync_conflict" });
+    assert.equal(d1.batches[0], 2);
+  });
+
   it("rejects revoked devices before reading the sync push body", async () => {
     const tokenHash = await authTokenHash(ACCESS_TOKEN);
     const d1 = testD1Database({ firstRows: [null] });