From 3c82693cc375e62c98a87e166d91682ea94e4abc Mon Sep 17 00:00:00 2001 From: Symon Baikov Date: Fri, 26 Jun 2026 13:18:50 +0300 Subject: [PATCH] fix(sync): exclude duplicate ops from quota gate (#8597) --- .../src/sync/sync.routes.ops-handler.ts | 12 +- .../src/sync/sync.routes.quota.ts | 38 ++++- .../src/sync/sync.service.ts | 4 + ...quest-dedup-failure-caching.routes.spec.ts | 6 + .../tests/sync-compressed-body.routes.spec.ts | 160 ++++++++++++++++++ .../sync-upload-rate-limit.routes.spec.ts | 12 ++ 6 files changed, 228 insertions(+), 4 deletions(-) diff --git a/packages/super-sync-server/src/sync/sync.routes.ops-handler.ts b/packages/super-sync-server/src/sync/sync.routes.ops-handler.ts index cede763900..0024759d59 100644 --- a/packages/super-sync-server/src/sync/sync.routes.ops-handler.ts +++ b/packages/super-sync-server/src/sync/sync.routes.ops-handler.ts @@ -25,7 +25,7 @@ import { sendCompressedBodyParseFailure, } from './sync.routes.payload'; import { - computeOpsStorageBytes, + computeOpsStorageBytesExcludingKnownDuplicates, enforceStorageQuota, getRawOpsCount, sendOpsBatchTooLargeReply, @@ -168,10 +168,16 @@ export const uploadOpsHandler = async ( // Check storage quota before processing (after dedup to allow retries). // Account using the same per-op payload+vectorClock measure that the // post-accept counter increment uses, so the gate and the increment - // cannot disagree on what "size" means. + // cannot disagree on what "size" means. Already-stored exact + // duplicates are rejected by uploadOps and never written, so don't make + // quota cleanup reserve space for them. const typedOpsForGate = ops as unknown as Operation[]; const { bytes: estimatedDelta, fallback: gateFallback } = - computeOpsStorageBytes(typedOpsForGate); + await computeOpsStorageBytesExcludingKnownDuplicates( + userId, + typedOpsForGate, + syncService.getMaxClockDriftMs(), + ); if (gateFallback > 0) { Logger.warn( `computeOpsStorageBytes: ${gateFallback}/${typedOpsForGate.length} unserializable op(s) ` + diff --git a/packages/super-sync-server/src/sync/sync.routes.quota.ts b/packages/super-sync-server/src/sync/sync.routes.quota.ts index e1e7ed8137..3e850361fc 100644 --- a/packages/super-sync-server/src/sync/sync.routes.quota.ts +++ b/packages/super-sync-server/src/sync/sync.routes.quota.ts @@ -3,7 +3,13 @@ import { prisma } from '../db'; import { Logger } from '../logger'; import { getSyncService } from './sync.service'; import { computeOpStorageBytes } from './sync.const'; -import { SYNC_ERROR_CODES } from './sync.types'; +import { + DUPLICATE_OP_SELECT, + Operation, + SYNC_ERROR_CODES, + type DuplicateOperationCandidate, +} from './sync.types'; +import { isSameDuplicateOperation } from './conflict'; import { errorMessage, MAX_OPS_PER_BATCH } from './sync.routes.payload'; /** @@ -31,6 +37,36 @@ export const computeOpsStorageBytes = ( return { bytes, fallback }; }; +export const computeOpsStorageBytesExcludingKnownDuplicates = async ( + userId: number, + ops: Operation[], + maxClockDriftMs: number, +): Promise<{ bytes: number; fallback: number }> => { + if (ops.length === 0) return { bytes: 0, fallback: 0 }; + + const existingOps = await prisma.operation.findMany({ + where: { id: { in: Array.from(new Set(ops.map((op) => op.id))) } }, + select: DUPLICATE_OP_SELECT, + }); + const existingOpById = new Map( + existingOps.map((existingOp) => [existingOp.id, existingOp]), + ); + + let bytes = 0; + let fallback = 0; + for (const op of ops) { + const existingOp = existingOpById.get(op.id); + if (existingOp && isSameDuplicateOperation(existingOp, userId, op, maxClockDriftMs)) { + continue; + } + + const sized = computeOpStorageBytes(op); + bytes += sized.bytes; + if (sized.fallback) fallback += 1; + } + return { bytes, fallback }; +}; + export const computeJsonStorageBytes = (value: unknown, fallback: unknown): number => { try { return Buffer.byteLength(JSON.stringify(value ?? fallback), 'utf8'); diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index 775c9ac645..9ac22cd7ca 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -98,6 +98,10 @@ export class SyncService { ); } + getMaxClockDriftMs(): number { + return this.config.maxClockDriftMs; + } + // === Upload Operations === async uploadOps( diff --git a/packages/super-sync-server/tests/request-dedup-failure-caching.routes.spec.ts b/packages/super-sync-server/tests/request-dedup-failure-caching.routes.spec.ts index 4a9ca3ca8f..fae2d6bac2 100644 --- a/packages/super-sync-server/tests/request-dedup-failure-caching.routes.spec.ts +++ b/packages/super-sync-server/tests/request-dedup-failure-caching.routes.spec.ts @@ -36,11 +36,13 @@ const mocks = vi.hoisted(() => { getOpsSinceWithSeq: vi.fn(), getStorageInfo: vi.fn(), getCachedSnapshotBytes: vi.fn(), + getMaxClockDriftMs: vi.fn(), }; const prisma = { operation: { findFirst: vi.fn(), findUnique: vi.fn(), + findMany: vi.fn(), }, }; return { syncService, prisma, notifyNewOps: vi.fn() }; @@ -116,6 +118,7 @@ describe('Request dedup — transaction-failure results are not cached (#8332)', storageQuotaBytes: QUOTA, }); mocks.syncService.getCachedSnapshotBytes.mockResolvedValue(0); + mocks.syncService.getMaxClockDriftMs.mockReturnValue(60_000); mocks.syncService.cacheSnapshotIfReplayable.mockResolvedValue({ deltaBytes: 0 }); mocks.syncService.prepareSnapshotCache.mockResolvedValue({ stateBytes: 0, @@ -123,6 +126,7 @@ describe('Request dedup — transaction-failure results are not cached (#8332)', cacheable: true, }); mocks.prisma.operation.findFirst.mockResolvedValue(null); + mocks.prisma.operation.findMany.mockResolvedValue([]); app = Fastify(); await app.register(syncRoutes, { prefix: '/api/sync' }); @@ -277,6 +281,8 @@ describe('Request dedup round-trip with the real cache (#8332)', () => { }); mocks.syncService.getLatestSeq.mockResolvedValue(1); mocks.syncService.getOpsSinceWithSeq.mockResolvedValue({ ops: [], latestSeq: 1 }); + mocks.syncService.getMaxClockDriftMs.mockReturnValue(60_000); + mocks.prisma.operation.findMany.mockResolvedValue([]); app = Fastify(); await app.register(syncRoutes, { prefix: '/api/sync' }); 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 1e9959cb3e..3796f83fc9 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 @@ -26,11 +26,13 @@ const mocks = vi.hoisted(() => { getStorageInfo: vi.fn(), getCachedSnapshotBytes: vi.fn(), markStorageNeedsReconcile: vi.fn(), + getMaxClockDriftMs: vi.fn(), }; const prisma = { operation: { findFirst: vi.fn(), findUnique: vi.fn(), + findMany: vi.fn(), }, }; @@ -65,6 +67,7 @@ vi.mock('../src/db', () => ({ import { syncRoutes } from '../src/sync/sync.routes'; import { SYNC_ERROR_CODES } from '../src/sync/sync.types'; +import { computeOpStorageBytes } from '../src/sync/sync.const'; import { SUPER_SYNC_MAX_OPS_PER_UPLOAD } from '@sp/shared-schema'; const gzipAsync = promisify(zlib.gzip); @@ -82,6 +85,23 @@ const createOp = (clientId: string) => ({ schemaVersion: 1, }); +const createStoredDuplicateOp = (op: ReturnType) => ({ + id: op.id, + userId: 1, + clientId: op.clientId, + actionType: op.actionType, + opType: op.opType, + entityType: op.entityType, + entityId: op.entityId, + payload: op.payload, + vectorClock: op.vectorClock, + schemaVersion: op.schemaVersion, + clientTimestamp: BigInt(op.timestamp), + receivedAt: BigInt(op.timestamp), + isPayloadEncrypted: false, + syncImportReason: null, +}); + const MiB = 1024 * 1024; describe('Sync compressed body routes', () => { @@ -93,6 +113,7 @@ describe('Sync compressed body routes', () => { mocks.syncService.isRateLimited.mockReturnValue(false); mocks.syncService.checkOpsRequestDedup.mockReturnValue(null); mocks.syncService.checkSnapshotRequestDedup.mockReturnValue(null); + mocks.syncService.getMaxClockDriftMs.mockReturnValue(60_000); mocks.syncService.checkStorageQuota.mockResolvedValue({ allowed: true, currentUsage: 0, @@ -145,6 +166,7 @@ describe('Sync compressed body routes', () => { }); mocks.syncService.getCachedSnapshotBytes.mockResolvedValue(0); mocks.prisma.operation.findFirst.mockResolvedValue(null); + mocks.prisma.operation.findMany.mockResolvedValue([]); app = Fastify(); await app.register(syncRoutes, { prefix: '/api/sync' }); @@ -189,6 +211,144 @@ describe('Sync compressed body routes', () => { expect(quotaCall[1]).toBeLessThan(payloadSize); }); + it('should subtract exact already-stored duplicate ops from the ops quota gate', async () => { + const clientId = 'known-duplicate-quota-client'; + const duplicateOp = { + ...createOp(clientId), + id: 'known-duplicate-op', + entityId: 'task-known-duplicate', + payload: { title: 'Already accepted task' }, + }; + const newOp = { + ...createOp(clientId), + id: 'new-op', + entityId: 'task-new', + payload: { title: 'New task' }, + }; + mocks.prisma.operation.findMany.mockResolvedValueOnce([ + createStoredDuplicateOp(duplicateOp), + ]); + mocks.syncService.uploadOps.mockResolvedValueOnce([ + { + opId: duplicateOp.id, + accepted: false, + error: 'Duplicate operation ID', + errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION, + }, + { opId: newOp.id, accepted: true, serverSeq: 2 }, + ]); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/ops', + headers: { + authorization: `Bearer ${authToken}`, + 'content-type': 'application/json', + }, + payload: { + ops: [duplicateOp, newOp], + clientId, + }, + }); + + expect(response.statusCode).toBe(200); + expect(mocks.prisma.operation.findMany).toHaveBeenCalledOnce(); + expect(mocks.syncService.checkStorageQuota).toHaveBeenCalledWith( + 1, + computeOpStorageBytes(newOp).bytes, + ); + expect(mocks.syncService.uploadOps).toHaveBeenCalledWith(1, clientId, [ + duplicateOp, + newOp, + ]); + }); + + it('should keep same-id different-content ops charged in the ops quota gate', async () => { + const clientId = 'id-collision-quota-client'; + const incomingOp = { + ...createOp(clientId), + id: 'colliding-op-id', + entityId: 'task-collision', + payload: { title: 'Incoming content' }, + }; + const storedDifferentOp = createStoredDuplicateOp({ + ...incomingOp, + payload: { title: 'Stored different content' }, + }); + mocks.prisma.operation.findMany.mockResolvedValueOnce([storedDifferentOp]); + mocks.syncService.uploadOps.mockResolvedValueOnce([ + { + opId: incomingOp.id, + accepted: false, + error: 'Operation ID already belongs to a different operation', + errorCode: SYNC_ERROR_CODES.INVALID_OP_ID, + }, + ]); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/ops', + headers: { + authorization: `Bearer ${authToken}`, + 'content-type': 'application/json', + }, + payload: { + ops: [incomingOp], + clientId, + }, + }); + + expect(response.statusCode).toBe(200); + expect(mocks.syncService.checkStorageQuota).toHaveBeenCalledWith( + 1, + computeOpStorageBytes(incomingOp).bytes, + ); + }); + + it('should not trigger cleanup for duplicate-only ops uploads near quota', async () => { + const clientId = 'duplicate-only-quota-client'; + const duplicateOp = { + ...createOp(clientId), + id: 'duplicate-only-op', + entityId: 'task-duplicate-only', + }; + mocks.prisma.operation.findMany.mockResolvedValueOnce([ + createStoredDuplicateOp(duplicateOp), + ]); + mocks.syncService.checkStorageQuota.mockImplementation( + async (_userId: number, storageDeltaBytes: number) => ({ + allowed: storageDeltaBytes === 0, + currentUsage: 100, + quota: 100, + }), + ); + mocks.syncService.uploadOps.mockResolvedValueOnce([ + { + opId: duplicateOp.id, + accepted: false, + error: 'Duplicate operation ID', + errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION, + }, + ]); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/ops', + headers: { + authorization: `Bearer ${authToken}`, + 'content-type': 'application/json', + }, + payload: { + ops: [duplicateOp], + clientId, + }, + }); + + expect(response.statusCode).toBe(200); + expect(mocks.syncService.checkStorageQuota).toHaveBeenCalledWith(1, 0); + expect(mocks.syncService.freeStorageForUpload).not.toHaveBeenCalled(); + }); + it('should reject oversized op batches before schema validation', async () => { const clientId = 'too-many-ops-client'; const payload = { diff --git a/packages/super-sync-server/tests/sync-upload-rate-limit.routes.spec.ts b/packages/super-sync-server/tests/sync-upload-rate-limit.routes.spec.ts index 9936f4ee9d..187a75f068 100644 --- a/packages/super-sync-server/tests/sync-upload-rate-limit.routes.spec.ts +++ b/packages/super-sync-server/tests/sync-upload-rate-limit.routes.spec.ts @@ -13,12 +13,18 @@ const mocks = vi.hoisted(() => { runWithStorageUsageLock: vi.fn(), getLatestSeq: vi.fn(), getOpsSinceWithSeq: vi.fn(), + getMaxClockDriftMs: vi.fn(), }; return { syncService, uploadCountsByUser, notifyNewOps: vi.fn(), + prisma: { + operation: { + findMany: vi.fn(), + }, + }, verifyToken: vi.fn(), }; }); @@ -37,6 +43,10 @@ vi.mock('../src/sync/services/websocket-connection.service', () => ({ }), })); +vi.mock('../src/db', () => ({ + prisma: mocks.prisma, +})); + import { syncRoutes } from '../src/sync/sync.routes'; const CUSTOM_UPLOAD_LIMIT = 3; @@ -95,6 +105,8 @@ describe('Sync upload route rate limiting', () => { ops: [], latestSeq: 1, }); + mocks.syncService.getMaxClockDriftMs.mockReturnValue(60_000); + mocks.prisma.operation.findMany.mockResolvedValue([]); app = Fastify(); await app.register(rateLimit, {