From 30da81ea5ae11ad324fa825569a248b468790781 Mon Sep 17 00:00:00 2001 From: johannesjo Date: Wed, 13 May 2026 00:14:35 +0200 Subject: [PATCH] fix supersync retry idempotency --- .../super-sync-server/src/sync/sync.routes.ts | 68 +++++++++++++------ .../src/sync/sync.service.ts | 13 ++-- .../duplicate-operation-precheck.spec.ts | 32 +++++++++ .../tests/sync-compressed-body.routes.spec.ts | 39 +++++++++++ .../super-sync/super-sync.spec.ts | 68 +++++++++++++++++++ .../sync-providers/super-sync/super-sync.ts | 57 ++++++++++++++++ 6 files changed, 250 insertions(+), 27 deletions(-) diff --git a/packages/super-sync-server/src/sync/sync.routes.ts b/packages/super-sync-server/src/sync/sync.routes.ts index ff3b6c8871..44416c7dcb 100644 --- a/packages/super-sync-server/src/sync/sync.routes.ts +++ b/packages/super-sync-server/src/sync/sync.routes.ts @@ -258,6 +258,42 @@ const sendQuotaExceededReply = ( ...body, }); +type ExistingSyncImport = { + id: string; + clientId: string; +}; + +const findExistingSyncImport = async ( + userId: number, +): Promise => + prisma.operation.findFirst({ + where: { + userId, + opType: { in: ['SYNC_IMPORT', 'BACKUP_IMPORT', 'REPAIR'] }, + }, + select: { id: true, clientId: true }, + }); + +const sendSyncImportExistsReply = ( + reply: FastifyReply, + userId: number, + clientId: string, + existingImport: ExistingSyncImport, +): FastifyReply => { + Logger.warn( + `[user:${userId}] Rejecting duplicate SYNC_IMPORT from client ${clientId}. ` + + `Existing import from client ${existingImport.clientId} (id: ${existingImport.id}). ` + + `Client should download and merge instead.`, + ); + return reply.status(409).send({ + error: 'SYNC_IMPORT_EXISTS', + errorCode: 'SYNC_IMPORT_EXISTS', + message: + 'A SYNC_IMPORT already exists. Download existing data and upload your changes as regular operations.', + existingImportId: existingImport.id, + }); +}; + type CompressedJsonBodyParseFailure = Extract< CompressedJsonBodyParseResult, { ok: false } @@ -961,27 +997,10 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise => { // - 'isCleanSlate': Password change or explicit clean slate request // Only 'initial' (first-time server migration) should be rejected if one exists. if (reason === 'initial' && !isCleanSlate) { - const existingImport = await prisma.operation.findFirst({ - where: { - userId, - opType: { in: ['SYNC_IMPORT', 'BACKUP_IMPORT', 'REPAIR'] }, - }, - select: { id: true, clientId: true }, - }); + const existingImport = await findExistingSyncImport(userId); if (existingImport) { - Logger.warn( - `[user:${userId}] Rejecting duplicate SYNC_IMPORT from client ${clientId}. ` + - `Existing import from client ${existingImport.clientId} (id: ${existingImport.id}). ` + - `Client should download and merge instead.`, - ); - return reply.status(409).send({ - error: 'SYNC_IMPORT_EXISTS', - errorCode: 'SYNC_IMPORT_EXISTS', - message: - 'A SYNC_IMPORT already exists. Download existing data and upload your changes as regular operations.', - existingImportId: existingImport.id, - }); + return sendSyncImportExistsReply(reply, userId, clientId, existingImport); } } @@ -1013,6 +1032,15 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise => { const result = await syncService.runWithStorageUsageLock( userId, async () => { + if (reason === 'initial' && !isCleanSlate) { + const existingImport = await findExistingSyncImport(userId); + + if (existingImport) { + sendSyncImportExistsReply(reply, userId, clientId, existingImport); + return null; + } + } + // Check storage quota before processing. For clean-slate uploads, use a // zero-current-usage baseline because uploadOps will wipe existing data // and reset storageUsedBytes inside its transaction. For regular @@ -1052,8 +1080,6 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise => { if (!quotaOk) return null; } - // Duplicate SYNC_IMPORT rejection already handled before the lock. - const results = await syncService.uploadOps( userId, clientId, diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index c5474f5145..c4b9897ac2 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -247,13 +247,14 @@ export class SyncService { return true; } - if (incomingOriginalTimestamp === incomingStoredTimestamp) { - return false; - } - + // A retry can arrive after server time advances enough that the original + // timestamp no longer needs clamping, so original===stored does not rule + // out a duplicate. // Future timestamps are clamped at receive time. A retry of the same op may - // be clamped to a later value, so allow exact-content duplicates whose stored - // timestamp came from an earlier clamp of the same original client timestamp. + // be clamped to a later value, or stop clamping once server time reaches the + // original timestamp's allowed drift window. Allow exact-content duplicates + // whose stored timestamp came from an earlier clamp of that same original + // client timestamp. const existingTimestampValue = BigInt(existingTimestamp); const existingReceivedAtValue = BigInt(existingReceivedAt); return ( diff --git a/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts b/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts index 1c06ec0905..42ad121a8a 100644 --- a/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts +++ b/packages/super-sync-server/tests/duplicate-operation-precheck.spec.ts @@ -253,6 +253,38 @@ describe('Duplicate Operation Pre-check', () => { } }); + it('should preserve clamped duplicate retries when the retry timestamp is no longer clamped', async () => { + const baseTimestamp = 1_700_000_000_000; + const retryTimestamp = baseTimestamp + 10_000; + const farFuture = baseTimestamp + DEFAULT_SYNC_CONFIG.maxClockDriftMs + 10_000; + + vi.useFakeTimers(); + vi.setSystemTime(baseTimestamp); + try { + const originalOp = createTestOp({ + id: 'clamp-boundary-duplicate-op', + timestamp: farFuture, + }); + const firstResult = await syncService.uploadOps(1, 'client-1', [originalOp]); + expect(firstResult[0]).toMatchObject({ accepted: true }); + + vi.setSystemTime(retryTimestamp); + const duplicateResult = await syncService.uploadOps(1, 'client-1', [ + createTestOp({ + id: 'clamp-boundary-duplicate-op', + timestamp: farFuture, + }), + ]); + + expect(duplicateResult[0]).toMatchObject({ + accepted: false, + errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION, + }); + } finally { + vi.useRealTimers(); + } + }); + it('should not advance server sequence for duplicate retries', async () => { const originalOp = createTestOp({ id: 'seq-original', diff --git a/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts b/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts index ce826e09f0..0b3a6ab700 100644 --- a/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts +++ b/packages/super-sync-server/tests/sync-compressed-body.routes.spec.ts @@ -25,9 +25,15 @@ const mocks = vi.hoisted(() => { getCachedSnapshotBytes: vi.fn(), markStorageNeedsReconcile: vi.fn(), }; + const prisma = { + operation: { + findFirst: vi.fn(), + }, + }; return { syncService, + prisma, notifyNewOps: vi.fn(), }; }); @@ -50,6 +56,10 @@ vi.mock('../src/sync/services/websocket-connection.service', () => ({ }), })); +vi.mock('../src/db', () => ({ + prisma: mocks.prisma, +})); + import { syncRoutes } from '../src/sync/sync.routes'; const gzipAsync = promisify(zlib.gzip); @@ -128,6 +138,7 @@ describe('Sync compressed body routes', () => { storageQuotaBytes: 100 * 1024 * 1024, }); mocks.syncService.getCachedSnapshotBytes.mockResolvedValue(0); + mocks.prisma.operation.findFirst.mockResolvedValue(null); app = Fastify(); await app.register(syncRoutes, { prefix: '/api/sync' }); @@ -548,6 +559,34 @@ describe('Sync compressed body routes', () => { expect(mocks.syncService.uploadOps).not.toHaveBeenCalled(); }); + it('should repeat initial snapshot duplicate detection inside the user lock', async () => { + const clientId = 'initial-race-client'; + mocks.prisma.operation.findFirst + .mockResolvedValueOnce(null) + .mockResolvedValueOnce({ id: 'existing-import', clientId: 'other-client' }); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/snapshot', + headers: { authorization: `Bearer ${authToken}` }, + payload: { + state: { TASK: { 'task-1': { id: 'task-1' } } }, + clientId, + reason: 'initial', + vectorClock: { [clientId]: 1 }, + }, + }); + + expect(response.statusCode).toBe(409); + expect(response.json()).toMatchObject({ + errorCode: 'SYNC_IMPORT_EXISTS', + existingImportId: 'existing-import', + }); + expect(mocks.prisma.operation.findFirst).toHaveBeenCalledTimes(2); + expect(mocks.syncService.checkStorageQuota).not.toHaveBeenCalled(); + expect(mocks.syncService.uploadOps).not.toHaveBeenCalled(); + }); + it('should reconcile the counter before rejecting on the cheap snapshot pre-gate', async () => { // Regression for W10: when the cached counter says we are at quota, the // route reconciles once before rejecting — a stale-high counter would diff --git a/src/app/op-log/sync-providers/super-sync/super-sync.spec.ts b/src/app/op-log/sync-providers/super-sync/super-sync.spec.ts index 2a01813d48..88304c7236 100644 --- a/src/app/op-log/sync-providers/super-sync/super-sync.spec.ts +++ b/src/app/op-log/sync-providers/super-sync/super-sync.spec.ts @@ -292,6 +292,8 @@ describe('SuperSyncProvider', () => { const body = JSON.parse(bodyJson); expect(body.ops).toEqual(ops); expect(body.clientId).toBe('client-1'); + expect(body.requestId).toMatch(/^ops-v1-/); + expect(body.requestId.length).toBeLessThanOrEqual(64); }); it('should include lastKnownServerSeq when provided', async () => { @@ -318,6 +320,72 @@ describe('SuperSyncProvider', () => { expect(body.lastKnownServerSeq).toBe(2); }); + it('should use a stable requestId for retried operation batches', async () => { + mockPrivateCfgStore.load.and.returnValue(Promise.resolve(testConfig)); + + fetchSpy.and.returnValue( + Promise.resolve({ + ok: true, + json: () => Promise.resolve({ results: [], latestSeq: 5 }), + } as Response), + ); + + const firstOp = createMockOperation({ + id: 'op-123', + payload: { top: 'value', nested: { a: 1, b: 2 } }, + vectorClock: { client1: 1, client2: 2 }, + }); + const secondOp = createMockOperation({ id: 'op-456', entityId: 'task-2' }); + const ops = [firstOp, secondOp]; + + await provider.uploadOps(ops, 'client-1', 2); + await provider.uploadOps(ops, 'client-1', 4); + await provider.uploadOps( + [ + { + ...firstOp, + payload: { nested: { b: 2, a: 1 }, top: 'value' }, + vectorClock: { client2: 2, client1: 1 }, + }, + secondOp, + ], + 'client-1', + 4, + ); + await provider.uploadOps( + [{ ...firstOp, payload: { title: 'Changed Task' } }, secondOp], + 'client-1', + 4, + ); + await provider.uploadOps( + [...ops, createMockOperation({ id: 'op-789', entityId: 'task-3' })], + 'client-1', + 4, + ); + + const firstBody = JSON.parse( + await decompressGzip(fetchSpy.calls.argsFor(0)[1].body), + ); + const retryBody = JSON.parse( + await decompressGzip(fetchSpy.calls.argsFor(1)[1].body), + ); + const reorderedContentBody = JSON.parse( + await decompressGzip(fetchSpy.calls.argsFor(2)[1].body), + ); + const changedContentBody = JSON.parse( + await decompressGzip(fetchSpy.calls.argsFor(3)[1].body), + ); + const changedBatchBody = JSON.parse( + await decompressGzip(fetchSpy.calls.argsFor(4)[1].body), + ); + + expect(retryBody.requestId).toBe(firstBody.requestId); + expect(reorderedContentBody.requestId).toBe(firstBody.requestId); + expect(changedContentBody.requestId).not.toBe(firstBody.requestId); + expect(changedBatchBody.requestId).not.toBe(firstBody.requestId); + expect(firstBody.requestId.length).toBeLessThanOrEqual(64); + }); + it('should throw MissingCredentialsSPError when config is missing', async () => { mockPrivateCfgStore.load.and.returnValue(Promise.resolve(null)); diff --git a/src/app/op-log/sync-providers/super-sync/super-sync.ts b/src/app/op-log/sync-providers/super-sync/super-sync.ts index 5302333afb..2f57aef7cf 100644 --- a/src/app/op-log/sync-providers/super-sync/super-sync.ts +++ b/src/app/op-log/sync-providers/super-sync/super-sync.ts @@ -36,6 +36,7 @@ import { } from './response-validators'; const LAST_SERVER_SEQ_KEY_PREFIX = 'super_sync_last_server_seq_'; +const OPS_UPLOAD_REQUEST_ID_PREFIX = 'ops-v1'; /** * Timeout for individual HTTP requests to SuperSync server. @@ -104,6 +105,61 @@ export class SuperSyncProvider } } + private _createOpsUploadRequestId(ops: SyncOperation[], clientId: string): string { + const opIds = ops.map((op) => op.id).join('|'); + let opsFingerprint = opIds; + try { + opsFingerprint = this._stableJsonStringify(ops); + } catch { + opsFingerprint = opIds; + } + const firstOpId = this._compactRequestIdPart(ops[0]?.id ?? 'empty'); + const lastOp = ops.length > 0 ? ops[ops.length - 1] : undefined; + const lastOpId = this._compactRequestIdPart(lastOp?.id ?? 'empty'); + const hash = this._hashRequestIdInput(`${clientId}|${opsFingerprint}`); + return `${OPS_UPLOAD_REQUEST_ID_PREFIX}-${ops.length}-${firstOpId}-${lastOpId}-${hash}`; + } + + private _compactRequestIdPart(id: string): string { + return id.replace(/[^a-zA-Z0-9_-]/g, '').slice(0, 8) || 'x'; + } + + private _hashRequestIdInput(input: string): string { + let hashA = 0x811c9dc5; + let hashB = 0x9e3779b9; + + for (let i = 0; i < input.length; i++) { + const code = input.charCodeAt(i); + hashA = Math.imul(hashA ^ code, 16777619); + hashB = Math.imul(hashB + code, 2246822519) ^ (hashB >>> 13); + } + + return `${(hashA >>> 0).toString(36)}${(hashB >>> 0).toString(36)}`; + } + + private _stableJsonStringify(value: unknown): string { + return JSON.stringify(this._toStableJsonValue(value)) ?? 'undefined'; + } + + private _toStableJsonValue(value: unknown): unknown { + if (Array.isArray(value)) { + return value.map((item) => this._toStableJsonValue(item)); + } + + if (value !== null && typeof value === 'object') { + return Object.fromEntries( + Object.keys(value as Record) + .sort() + .map((key) => [ + key, + this._toStableJsonValue((value as Record)[key]), + ]), + ); + } + + return value; + } + // === Operation Sync Implementation === async uploadOps( @@ -122,6 +178,7 @@ export class SuperSyncProvider ops, clientId, lastKnownServerSeq, + requestId: this._createOpsUploadRequestId(ops, clientId), }); // On native platforms (Android/iOS), use CapacitorHttp with base64-encoded gzip