diff --git a/cloudflare/src/sync_push.ts b/cloudflare/src/sync_push.ts index 4bc0169..c6a73f0 100644 --- a/cloudflare/src/sync_push.ts +++ b/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 || diff --git a/cloudflare/tests/sync_push_routes.test.ts b/cloudflare/tests/sync_push_routes.test.ts index 55b275c..b522518 100644 --- a/cloudflare/tests/sync_push_routes.test.ts +++ b/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] });