diff --git a/packages/shared-schema/src/supersync-http-contract.ts b/packages/shared-schema/src/supersync-http-contract.ts index 5b7e393283..c9a0099786 100644 --- a/packages/shared-schema/src/supersync-http-contract.ts +++ b/packages/shared-schema/src/supersync-http-contract.ts @@ -112,19 +112,29 @@ export const SuperSyncDownloadOpsQuerySchema = z.object({ excludeClient: SuperSyncClientIdSchema.optional(), }); -export const SuperSyncUploadSnapshotRequestSchema = z.object({ - state: z.unknown(), - clientId: SuperSyncClientIdSchema, - reason: z.enum(SUPER_SYNC_SNAPSHOT_REASONS), - vectorClock: SuperSyncVectorClockSchema, - schemaVersion: z.number().int().min(1).max(100).optional(), - isPayloadEncrypted: z.boolean().optional(), - syncImportReason: z.enum(SUPER_SYNC_IMPORT_REASONS).optional(), - opId: z.string().uuid().optional(), - isCleanSlate: z.boolean().optional(), - snapshotOpType: z.enum(SUPER_SYNC_SNAPSHOT_OP_TYPES).optional(), - requestId: SuperSyncRequestIdSchema.optional(), -}); +export const SuperSyncUploadSnapshotRequestSchema = z + .object({ + state: z.unknown(), + clientId: SuperSyncClientIdSchema, + reason: z.enum(SUPER_SYNC_SNAPSHOT_REASONS), + vectorClock: SuperSyncVectorClockSchema, + schemaVersion: z.number().int().min(1).max(100).optional(), + isPayloadEncrypted: z.boolean().optional(), + syncImportReason: z.enum(SUPER_SYNC_IMPORT_REASONS).optional(), + opId: z.string().uuid().optional(), + isCleanSlate: z.boolean().optional(), + snapshotOpType: z.enum(SUPER_SYNC_SNAPSHOT_OP_TYPES).optional(), + requestId: SuperSyncRequestIdSchema.optional(), + }) + .superRefine((request, context) => { + if (request.isCleanSlate && !request.opId) { + context.addIssue({ + code: 'custom', + path: ['opId'], + message: 'opId is required for clean-slate snapshot idempotency', + }); + } + }); export const SuperSyncOperationResponseSchema = SuperSyncOperationSchema.passthrough(); diff --git a/packages/shared-schema/tests/supersync-http-contract.spec.ts b/packages/shared-schema/tests/supersync-http-contract.spec.ts index c8ff1a9582..34048852f8 100644 --- a/packages/shared-schema/tests/supersync-http-contract.spec.ts +++ b/packages/shared-schema/tests/supersync-http-contract.spec.ts @@ -188,6 +188,19 @@ describe('SuperSync HTTP contract schemas', () => { expect(parsed.requestId).toBe('snapshot-v1-request'); }); + it('requires an operation ID for destructive clean-slate snapshots', () => { + expect(() => + SuperSyncUploadSnapshotRequestSchema.parse({ + state: {}, + clientId: 'client_1', + reason: 'recovery', + vectorClock: { client_1: 1 }, + schemaVersion: 1, + isCleanSlate: true, + }), + ).toThrow(); + }); + it('rejects requestIds containing characters outside the safe-log charset', () => { // Control character — would be unsafe to embed in server log lines. expect(() => diff --git a/packages/super-sync-server/src/sync/services/operation-upload.service.ts b/packages/super-sync-server/src/sync/services/operation-upload.service.ts index 5b7750ae4c..6e366ae0f3 100644 --- a/packages/super-sync-server/src/sync/services/operation-upload.service.ts +++ b/packages/super-sync-server/src/sync/services/operation-upload.service.ts @@ -1,6 +1,10 @@ import { Prisma } from '@prisma/client'; import { Logger } from '../../logger'; -import { computeOpStorageBytes } from '../sync.const'; +import { + CLIENT_ID_REGEX, + computeOpStorageBytes, + MAX_CLIENT_ID_LENGTH, +} from '../sync.const'; import { AcceptedBatchOperation, BatchUploadCandidate, @@ -11,6 +15,7 @@ import { isFullStateOpType, limitVectorClockSize, Operation, + OP_TYPES, ProcessOperationResult, SyncConfig, SYNC_ERROR_CODES, @@ -29,13 +34,58 @@ import { pruneVectorClockForStorage, resolveConflictForExistingOp, } from '../conflict'; -import { ValidationService, type ValidationResult } from './validation.service'; +import { + ALLOWED_ENTITY_TYPES, + ValidationService, + type ValidationResult, +} from './validation.service'; // Observability threshold: log a warning when the full-state op aggregate scan // exceeds this duration. Mirrors the threshold used by the legacy snapshot // vector-clock aggregate in OperationDownloadService so production logs use a // consistent slow-aggregate signal. const SLOW_FULL_STATE_AGGREGATE_MS = 5_000; +const INVALID_AUDIT_FIELD = '[invalid]'; +const SAFE_AUDIT_ID_REGEX = /^[A-Za-z0-9_-]+$/; + +const isSafeAuditIdentifier = (value: unknown, maxLength: number): value is string => + typeof value === 'string' && + value.length > 0 && + value.length <= maxLength && + SAFE_AUDIT_ID_REGEX.test(value); + +const getSafeAuditOperationMetadata = ( + op: Operation, +): { opId: string; entityType: string; entityId?: string; opType: string } => { + const rawOp = op as unknown as Record; + const rawOpId = rawOp['id']; + const rawEntityType = rawOp['entityType']; + const rawEntityId = rawOp['entityId']; + const rawOpType = rawOp['opType']; + + return { + opId: isSafeAuditIdentifier(rawOpId, 255) ? rawOpId : INVALID_AUDIT_FIELD, + entityType: + typeof rawEntityType === 'string' && ALLOWED_ENTITY_TYPES.has(rawEntityType) + ? rawEntityType + : INVALID_AUDIT_FIELD, + entityId: + rawEntityId === undefined || rawEntityId === null + ? undefined + : isSafeAuditIdentifier(rawEntityId, 255) + ? rawEntityId + : INVALID_AUDIT_FIELD, + opType: + typeof rawOpType === 'string' && OP_TYPES.includes(rawOpType as Operation['opType']) + ? rawOpType + : INVALID_AUDIT_FIELD, + }; +}; + +const getSafeAuditClientId = (clientId: string): string => + clientId.length <= MAX_CLIENT_ID_LENGTH && CLIENT_ID_REGEX.test(clientId) + ? clientId + : INVALID_AUDIT_FIELD; const toSafeServerSeq = (value: number | bigint | undefined, userId: number): number => { if (typeof value === 'bigint') { @@ -75,9 +125,8 @@ export class OperationUploadService { Logger.audit({ event: 'TIMESTAMP_CLAMPED', userId, - clientId, - opId: op.id, - entityType: op.entityType, + clientId: getSafeAuditClientId(clientId), + ...getSafeAuditOperationMetadata(op), originalTimestamp, clampedTo: maxAllowedTimestamp, driftMs: originalTimestamp - now, @@ -97,13 +146,10 @@ export class OperationUploadService { Logger.audit({ event: 'OP_REJECTED', userId, - clientId, - opId: op.id, - entityType: op.entityType, - entityId: op.entityId, + clientId: getSafeAuditClientId(clientId), + ...getSafeAuditOperationMetadata(op), errorCode, - reason: error, - opType: op.opType, + reason: errorCode ?? 'OP_REJECTED', }); return { @@ -281,8 +327,9 @@ export class OperationUploadService { } /** - * Stage 1: clamp future timestamps and validate every op in memory (no DB). - * Invalid ops get a terminal rejection written into `results` by index. + * Stage 1: reserve each transport-level operation ID, clamp timestamps, and + * validate every first occurrence in memory (no DB). Invalid first ops and + * every later same-ID sibling get terminal results by index. */ private validateAndClampBatch( userId: number, @@ -293,9 +340,36 @@ export class OperationUploadService { prevalidatedResults?: ReadonlyMap, ): BatchUploadCandidate[] { const validatedCandidates: BatchUploadCandidate[] = []; + const firstOperationById = new Map< + string, + { op: Operation; originalTimestamp: number } + >(); for (let i = 0; i < ops.length; i++) { const op = ops[i]; const originalTimestamp = this.clampFutureTimestamp(userId, clientId, op, now); + const firstOperation = firstOperationById.get(op.id); + if (firstOperation) { + const isExactRetry = isSameIncomingOperation( + firstOperation.op, + op, + firstOperation.originalTimestamp, + originalTimestamp, + ); + results[i] = this.rejectedUploadResult( + userId, + clientId, + op, + isExactRetry + ? 'Duplicate operation ID' + : 'Operation ID already belongs to a different operation', + isExactRetry + ? SYNC_ERROR_CODES.DUPLICATE_OPERATION + : SYNC_ERROR_CODES.INVALID_OP_ID, + ); + continue; + } + firstOperationById.set(op.id, { op, originalTimestamp }); + const validation = prevalidatedResults?.get(op) ?? this.validationService.validateOp(op, clientId); @@ -634,6 +708,7 @@ export class OperationUploadService { tx: Prisma.TransactionClient, prevalidatedResult?: ValidationResult, wasOccupiedAtRequestStart?: boolean, + firstRequestOperation?: { op: Operation; originalTimestamp: number }, ): Promise { // Rejected ops have no storage cost; the caller only reads storageBytes when // result.accepted is true. @@ -650,25 +725,40 @@ export class OperationUploadService { // Validate operation (including clientId match) const validation = prevalidatedResult ?? this.validationService.validateOp(op, clientId); - if (!validation.valid) { - Logger.audit({ - event: 'OP_REJECTED', - userId, - clientId, - opId: op.id, - entityType: op.entityType, - entityId: op.entityId, - errorCode: validation.errorCode, - reason: validation.error, - opType: op.opType, - }); - return reject({ - opId: op.id, - accepted: false, - error: validation.error, - errorCode: validation.errorCode, - }); + if (firstRequestOperation) { + const isExactRetry = isSameIncomingOperation( + firstRequestOperation.op, + op, + firstRequestOperation.originalTimestamp, + originalTimestamp, + ); + return reject( + this.rejectedUploadResult( + userId, + clientId, + op, + isExactRetry + ? 'Duplicate operation ID' + : 'Operation ID already belongs to a different operation', + isExactRetry + ? SYNC_ERROR_CODES.DUPLICATE_OPERATION + : SYNC_ERROR_CODES.INVALID_OP_ID, + ), + ); } + + if (!validation.valid) { + return reject( + this.rejectedUploadResult( + userId, + clientId, + op, + validation.error, + validation.errorCode, + ), + ); + } + // Capture the *unpruned* vector clock for full-state ops. The op row stores // the pruned clock (see `limitVectorClockSize` call below); persisting the // unpruned copy on `user_sync_state` lets the download path re-prune at @@ -696,42 +786,26 @@ export class OperationUploadService { originalTimestamp, ) ) { - Logger.audit({ - event: 'OP_REJECTED', - userId, - clientId, - opId: op.id, - entityType: op.entityType, - entityId: op.entityId, - errorCode: SYNC_ERROR_CODES.INVALID_OP_ID, - reason: 'Operation ID already belongs to a different operation', - opType: op.opType, - }); - return reject({ - opId: op.id, - accepted: false, - error: 'Operation ID already belongs to a different operation', - errorCode: SYNC_ERROR_CODES.INVALID_OP_ID, - }); + return reject( + this.rejectedUploadResult( + userId, + clientId, + op, + 'Operation ID already belongs to a different operation', + SYNC_ERROR_CODES.INVALID_OP_ID, + ), + ); } - Logger.audit({ - event: 'OP_REJECTED', - userId, - clientId, - opId: op.id, - entityType: op.entityType, - entityId: op.entityId, - errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION, - reason: 'Duplicate operation ID (pre-check)', - opType: op.opType, - }); - return reject({ - opId: op.id, - accepted: false, - error: 'Duplicate operation ID', - errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION, - }); + return reject( + this.rejectedUploadResult( + userId, + clientId, + op, + 'Duplicate operation ID', + SYNC_ERROR_CODES.DUPLICATE_OPERATION, + ), + ); } if (wasOccupiedAtRequestStart) { @@ -754,24 +828,16 @@ export class OperationUploadService { conflict.conflictType === 'equal_different_client' ? SYNC_ERROR_CODES.CONFLICT_CONCURRENT : SYNC_ERROR_CODES.CONFLICT_SUPERSEDED; - Logger.audit({ - event: 'OP_REJECTED', - userId, - clientId, - opId: op.id, - entityType: op.entityType, - entityId: op.entityId, - errorCode, - reason: conflict.reason, - opType: op.opType, - }); - return reject({ - opId: op.id, - accepted: false, - error: conflict.reason, - errorCode, - existingClock: conflict.existingClock, - }); + return reject( + this.rejectedUploadResult( + userId, + clientId, + op, + conflict.reason, + errorCode, + conflict.existingClock, + ), + ); } // Get next sequence number @@ -797,24 +863,16 @@ export class OperationUploadService { finalConflict.conflictType === 'equal_different_client' ? SYNC_ERROR_CODES.CONFLICT_CONCURRENT : SYNC_ERROR_CODES.CONFLICT_SUPERSEDED; - Logger.audit({ - event: 'OP_REJECTED', - userId, - clientId, - opId: op.id, - entityType: op.entityType, - entityId: op.entityId, - errorCode, - reason: `[RACE] ${finalConflict.reason}`, - opType: op.opType, - }); - return reject({ - opId: op.id, - accepted: false, - error: finalConflict.reason, - errorCode, - existingClock: finalConflict.existingClock, - }); + return reject( + this.rejectedUploadResult( + userId, + clientId, + op, + finalConflict.reason, + errorCode, + finalConflict.existingClock, + ), + ); } // Prune vector clock AFTER conflict detection but BEFORE storage. @@ -896,42 +954,26 @@ export class OperationUploadService { originalTimestamp, ) ) { - Logger.audit({ - event: 'OP_REJECTED', - userId, - clientId, - opId: op.id, - entityType: op.entityType, - entityId: op.entityId, - errorCode: SYNC_ERROR_CODES.INVALID_OP_ID, - reason: 'Operation ID already belongs to a different operation', - opType: op.opType, - }); - return reject({ - opId: op.id, - accepted: false, - error: 'Operation ID already belongs to a different operation', - errorCode: SYNC_ERROR_CODES.INVALID_OP_ID, - }); + return reject( + this.rejectedUploadResult( + userId, + clientId, + op, + 'Operation ID already belongs to a different operation', + SYNC_ERROR_CODES.INVALID_OP_ID, + ), + ); } - Logger.audit({ - event: 'OP_REJECTED', - userId, - clientId, - opId: op.id, - entityType: op.entityType, - entityId: op.entityId, - errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION, - reason: 'Duplicate operation ID (insert race)', - opType: op.opType, - }); - return reject({ - opId: op.id, - accepted: false, - error: 'Duplicate operation ID', - errorCode: SYNC_ERROR_CODES.DUPLICATE_OPERATION, - }); + return reject( + this.rejectedUploadResult( + userId, + clientId, + op, + 'Duplicate operation ID', + SYNC_ERROR_CODES.DUPLICATE_OPERATION, + ), + ); } if (fullStateVectorClock) { 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 568ad1c689..feda3e9be7 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 @@ -250,8 +250,8 @@ export const uploadOpsHandler = async ( if (rejected > 0) { Logger.debug( - `[user:${userId}] Rejected ops:`, - results.filter((r) => !r.accepted), + `[user:${userId}] Rejected op error codes:`, + results.filter((r) => !r.accepted).map((r) => r.errorCode ?? 'UNKNOWN'), ); } diff --git a/packages/super-sync-server/src/sync/sync.routes.snapshot-handler.ts b/packages/super-sync-server/src/sync/sync.routes.snapshot-handler.ts index 281e695406..d5b718887c 100644 --- a/packages/super-sync-server/src/sync/sync.routes.snapshot-handler.ts +++ b/packages/super-sync-server/src/sync/sync.routes.snapshot-handler.ts @@ -6,7 +6,13 @@ import { Logger } from '../logger'; import { getAuthUser } from '../middleware'; import { getSyncService } from './sync.service'; import { getWsConnectionService } from './services/websocket-connection.service'; -import { SYNC_ERROR_CODES, UploadResult } from './sync.types'; +import { + DUPLICATE_OP_SELECT, + Operation, + SYNC_ERROR_CODES, + UploadResult, +} from './sync.types'; +import { isSameIncomingOperation } from './conflict'; import { isSingleTokenGzipEncoding, parseCompressedJsonBody, @@ -222,6 +228,57 @@ export const uploadSnapshotHandler = async ( const result = await syncService.runWithStorageUsageLock( userId, async () => { + // Clean-slate uploads are destructive, so request-cache deduplication is + // not sufficient: it is process-local and expires. The client-supplied + // opId is the durable idempotency key. Check it inside the per-user lock + // before quota work or uploadOps can delete any existing data. + if (isCleanSlate && opId) { + const existingCleanSlateOp = await prisma.operation.findUnique({ + where: { id: opId }, + select: { ...DUPLICATE_OP_SELECT, serverSeq: true }, + }); + if (existingCleanSlateOp) { + const existingOperation: Operation = { + id: existingCleanSlateOp.id, + clientId: existingCleanSlateOp.clientId, + actionType: existingCleanSlateOp.actionType, + opType: existingCleanSlateOp.opType as Operation['opType'], + entityType: existingCleanSlateOp.entityType, + entityId: existingCleanSlateOp.entityId ?? undefined, + entityIds: existingCleanSlateOp.entityIds, + payload: existingCleanSlateOp.payload, + vectorClock: existingCleanSlateOp.vectorClock as Operation['vectorClock'], + timestamp: op.timestamp, + schemaVersion: existingCleanSlateOp.schemaVersion, + isPayloadEncrypted: existingCleanSlateOp.isPayloadEncrypted, + syncImportReason: existingCleanSlateOp.syncImportReason ?? undefined, + }; + const isExactRetry = + existingCleanSlateOp.userId === userId && + (existingCleanSlateOp.opType === 'SYNC_IMPORT' || + existingCleanSlateOp.opType === 'BACKUP_IMPORT' || + existingCleanSlateOp.opType === 'REPAIR') && + isSameIncomingOperation(existingOperation, op, 0, 0); + if (isExactRetry) { + Logger.info( + `[user:${userId}] Idempotent clean-slate retry from client ${clientId} ` + + `for existing op seq=${existingCleanSlateOp.serverSeq}`, + ); + return { + opId, + accepted: true, + serverSeq: existingCleanSlateOp.serverSeq, + }; + } + return { + opId, + accepted: false, + error: 'Operation ID already belongs to a different operation', + errorCode: SYNC_ERROR_CODES.INVALID_OP_ID, + }; + } + } + if (reason === 'initial' && !isCleanSlate) { const existingImport = await findExistingSyncImport(userId, opId); diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index a1d3697e7c..81b3f4ba1d 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -111,10 +111,13 @@ export class SyncService { * and block otherwise valid operations in the same request. */ filterValidOpsForQuota(ops: Operation[], clientId: string): Operation[] { + const seenOperationIds = new Set(); return ops.filter((op) => { + const isFirstOccurrence = !seenOperationIds.has(op.id); + seenOperationIds.add(op.id); const validation = this.validationService.validateOp(op, clientId); this.prevalidatedOps.set(op, validation); - return validation.valid; + return isFirstOccurrence && validation.valid; }); } @@ -218,7 +221,24 @@ export class SyncService { }); uploadDbRoundtrips++; + const firstOperationById = new Map< + string, + { op: Operation; originalTimestamp: number } + >(); + for (const op of ops) { + const firstRequestOperation = firstOperationById.get(op.id); + if (!firstRequestOperation) { + firstOperationById.set(op.id, { + op, + originalTimestamp: op.timestamp, + }); + } + const validation = + prevalidatedResults.get(op) ?? + this.validationService.validateOp(op, clientId); + prevalidatedResults.set(op, validation); + const { result, storageBytes, fallback } = await this.operationUploadService.processOperation( userId, @@ -226,8 +246,9 @@ export class SyncService { op, now, tx, - prevalidatedResults.get(op), + validation, requestStartOccupiedIds?.has(op.id), + firstRequestOperation, ); results.push(result); if (result.accepted) { 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 b4a112d193..ce8a9d1ee3 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 @@ -70,7 +70,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'; +import { CURRENT_SCHEMA_VERSION, SUPER_SYNC_MAX_OPS_PER_UPLOAD } from '@sp/shared-schema'; const gzipAsync = promisify(zlib.gzip); @@ -170,6 +170,7 @@ describe('Sync compressed body routes', () => { }); mocks.syncService.getCachedSnapshotBytes.mockResolvedValue(0); mocks.prisma.operation.findFirst.mockResolvedValue(null); + mocks.prisma.operation.findUnique.mockReset().mockResolvedValue(null); mocks.prisma.operation.findMany.mockResolvedValue([]); app = Fastify(); @@ -905,6 +906,7 @@ describe('Sync compressed body routes', () => { clientId, reason: 'initial', vectorClock: {}, + opId: '018f2f0b-1c2d-7a1b-8c3d-123456789abc', isCleanSlate: true, }, }); @@ -914,6 +916,98 @@ describe('Sync compressed body routes', () => { expect(mocks.syncService.uploadOps).not.toHaveBeenCalled(); }); + it('should return persistent success for a clean-slate retry before deleting data', async () => { + const opId = '018f2f0b-1c2d-7a1b-8c3d-123456789abc'; + const clientId = 'clean-slate-retry-client'; + const state = { TASK: { 'task-1': { id: 'task-1' } } }; + const vectorClock = { [clientId]: 1 }; + mocks.prisma.operation.findUnique.mockResolvedValueOnce({ + id: opId, + userId: 1, + clientId, + actionType: '[SP_ALL] Load(import) all data', + opType: 'SYNC_IMPORT', + entityType: 'ALL', + entityId: null, + entityIds: [], + payload: state, + vectorClock, + schemaVersion: CURRENT_SCHEMA_VERSION, + clientTimestamp: BigInt(1), + receivedAt: BigInt(1), + isPayloadEncrypted: false, + syncImportReason: null, + serverSeq: 77, + }); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/snapshot', + headers: { authorization: `Bearer ${authToken}` }, + payload: { + state, + clientId, + reason: 'recovery', + vectorClock, + schemaVersion: CURRENT_SCHEMA_VERSION, + opId, + isCleanSlate: true, + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toEqual({ accepted: true, serverSeq: 77 }); + expect(mocks.syncService.uploadOps).not.toHaveBeenCalled(); + expect(mocks.syncService.checkStorageQuota).not.toHaveBeenCalled(); + }); + + it('should reject a clean-slate retry whose opId belongs to different content', async () => { + const opId = '018f2f0b-1c2d-7a1b-8c3d-123456789abc'; + const clientId = 'clean-slate-collision-client'; + const vectorClock = { [clientId]: 1 }; + mocks.prisma.operation.findUnique.mockResolvedValueOnce({ + id: opId, + userId: 1, + clientId, + actionType: '[SP_ALL] Load(import) all data', + opType: 'SYNC_IMPORT', + entityType: 'ALL', + entityId: null, + entityIds: [], + payload: { TASK: { existing: { id: 'existing' } } }, + vectorClock, + schemaVersion: CURRENT_SCHEMA_VERSION, + clientTimestamp: BigInt(1), + receivedAt: BigInt(1), + isPayloadEncrypted: false, + syncImportReason: null, + serverSeq: 77, + }); + + const response = await app.inject({ + method: 'POST', + url: '/api/sync/snapshot', + headers: { authorization: `Bearer ${authToken}` }, + payload: { + state: { TASK: { replacement: { id: 'replacement' } } }, + clientId, + reason: 'recovery', + vectorClock, + schemaVersion: CURRENT_SCHEMA_VERSION, + opId, + isCleanSlate: true, + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toEqual({ + accepted: false, + error: 'Operation ID already belongs to a different operation', + }); + expect(mocks.syncService.uploadOps).not.toHaveBeenCalled(); + expect(mocks.syncService.checkStorageQuota).not.toHaveBeenCalled(); + }); + it('should repeat initial snapshot duplicate detection inside the user lock', async () => { const clientId = 'initial-race-client'; mocks.prisma.operation.findFirst diff --git a/packages/super-sync-server/tests/sync.service.spec.ts b/packages/super-sync-server/tests/sync.service.spec.ts index 3dd887c608..67a3704433 100644 --- a/packages/super-sync-server/tests/sync.service.spec.ts +++ b/packages/super-sync-server/tests/sync.service.spec.ts @@ -79,6 +79,23 @@ vi.mock('../src/db', async () => { } } } + if (args.where?.entityType && Array.isArray(args.where?.OR)) { + const targetEntityId = + args.where.OR.find((condition: any) => condition.entityId !== undefined) + ?.entityId ?? + args.where.OR.find((condition: any) => condition.entityIds?.has !== undefined) + ?.entityIds.has; + const ops = Array.from(state.operations.values()) + .filter( + (op: any) => + op.userId === args.where.userId && + op.entityType === args.where.entityType && + (op.entityId === targetEntityId || + op.entityIds?.includes(targetEntityId)), + ) + .sort((a: any, b: any) => b.serverSeq - a.serverSeq); + return applyOperationSelect(ops[0], args.select) || null; + } if (args.where?.entityId && args.where?.entityType) { const ops = Array.from(state.operations.values()) .filter( @@ -359,6 +376,7 @@ vi.mock('../src/db', async () => { entityId: op.entityId, clientId: op.clientId, vectorClock: op.vectorClock, + serverSeq: op.serverSeq, })); } if (sql.includes('jsonb_each_text(vector_clock)')) { @@ -605,6 +623,8 @@ import { DeviceService } from '../src/sync/services/device.service'; import { OperationDownloadService } from '../src/sync/services/operation-download.service'; import { Operation, DEFAULT_SYNC_CONFIG, SYNC_ERROR_CODES } from '../src/sync/sync.types'; import { prisma } from '../src/db'; +import { Logger } from '../src/logger'; +import { CURRENT_SCHEMA_VERSION } from '@sp/shared-schema'; describe('SyncService', () => { const userId = 1; @@ -628,6 +648,19 @@ describe('SyncService', () => { ...overrides, }); + const makeGlobalConfigOp = (overrides: Partial = {}): Operation => + makeOp({ + actionType: '[GLOBAL_CONFIG] Update section', + opType: 'UPD', + entityType: 'GLOBAL_CONFIG', + entityId: 'misc', + payload: { + sectionKey: 'misc', + sectionCfg: { defaultProjectId: 'project-1' }, + }, + ...overrides, + }); + beforeEach(() => { // Reset all test data stores resetTestState(); @@ -676,6 +709,23 @@ describe('SyncService', () => { expect(result).toEqual([validOp]); }); + + it('does not charge a later valid sibling when an invalid op reserved its ID', () => { + const service = new SyncService(); + const invalidFirst = makeOp({ + id: 'reserved-by-invalid-op', + entityType: 'INVALID_ENTITY_TYPE', + }); + const laterLargeSibling = makeOp({ + id: invalidFirst.id, + entityId: 'fresh-task', + payload: { data: 'x'.repeat(10_000) }, + }); + + expect( + service.filterValidOpsForQuota([invalidFirst, laterLargeSibling], clientId), + ).toEqual([]); + }); }); describe('uploadOps', () => { @@ -826,6 +876,114 @@ describe('SyncService', () => { ); }); + it('terminally rejects a later serial same-ID sibling when the first one conflicts', async () => { + const service = new SyncService({ batchUpload: false }); + const otherClientId = 'other-device'; + const existing = makeOp({ + id: 'existing-op', + clientId: otherClientId, + entityId: 'blocked-task', + vectorClock: { [otherClientId]: 1 }, + }); + expect( + (await service.uploadOps(userId, otherClientId, [existing]))[0].accepted, + ).toBe(true); + + const repeatedId = 'repeated-request-id'; + const first = makeOp({ + id: repeatedId, + entityId: 'blocked-task', + payload: { title: 'small' }, + vectorClock: { [clientId]: 1 }, + }); + const laterLargeSibling = makeOp({ + id: repeatedId, + entityId: 'fresh-task', + payload: { data: 'x'.repeat(10_000) }, + vectorClock: { [clientId]: 2 }, + timestamp: first.timestamp + 1, + }); + + const results = await service.uploadOps(userId, clientId, [ + first, + laterLargeSibling, + ]); + + expect(results[0]).toEqual( + expect.objectContaining({ + accepted: false, + errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT, + }), + ); + expect(results[1]).toEqual( + expect.objectContaining({ + accepted: false, + errorCode: SYNC_ERROR_CODES.INVALID_OP_ID, + }), + ); + expect(testState.operations.has(repeatedId)).toBe(false); + }); + + it('redacts malformed operation metadata from audit logs', async () => { + const service = new SyncService({ batchUpload: false }); + const privateText = 'private task title that must not be logged'; + const auditSpy = vi.spyOn(Logger, 'audit').mockImplementation(() => undefined); + const malformed = makeOp({ + id: privateText, + entityType: privateText, + }); + + const result = await service.uploadOps(userId, clientId, [malformed]); + + expect(result[0].accepted).toBe(false); + const rejection = auditSpy.mock.calls + .map(([entry]) => entry) + .find((entry) => entry.event === 'OP_REJECTED'); + expect(rejection).toBeDefined(); + expect(rejection?.opId).toBe('[invalid]'); + expect(rejection?.entityType).toBe('[invalid]'); + expect(rejection?.reason).toBe(SYNC_ERROR_CODES.INVALID_ENTITY_TYPE); + expect(JSON.stringify(rejection)).not.toContain(privateText); + }); + + it.each([ + ['serial', false], + ['batch', true], + ])( + 'terminally rejects a valid %s sibling whose ID was reserved by an invalid op', + async (_label, batchUpload) => { + const service = new SyncService({ batchUpload }); + const invalidFirst = makeOp({ + id: 'invalid-first-shared-id', + entityType: 'INVALID_ENTITY_TYPE', + }); + const laterLargeSibling = makeOp({ + id: invalidFirst.id, + entityId: 'fresh-task', + payload: { data: 'x'.repeat(10_000) }, + }); + + const results = await service.uploadOps(userId, clientId, [ + invalidFirst, + laterLargeSibling, + ]); + + expect(results[0]).toEqual( + expect.objectContaining({ + accepted: false, + errorCode: SYNC_ERROR_CODES.INVALID_ENTITY_TYPE, + }), + ); + expect(results[1]).toEqual( + expect.objectContaining({ + accepted: false, + errorCode: SYNC_ERROR_CODES.INVALID_OP_ID, + }), + ); + expect(testState.operations.has(invalidFirst.id)).toBe(false); + }, + ); + it('should reject intra-batch entity conflicts in order', async () => { const service = new SyncService({ batchUpload: true }); const ops: Operation[] = [ @@ -868,6 +1026,118 @@ describe('SyncService', () => { expect(testState.operations.size).toBe(1); }); + it.each([ + ['serial', false], + ['batch', true], + ])( + 'rejects a v2 tasks write against an already-stored raw v1 misc row in the %s path', + async (_label, batchUpload) => { + const legacyClientId = 'legacy-client'; + testState.userSyncStates.set(userId, { userId, lastSeq: 1 }); + testState.serverSeqCounter = 1; + testState.operations.set('stored-legacy-misc', { + id: 'stored-legacy-misc', + userId, + clientId: legacyClientId, + serverSeq: 1, + actionType: '[GLOBAL_CONFIG] Update section', + opType: 'UPD', + entityType: 'GLOBAL_CONFIG', + entityId: 'misc', + entityIds: [], + payload: { + sectionKey: 'misc', + sectionCfg: { defaultProjectId: 'legacy-project' }, + }, + payloadBytes: BigInt(10), + vectorClock: { [legacyClientId]: 1 }, + schemaVersion: 1, + clientTimestamp: BigInt(Date.now() - 1_000), + receivedAt: BigInt(Date.now() - 1_000), + isPayloadEncrypted: false, + syncImportReason: null, + }); + + const service = new SyncService({ batchUpload }); + const result = await service.uploadOps(userId, clientId, [ + makeGlobalConfigOp({ + id: 'current-tasks-write', + entityId: 'tasks', + payload: { + sectionKey: 'tasks', + sectionCfg: { defaultProjectId: 'current-project' }, + }, + vectorClock: { [clientId]: 1 }, + schemaVersion: CURRENT_SCHEMA_VERSION, + }), + ]); + + expect(result).toEqual([ + expect.objectContaining({ + opId: 'current-tasks-write', + accepted: false, + errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT, + existingClock: { [legacyClientId]: 1 }, + }), + ]); + expect(testState.operations.size).toBe(1); + }, + ); + + it.each([ + ['serial', false], + ['batch', true], + ])( + 'atomically rejects a new mixed v1 misc upload that conflicts with v2 tasks in the %s path', + async (_label, batchUpload) => { + const currentClientId = 'current-client'; + const service = new SyncService({ batchUpload }); + const currentResult = await service.uploadOps(userId, currentClientId, [ + makeGlobalConfigOp({ + id: 'existing-current-tasks', + clientId: currentClientId, + entityId: 'tasks', + payload: { + sectionKey: 'tasks', + sectionCfg: { defaultProjectId: 'current-project' }, + }, + vectorClock: { [currentClientId]: 1 }, + schemaVersion: CURRENT_SCHEMA_VERSION, + }), + ]); + expect(currentResult[0].accepted).toBe(true); + + const sourceId = 'incoming-legacy-mixed'; + const legacyResult = await service.uploadOps(userId, clientId, [ + makeGlobalConfigOp({ + id: sourceId, + payload: { + sectionKey: 'misc', + sectionCfg: { + defaultProjectId: 'legacy-project', + isMinimizeToTray: true, + }, + }, + vectorClock: { [clientId]: 1 }, + schemaVersion: 1, + }), + ]); + + expect(legacyResult).toEqual([ + expect.objectContaining({ + opId: sourceId, + accepted: false, + errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT, + existingClock: { [currentClientId]: 1 }, + }), + ]); + expect(testState.operations.has(`${sourceId}_misc`)).toBe(false); + expect(testState.operations.has(`${sourceId}_tasks`)).toBe(false); + expect(testState.operations.has(sourceId)).toBe(false); + expect(testState.operations.size).toBe(1); + }, + ); + it('should use entityIds when prefetching batch conflicts', async () => { const service = new SyncService({ batchUpload: true }); testState.userSyncStates.set(userId, { userId, lastSeq: 1 });