From 47ce4b304c4b3a456e5d41f37deb8b1ddcc54929 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Wed, 13 May 2026 11:31:59 +0200 Subject: [PATCH] perf(super-sync): optimize status and conflict checks --- packages/shared-schema/src/index.ts | 1 + .../src/supersync-http-contract.ts | 6 +- .../tests/supersync-http-contract.spec.ts | 18 +++ .../src/sync/services/snapshot.service.ts | 14 ++ .../super-sync-server/src/sync/sync.routes.ts | 16 +- .../src/sync/sync.service.ts | 145 +++++++++++++----- .../tests/conflict-detection.spec.ts | 76 +++++++++ .../tests/snapshot.service.spec.ts | 26 ++++ .../tests/sync.service.test-state.ts | 4 + 9 files changed, 261 insertions(+), 45 deletions(-) diff --git a/packages/shared-schema/src/index.ts b/packages/shared-schema/src/index.ts index 5938621d65..abc0ddeee1 100644 --- a/packages/shared-schema/src/index.ts +++ b/packages/shared-schema/src/index.ts @@ -45,6 +45,7 @@ export { SUPER_SYNC_CLIENT_ID_REGEX, SUPER_SYNC_MAX_CLIENT_ID_LENGTH, SUPER_SYNC_MAX_OPS_PER_UPLOAD, + SUPER_SYNC_MAX_ENTITY_IDS_PER_OP, SUPER_SYNC_OP_TYPES, SUPER_SYNC_IMPORT_REASONS, SUPER_SYNC_SNAPSHOT_REASONS, diff --git a/packages/shared-schema/src/supersync-http-contract.ts b/packages/shared-schema/src/supersync-http-contract.ts index 0068298d2e..3c2df214cf 100644 --- a/packages/shared-schema/src/supersync-http-contract.ts +++ b/packages/shared-schema/src/supersync-http-contract.ts @@ -3,6 +3,7 @@ import { z } from 'zod'; export const SUPER_SYNC_CLIENT_ID_REGEX = /^[a-zA-Z0-9_-]+$/; export const SUPER_SYNC_MAX_CLIENT_ID_LENGTH = 255; export const SUPER_SYNC_MAX_OPS_PER_UPLOAD = 100; +export const SUPER_SYNC_MAX_ENTITY_IDS_PER_OP = 1000; export const SUPER_SYNC_OP_TYPES = [ 'CRT', @@ -50,7 +51,10 @@ export const SuperSyncOperationSchema = z.object({ opType: z.enum(SUPER_SYNC_OP_TYPES), entityType: z.string().min(1).max(255), entityId: z.string().max(255).optional(), - entityIds: z.array(z.string().max(255)).optional(), + entityIds: z + .array(z.string().max(255)) + .max(SUPER_SYNC_MAX_ENTITY_IDS_PER_OP) + .optional(), payload: z.unknown(), vectorClock: SuperSyncVectorClockSchema, timestamp: z.number(), diff --git a/packages/shared-schema/tests/supersync-http-contract.spec.ts b/packages/shared-schema/tests/supersync-http-contract.spec.ts index e314213240..ab622b21e8 100644 --- a/packages/shared-schema/tests/supersync-http-contract.spec.ts +++ b/packages/shared-schema/tests/supersync-http-contract.spec.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from 'vitest'; import { + SUPER_SYNC_MAX_ENTITY_IDS_PER_OP, SUPER_SYNC_MAX_OPS_PER_UPLOAD, SuperSyncDownloadOpsQuerySchema, SuperSyncDownloadOpsResponseSchema, @@ -45,6 +46,23 @@ describe('SuperSync HTTP contract schemas', () => { expect('extraOpField' in parsed.ops[0]).toBe(false); }); + it('caps entityIds per operation', () => { + expect(() => + SuperSyncUploadOpsRequestSchema.parse({ + ops: [ + { + ...createValidOperation(), + entityIds: Array.from( + { length: SUPER_SYNC_MAX_ENTITY_IDS_PER_OP + 1 }, + (_, i) => `task-${i}`, + ), + }, + ], + clientId: 'client_1', + }), + ).toThrow(); + }); + it('rejects invalid client IDs in upload requests', () => { expect(() => SuperSyncUploadOpsRequestSchema.parse({ diff --git a/packages/super-sync-server/src/sync/services/snapshot.service.ts b/packages/super-sync-server/src/sync/services/snapshot.service.ts index cfa470f95f..3609993d84 100644 --- a/packages/super-sync-server/src/sync/services/snapshot.service.ts +++ b/packages/super-sync-server/src/sync/services/snapshot.service.ts @@ -191,6 +191,20 @@ export class SnapshotService { return Number(result[0]?.bytes ?? 0); } + /** + * Return cached snapshot timestamp without loading the snapshot blob. + * Status polling only needs snapshot age; reading, gunzipping, and parsing + * snapshotData there can turn a cheap endpoint into a large CPU/memory hit. + */ + async getCachedSnapshotGeneratedAt(userId: number): Promise { + const row = await prisma.userSyncState.findUnique({ + where: { userId }, + select: { snapshotAt: true }, + }); + + return row?.snapshotAt != null ? Number(row.snapshotAt) : null; + } + /** * Get cached snapshot for a user. */ diff --git a/packages/super-sync-server/src/sync/sync.routes.ts b/packages/super-sync-server/src/sync/sync.routes.ts index 44416c7dcb..19cbcfa1c9 100644 --- a/packages/super-sync-server/src/sync/sync.routes.ts +++ b/packages/super-sync-server/src/sync/sync.routes.ts @@ -1154,13 +1154,15 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise => { const userId = getAuthUser(req).userId; const syncService = getSyncService(); - const latestSeq = await syncService.getLatestSeq(userId); - const devicesOnline = await syncService.getOnlineDeviceCount(userId); - - const cached = await syncService.getCachedSnapshot(userId); - const snapshotAge = cached ? Date.now() - cached.generatedAt : undefined; - - const storageInfo = await syncService.getStorageInfo(userId); + const [latestSeq, devicesOnline, snapshotGeneratedAt, storageInfo] = + await Promise.all([ + syncService.getLatestSeq(userId), + syncService.getOnlineDeviceCount(userId), + syncService.getCachedSnapshotGeneratedAt(userId), + syncService.getStorageInfo(userId), + ]); + const snapshotAge = + snapshotGeneratedAt !== null ? Date.now() - snapshotGeneratedAt : undefined; Logger.debug( `[user:${userId}] Status: seq=${latestSeq}, devices=${devicesOnline}`, diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index c4b9897ac2..5ac3e0ae0b 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -42,6 +42,16 @@ interface DuplicateOperationCandidate { syncImportReason: string | null; } +interface LatestEntityOperationRow { + entityId: string; + clientId: string; + vectorClock: unknown; +} + +// Conservative enough to avoid planner-heavy BitmapOr + Sort plans on large +// histories while still replacing up to 100 per-entity round trips with one query. +const CONFLICT_DETECTION_ENTITY_BATCH_SIZE = 100; + /** * Main sync orchestration service. * @@ -98,60 +108,82 @@ export class SyncService { // Build list of entity IDs to check for conflicts. // Operations may have either entityId (singular) or entityIds (batch operations). - const entityIdsToCheck: string[] = op.entityIds?.length + const rawEntityIdsToCheck = op.entityIds?.length ? op.entityIds : op.entityId ? [op.entityId] : []; // Skip if no entity IDs (can't have entity-level conflicts) - if (entityIdsToCheck.length === 0) { + if (rawEntityIdsToCheck.length === 0) { return { hasConflict: false }; } - // Check each entity for conflicts - for (const entityId of entityIdsToCheck) { - const conflictResult = await this.detectConflictForEntity(userId, op, entityId, tx); - if (conflictResult.hasConflict) { - return conflictResult; + if (rawEntityIdsToCheck.length === 1) { + return this.detectConflictForEntity(userId, op, rawEntityIdsToCheck[0], tx); + } + + const entityIdsToCheck = Array.from(new Set(rawEntityIdsToCheck)); + return this.detectConflictForEntities(userId, op, entityIdsToCheck, tx); + } + + private async detectConflictForEntities( + userId: number, + op: Operation, + entityIdsToCheck: string[], + tx: Prisma.TransactionClient, + ): Promise { + for ( + let start = 0; + start < entityIdsToCheck.length; + start += CONFLICT_DETECTION_ENTITY_BATCH_SIZE + ) { + const batchEntityIds = entityIdsToCheck.slice( + start, + start + CONFLICT_DETECTION_ENTITY_BATCH_SIZE, + ); + const latestOps = await tx.$queryRaw` + SELECT DISTINCT ON (entity_id) + entity_id AS "entityId", + client_id AS "clientId", + vector_clock AS "vectorClock" + FROM operations + WHERE user_id = ${userId} + AND entity_type = ${op.entityType} + AND entity_id IN (${Prisma.join(batchEntityIds)}) + ORDER BY entity_id, server_seq DESC + `; + + const latestOpByEntityId = new Map(); + for (const latestOp of latestOps) { + latestOpByEntityId.set(latestOp.entityId, latestOp); + } + + for (const entityId of batchEntityIds) { + const existingOp = latestOpByEntityId.get(entityId); + if (!existingOp) continue; + + const conflictResult = this.resolveConflictForExistingOp( + op, + entityId, + existingOp, + ); + if (conflictResult.hasConflict) { + return conflictResult; + } } } return { hasConflict: false }; } - /** - * Checks for conflicts on a single entity. - * Extracted from detectConflict to support multi-entity operations. - */ - private async detectConflictForEntity( - userId: number, + private resolveConflictForExistingOp( op: Operation, entityId: string, - tx: Prisma.TransactionClient, - ): Promise { - // Get the latest operation for this entity - const existingOp = await tx.operation.findFirst({ - where: { - userId, - entityType: op.entityType, - entityId, - }, - select: { - clientId: true, - vectorClock: true, - }, - orderBy: { - serverSeq: 'desc', - }, - }); - - // No existing operation = no conflict - if (!existingOp) { - return { hasConflict: false }; - } - - // Parse the existing operation's vector clock (Prisma returns Json, cast to VectorClock) + existingOp: { clientId: string; vectorClock: unknown }, + ): ConflictResult { + // Stored JSON/vector_clock values arrive as unknown from both Prisma model + // reads and raw SQL rows; cast only at the vector-clock comparison boundary. const existingClock = existingOp.vectorClock as unknown as VectorClock; // Compare vector clocks @@ -208,6 +240,41 @@ export class SyncService { }; } + /** + * Checks conflicts for the common single-entity upload path using Prisma's + * typed model API. Multi-entity operations use the batched raw-SQL path above + * to avoid one round trip per entity. + */ + private async detectConflictForEntity( + userId: number, + op: Operation, + entityId: string, + tx: Prisma.TransactionClient, + ): Promise { + // Get the latest operation for this entity + const existingOp = await tx.operation.findFirst({ + where: { + userId, + entityType: op.entityType, + entityId, + }, + select: { + clientId: true, + vectorClock: true, + }, + orderBy: { + serverSeq: 'desc', + }, + }); + + // No existing operation = no conflict + if (!existingOp) { + return { hasConflict: false }; + } + + return this.resolveConflictForExistingOp(op, entityId, existingOp); + } + private isSameDuplicateOperation( existingOp: DuplicateOperationCandidate, userId: number, @@ -855,6 +922,10 @@ export class SyncService { return this.snapshotService.getCachedSnapshotBytes(userId); } + async getCachedSnapshotGeneratedAt(userId: number): Promise { + return this.snapshotService.getCachedSnapshotGeneratedAt(userId); + } + async cacheSnapshot( userId: number, state: unknown, diff --git a/packages/super-sync-server/tests/conflict-detection.spec.ts b/packages/super-sync-server/tests/conflict-detection.spec.ts index 889ee614d0..ea3f604e1c 100644 --- a/packages/super-sync-server/tests/conflict-detection.spec.ts +++ b/packages/super-sync-server/tests/conflict-detection.spec.ts @@ -64,6 +64,7 @@ vi.mock('../src/db', async () => { ); } if (args.where?.entityId && args.where?.entityType) { + state.entityConflictFindFirstCount++; const ops = Array.from(state.operations.values()) .filter( (op: any) => @@ -162,6 +163,48 @@ vi.mock('../src/db', async () => { }, // Upload transaction writes the storage counter atomically via $executeRaw. $executeRaw: vi.fn().mockResolvedValue(0), + $queryRaw: vi + .fn() + .mockImplementation( + async ( + _strings: any, + userId: number, + entityType: string, + entityIdsSql: Prisma.Sql, + ) => { + state.batchConflictQueryCount++; + if (!Array.isArray(entityIdsSql.values)) { + throw new Error( + 'Expected batched conflict query entity IDs to be passed via Prisma.join(...)', + ); + } + const batchEntityIds = new Set( + entityIdsSql.values.filter( + (entityId): entityId is string => typeof entityId === 'string', + ), + ); + const latestByEntityId = new Map(); + const ops = Array.from(state.operations.values()) + .filter( + (op: any) => + op.userId === userId && + op.entityType === entityType && + batchEntityIds.has(op.entityId), + ) + .sort((a: any, b: any) => b.serverSeq - a.serverSeq); + + for (const op of ops) { + if (!op.entityId || latestByEntityId.has(op.entityId)) continue; + latestByEntityId.set(op.entityId, op); + } + + return Array.from(latestByEntityId.values()).map((op: any) => ({ + entityId: op.entityId, + clientId: op.clientId, + vectorClock: op.vectorClock, + })); + }, + ), }); return { @@ -877,5 +920,38 @@ describe('Conflict Detection', () => { const result = await service.uploadOps(userId, clientA, [op]); expect(result[0].accepted).toBe(true); }); + + it('should batch latest-op lookups for multi-entity conflict detection', async () => { + const service = getSyncService(); + + const existingOp = createOp({ + entityId: 'task-2', + clientId: clientB, + vectorClock: { [clientB]: 1 }, + }); + const existingResult = await service.uploadOps(userId, clientB, [existingOp]); + expect(existingResult[0].accepted).toBe(true); + const beforeMultiEntityFindFirstCount = testState.entityConflictFindFirstCount; + + const multiEntityOp = createOp({ + entityId: 'task-1', + entityIds: ['task-1', 'task-2', 'task-3'], + clientId: clientA, + vectorClock: { [clientA]: 1 }, + }); + + const result = await service.uploadOps(userId, clientA, [multiEntityOp]); + + expect(testState.batchConflictQueryCount).toBeGreaterThan(0); + expect(testState.entityConflictFindFirstCount).toBe( + beforeMultiEntityFindFirstCount, + ); + expect(result[0]).toMatchObject({ + accepted: false, + errorCode: SYNC_ERROR_CODES.CONFLICT_CONCURRENT, + existingClock: { [clientB]: 1 }, + }); + expect(result[0].error).toContain('TASK:task-2'); + }); }); }); diff --git a/packages/super-sync-server/tests/snapshot.service.spec.ts b/packages/super-sync-server/tests/snapshot.service.spec.ts index 9a54b8b6b8..b14d2a3ec6 100644 --- a/packages/super-sync-server/tests/snapshot.service.spec.ts +++ b/packages/super-sync-server/tests/snapshot.service.spec.ts @@ -10,6 +10,7 @@ vi.mock('../src/db', () => ({ prisma: { userSyncState: { findUnique: vi.fn(), + findFirst: vi.fn(), update: vi.fn(), updateMany: vi.fn(), create: vi.fn(), @@ -119,6 +120,31 @@ describe('SnapshotService', () => { }); }); + describe('getCachedSnapshotGeneratedAt', () => { + it('should return null when no cached snapshot metadata exists', async () => { + vi.mocked(prisma.userSyncState.findUnique).mockResolvedValue(null); + + const result = await service.getCachedSnapshotGeneratedAt(1); + + expect(result).toBeNull(); + }); + + it('should read only snapshot metadata', async () => { + const snapshotAt = BigInt(1700000000000); + vi.mocked(prisma.userSyncState.findUnique).mockResolvedValue({ + snapshotAt, + } as any); + + const result = await service.getCachedSnapshotGeneratedAt(1); + + expect(result).toBe(Number(snapshotAt)); + expect(prisma.userSyncState.findUnique).toHaveBeenCalledWith({ + where: { userId: 1 }, + select: { snapshotAt: true }, + }); + }); + }); + describe('cacheSnapshot', () => { it('should compress and conditionally update an existing row', async () => { vi.useFakeTimers(); diff --git a/packages/super-sync-server/tests/sync.service.test-state.ts b/packages/super-sync-server/tests/sync.service.test-state.ts index 30fbc21834..e2a1a719be 100644 --- a/packages/super-sync-server/tests/sync.service.test-state.ts +++ b/packages/super-sync-server/tests/sync.service.test-state.ts @@ -10,6 +10,8 @@ export const testState = { userSyncStates: new Map(), users: new Map(), serverSeqCounter: 0, + batchConflictQueryCount: 0, + entityConflictFindFirstCount: 0, }; export function resetTestState(): void { @@ -18,6 +20,8 @@ export function resetTestState(): void { testState.userSyncStates = new Map(); testState.users = new Map(); testState.serverSeqCounter = 0; + testState.batchConflictQueryCount = 0; + testState.entityConflictFindFirstCount = 0; } export function applyOperationSelect(op: any, select?: Record): any {