fix(sync): exclude duplicate ops from quota gate (#8597)

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) ` +

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<string, DuplicateOperationCandidate>(
+    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');

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(

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' });

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<typeof createOp>) => ({
+  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 = {

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, {