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