fix supersync retry idempotency

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<ExistingSyncImport | null> =>
+  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<void> => {
         // - '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<void> => {
         const result = await syncService.runWithStorageUsageLock<UploadResult | null>(
           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<void> => {
               if (!quotaOk) return null;
             }
 
-            // Duplicate SYNC_IMPORT rejection already handled before the lock.
-
             const results = await syncService.uploadOps(
               userId,
               clientId,

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 (

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

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

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

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<string, unknown>)
+          .sort()
+          .map((key) => [
+            key,
+            this._toStableJsonValue((value as Record<string, unknown>)[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