From 43c10ed73b1360ee82dfac4c50e23fde416c5eab Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Tue, 12 May 2026 16:46:38 +0200 Subject: [PATCH] refactor(sync): polish storage delta tracking and update tests Picked up the remaining items from the multi-agent review: - Hoist Operation / ServerOperation imports in sync.routes.ts so the upload path no longer needs `as unknown as import('./sync.types').*` double casts. The Set + filter pattern for accepted ops becomes an index-zip loop, dropping two iterations + the Set allocation. - Pull APPROX_BYTES_PER_OP out of two inline locals in sync.service.ts into sync.const.ts so the constant is a single source of truth and the trade-off (over-estimate vs observed median op size) is documented in one place. - Log when computeOpsStorageBytes silently skips an op whose payload fails JSON.stringify (BigInt, circular refs). Counter still under-counts but the cause is visible in logs. - Update storage-quota docstrings to reflect that calculateStorageUsage / updateStorageUsage are now legitimately called from the freeStorageForUpload reconcile path (rare cleanup events), not only offline. Tests: - cleanup.spec.ts: assert updateStorageUsage is NOT called after deleteOldSyncedOpsForAllUsers (the per-user SUM loop was the DoS). - storage-quota-cleanup.spec.ts: drop the assertion that deleteOldestRestorePointAndOps issues a pg_column_size query (it does not anymore); add a $executeRaw mock that mutates testUsers so the iterative-cleanup test exercises the real decrement path; tune starting storage to within a few KB of quota so cleanup's approximate decrement can detect progress within a few iterations. - storage-quota.service.spec.ts: add the $executeRaw prisma mock and unit tests for incrementStorageUsage / decrementStorageUsage (positive path, Math.floor of non-integer deltas, no-op on NaN/Infinity/zero/negative, clamped UPDATE for decrement). --- .../sync/services/storage-quota.service.ts | 19 ++-- .../super-sync-server/src/sync/sync.const.ts | 11 +++ .../super-sync-server/src/sync/sync.routes.ts | 28 +++--- .../src/sync/sync.service.ts | 3 +- .../super-sync-server/tests/cleanup.spec.ts | 12 +-- .../tests/storage-quota-cleanup.spec.ts | 87 +++++++++---------- .../tests/storage-quota.service.spec.ts | 63 ++++++++++++++ 7 files changed, 151 insertions(+), 72 deletions(-) diff --git a/packages/super-sync-server/src/sync/services/storage-quota.service.ts b/packages/super-sync-server/src/sync/services/storage-quota.service.ts index dfec5a54bc..b1d2c3aed6 100644 --- a/packages/super-sync-server/src/sync/services/storage-quota.service.ts +++ b/packages/super-sync-server/src/sync/services/storage-quota.service.ts @@ -18,11 +18,14 @@ export class StorageQuotaService { /** * Calculate actual storage usage for a user by summing on-disk payload sizes. * - * OFFLINE / ADMIN USE ONLY. SUM(pg_column_size(payload)) forces PostgreSQL to - * detoast every payload for the user (TOAST table reads), which on active - * users takes minutes and saturates disk I/O. Never call this on the request - * path. Hot-path tracking uses incrementStorageUsage / decrementStorageUsage - * with deltas computed locally on the Node side. + * SLOW PATH — DO NOT CALL PER REQUEST. SUM(pg_column_size(payload)) forces + * PostgreSQL to detoast every payload for the user (TOAST table reads), + * which on active users takes minutes and saturates disk I/O. Reserved for: + * 1. Quota-cache reconciliation, run at most once per quota-cleanup event + * (rare per user) — see SyncService.freeStorageForUpload. + * 2. Offline / admin reconciliation scripts. + * Hot-path tracking uses incrementStorageUsage / decrementStorageUsage with + * deltas computed locally on the Node side. */ async calculateStorageUsage(userId: number): Promise<{ operationsBytes: number; @@ -105,11 +108,7 @@ export class StorageQuotaService { /** * Recompute the cached storage usage from scratch via calculateStorageUsage. - * - * OFFLINE / ADMIN USE ONLY. Same warning as calculateStorageUsage: forces - * full-payload detoasting and was the source of a production disk-I/O DoS - * when called on the request path. Use incrementStorageUsage / - * decrementStorageUsage on hot paths. Keep this for admin backfill scripts. + * Same slow-path warning applies — see calculateStorageUsage. */ async updateStorageUsage(userId: number): Promise { const { totalBytes } = await this.calculateStorageUsage(userId); diff --git a/packages/super-sync-server/src/sync/sync.const.ts b/packages/super-sync-server/src/sync/sync.const.ts index 845aea290e..e28b452d89 100644 --- a/packages/super-sync-server/src/sync/sync.const.ts +++ b/packages/super-sync-server/src/sync/sync.const.ts @@ -2,3 +2,14 @@ export { SUPER_SYNC_CLIENT_ID_REGEX as CLIENT_ID_REGEX, SUPER_SYNC_MAX_CLIENT_ID_LENGTH as MAX_CLIENT_ID_LENGTH, } from '@sp/shared-schema'; + +/** + * Approximate bytes-per-op used when decrementing `users.storage_used_bytes` + * during cleanup-deletes. The exact figure would require detoasting every + * deleted payload via `pg_column_size`, which was the source of the production + * disk-I/O DoS. Picked as a conservative over-estimate vs the observed median + * task-op (~150-300 bytes) so the cleanup loop reliably makes progress; drift + * is reconciled once at the end of `freeStorageForUpload` via a single + * `updateStorageUsage` scan. + */ +export const APPROX_BYTES_PER_OP = 1024; diff --git a/packages/super-sync-server/src/sync/sync.routes.ts b/packages/super-sync-server/src/sync/sync.routes.ts index 4bfe631d80..2e6417f4be 100644 --- a/packages/super-sync-server/src/sync/sync.routes.ts +++ b/packages/super-sync-server/src/sync/sync.routes.ts @@ -12,6 +12,8 @@ import { getWsConnectionService } from './services/websocket-connection.service' import { Logger } from '../logger'; import { prisma } from '../db'; import { + Operation, + ServerOperation, UploadOpsRequest, UploadOpsResponse, DownloadOpsResponse, @@ -63,9 +65,10 @@ const errorMessage = (err: unknown): string => * upload response after ops were persisted. */ const computeOpsStorageBytes = ( - ops: Array<{ payload: unknown; vectorClock: unknown }>, + ops: Array<{ id?: string; payload: unknown; vectorClock: unknown }>, ): number => { let total = 0; + let skipped = 0; for (const op of ops) { try { total += Buffer.byteLength(JSON.stringify(op.payload ?? null), 'utf8'); @@ -73,8 +76,12 @@ const computeOpsStorageBytes = ( } catch { // Skip unserializable op — counter under-counts slightly. Quota is // advisory; offline reconciliation corrects drift. + skipped++; } } + if (skipped > 0) { + Logger.warn(`computeOpsStorageBytes: skipped ${skipped} unserializable op(s)`); + } return total; }; @@ -310,7 +317,7 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise => { // The original response may have contained newOps that the client missed if the // network dropped the response. By using the CURRENT request's lastKnownServerSeq, // we ensure the client gets all ops it hasn't seen yet. - let newOps: import('./sync.types').ServerOperation[] | undefined; + let newOps: ServerOperation[] | undefined; let latestSeq: number; let hasMorePiggyback = false; const PIGGYBACK_LIMIT = 500; @@ -362,7 +369,7 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise => { const results = await syncService.uploadOps( userId, clientId, - ops as unknown as import('./sync.types').Operation[], + ops as unknown as Operation[], isCleanSlate, ); @@ -390,12 +397,13 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise => { // accepted and persisted, so a counter update failure must not 500 the // response (would cause client retry + double-upload). if (accepted > 0) { - const acceptedIds = new Set( - results.filter((r) => r.accepted).map((r) => r.opId), - ); - const acceptedOps = ( - ops as unknown as import('./sync.types').Operation[] - ).filter((op) => acceptedIds.has(op.id)); + // results[i] corresponds to ops[i] (uploadOps preserves order). + // Zip rather than rebuild a Set — avoids two extra passes + alloc. + const typedOps = ops as unknown as Operation[]; + const acceptedOps: Operation[] = []; + for (let i = 0; i < typedOps.length; i++) { + if (results[i]?.accepted) acceptedOps.push(typedOps[i]); + } const deltaBytes = computeOpsStorageBytes(acceptedOps); if (deltaBytes > 0) { try { @@ -409,7 +417,7 @@ export const syncRoutes = async (fastify: FastifyInstance): Promise => { } // Optionally include new ops from other clients (with atomic latestSeq read) - let newOps: import('./sync.types').ServerOperation[] | undefined; + let newOps: ServerOperation[] | undefined; let latestSeq: number; let hasMorePiggyback = false; const PIGGYBACK_LIMIT = 500; diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index 73ee156a3a..2560496eb7 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -12,6 +12,7 @@ import { ConflictType, ConflictResult, } from './sync.types'; +import { APPROX_BYTES_PER_OP } from './sync.const'; import { Logger } from '../logger'; import { CURRENT_SCHEMA_VERSION } from '@sp/shared-schema'; import { Prisma } from '@prisma/client'; @@ -824,7 +825,6 @@ export class SyncService { let totalDeleted = 0; const affectedUserIds: number[] = []; - const APPROX_BYTES_PER_OP = 1024; for (const state of states) { const snapshotAt = Number(state.snapshotAt); @@ -911,7 +911,6 @@ export class SyncService { // freeStorageForUpload's loop can make progress without recomputing per-row // pg_column_size (the original DoS pattern). Reconciled to exact value once // at the end of freeStorageForUpload via a single updateStorageUsage call. - const APPROX_BYTES_PER_OP = 1024; // Delete the operations const result = await prisma.operation.deleteMany({ diff --git a/packages/super-sync-server/tests/cleanup.spec.ts b/packages/super-sync-server/tests/cleanup.spec.ts index 019e802889..6fbc5a85e3 100644 --- a/packages/super-sync-server/tests/cleanup.spec.ts +++ b/packages/super-sync-server/tests/cleanup.spec.ts @@ -95,7 +95,11 @@ describe('Cleanup Jobs', () => { expect(mockSyncService.deleteOldSyncedOpsForAllUsers).toHaveBeenCalledTimes(3); }); - it('should update storage usage for affected users after op cleanup', async () => { + it('should not call full-scan updateStorageUsage after op cleanup', async () => { + // The daily cleanup used to call updateStorageUsage(userId) for every + // affected user, which forced a full-payload TOAST scan and caused the + // production disk-I/O DoS. Counter decrement now happens incrementally + // inside deleteOldSyncedOpsForAllUsers via decrementStorageUsage. mockSyncService.deleteOldSyncedOpsForAllUsers.mockResolvedValueOnce({ totalDeleted: 100, affectedUserIds: [1, 2, 3], @@ -104,11 +108,7 @@ describe('Cleanup Jobs', () => { startCleanupJobs(); await vi.advanceTimersByTimeAsync(10_000); - // Should update storage for each affected user - expect(mockSyncService.updateStorageUsage).toHaveBeenCalledTimes(3); - expect(mockSyncService.updateStorageUsage).toHaveBeenCalledWith(1); - expect(mockSyncService.updateStorageUsage).toHaveBeenCalledWith(2); - expect(mockSyncService.updateStorageUsage).toHaveBeenCalledWith(3); + expect(mockSyncService.updateStorageUsage).not.toHaveBeenCalled(); }); }); diff --git a/packages/super-sync-server/tests/storage-quota-cleanup.spec.ts b/packages/super-sync-server/tests/storage-quota-cleanup.spec.ts index f3d368a721..7a8e3d78d7 100644 --- a/packages/super-sync-server/tests/storage-quota-cleanup.spec.ts +++ b/packages/super-sync-server/tests/storage-quota-cleanup.spec.ts @@ -196,6 +196,8 @@ vi.mock('../src/db', () => { .fn() .mockImplementation(async (_query, userIdArg, deleteUpToSeqArg) => { // Compute total bytes from test operations that match the SQL WHERE params. + // Only used by calculateStorageUsage (offline path) — hot paths no + // longer SUM pg_column_size. let total = BigInt(0); for (const op of testOperations.values()) { if (typeof userIdArg === 'number' && op.userId !== userIdArg) { @@ -211,6 +213,24 @@ vi.mock('../src/db', () => { } return [{ total }]; }), + // decrementStorageUsage uses $executeRaw with a clamped UPDATE. + // Simulate by mutating the matching user's storageUsedBytes in-place. + $executeRaw: vi.fn().mockImplementation(async (...args: unknown[]) => { + const interpolated = args.slice(1); + const delta = interpolated.find((v) => typeof v === 'bigint') as + | bigint + | undefined; + const uid = interpolated.find((v) => typeof v === 'number') as + | number + | undefined; + if (delta === undefined || uid === undefined) return 0; + const user = testUsers.get(uid); + if (!user) return 0; + const current = BigInt(user.storageUsedBytes ?? 0); + const next = current > delta ? current - delta : BigInt(0); + testUsers.set(uid, { ...user, storageUsedBytes: next }); + return 1; + }), }, initDb: vi.fn(), getDb: vi.fn(), @@ -347,7 +367,11 @@ describe('Storage Quota Cleanup', () => { expect(testOperations.size).toBe(4); // ops 4, 5, 6, 7 remain }); - it('should avoid materializing deleted JSON payloads as text', async () => { + it('should not run any per-row pg_column_size query during cleanup-delete', async () => { + // The cleanup path used to SUM(pg_column_size(payload)) over the range + // being deleted, which forced PostgreSQL to detoast every payload and + // caused the production disk-I/O DoS. Counter is now decremented via + // an approximate count*const decrement; no $queryRaw should run. const { initSyncService, getSyncService } = await import('../src/sync/sync.service'); const { prisma } = await import('../src/db'); @@ -358,21 +382,17 @@ describe('Storage Quota Cleanup', () => { createOp(clientId, userId); createRestorePoint(clientId, userId); + vi.mocked(prisma.$queryRaw).mockClear(); + await service.deleteOldestRestorePointAndOps(userId); - const [queryParts] = vi.mocked(prisma.$queryRaw).mock.calls[0] as unknown as [ - TemplateStringsArray, - number, - number, - ]; - const query = Array.from(queryParts).join(''); - - expect(query).toContain('pg_column_size(payload)'); - expect(query).toContain('pg_column_size(vector_clock)'); - expect(query).toMatch(/WHERE user_id =/); - expect(query).toMatch(/server_seq <=/); - expect(query).not.toContain('payload::text'); - expect(query).not.toContain('vector_clock::text'); + const offendingCalls = vi + .mocked(prisma.$queryRaw) + .mock.calls.filter((call) => { + const queryParts = call[0] as unknown as TemplateStringsArray; + return Array.from(queryParts).join('').includes('pg_column_size'); + }); + expect(offendingCalls).toHaveLength(0); }); it('should keep single restore point but delete ops before it', async () => { @@ -571,34 +591,23 @@ describe('Storage Quota Cleanup', () => { expect(result.deletedRestorePoints).toBe(0); }); - it('should delete multiple restore points iteratively until quota is satisfied', async () => { + it('should delete restore points iteratively until quota is satisfied', async () => { + // Counter is now decremented incrementally by deleteOldestRestorePointAndOps + // via $executeRaw (mocked above to mutate testUsers), so the loop can + // detect progress without per-row pg_column_size scans. const { initSyncService, getSyncService } = await import('../src/sync/sync.service'); initSyncService(); const service = getSyncService(); - // Track storage updates to simulate freeing space after each deletion - let currentStorage = 150 * 1024 * 1024; // Start at 150MB - const storagePerRestorePoint = 30 * 1024 * 1024; // 30MB per restore point - - // Mock user lookup to return decreasing storage after each update - const originalGet = testUsers.get.bind(testUsers); - testUsers.get = (key: number) => { - const user = originalGet(key); - if (user) { - return { - ...user, - storageUsedBytes: BigInt(currentStorage), - }; - } - return user; - }; - + // Start a few KB above quota so cleanup's approximate decrement + // (count * 1024) can bring us under within a few iterations. + const quota = 100 * 1024 * 1024; testUsers.set(userId, { id: userId, email: 'test@test.com', - storageUsedBytes: BigInt(currentStorage), - storageQuotaBytes: BigInt(100 * 1024 * 1024), // 100MB quota + storageUsedBytes: BigInt(quota + 3000), + storageQuotaBytes: BigInt(quota), }); // Create: RESTORE1, ops, RESTORE2, ops, RESTORE3, ops, RESTORE4 (current) @@ -614,18 +623,8 @@ describe('Storage Quota Cleanup', () => { expect(testOperations.size).toBe(9); - // Mock updateStorageUsage to decrease storage - const originalUpdateStorageUsage = service.updateStorageUsage.bind(service); - service.updateStorageUsage = async (uid: number) => { - currentStorage -= storagePerRestorePoint; - if (currentStorage < 0) currentStorage = 0; - await originalUpdateStorageUsage(uid); - }; - const result = await service.freeStorageForUpload(userId, 1000); - // Should succeed after deleting enough restore points - // Need to delete 2 restore points to go from 150MB to 90MB (under 100MB quota) expect(result.success).toBe(true); expect(result.deletedRestorePoints).toBeGreaterThanOrEqual(1); expect(result.deletedOps).toBeGreaterThan(0); diff --git a/packages/super-sync-server/tests/storage-quota.service.spec.ts b/packages/super-sync-server/tests/storage-quota.service.spec.ts index 965835cdcb..2b58eb16ae 100644 --- a/packages/super-sync-server/tests/storage-quota.service.spec.ts +++ b/packages/super-sync-server/tests/storage-quota.service.spec.ts @@ -5,6 +5,7 @@ import { StorageQuotaService } from '../src/sync/services/storage-quota.service' vi.mock('../src/db', () => ({ prisma: { $queryRaw: vi.fn(), + $executeRaw: vi.fn(), user: { findUnique: vi.fn(), update: vi.fn(), @@ -163,6 +164,68 @@ describe('StorageQuotaService', () => { }); }); + describe('incrementStorageUsage', () => { + it('should atomically increment storage_used_bytes', async () => { + vi.mocked(prisma.user.update).mockResolvedValue({} as any); + + await service.incrementStorageUsage(1, 4096); + + expect(prisma.user.update).toHaveBeenCalledWith({ + where: { id: 1 }, + data: { storageUsedBytes: { increment: BigInt(4096) } }, + }); + }); + + it('should floor non-integer deltas', async () => { + vi.mocked(prisma.user.update).mockResolvedValue({} as any); + + await service.incrementStorageUsage(1, 4096.9); + + expect(prisma.user.update).toHaveBeenCalledWith({ + where: { id: 1 }, + data: { storageUsedBytes: { increment: BigInt(4096) } }, + }); + }); + + it.each([0, -1, NaN, Infinity, -Infinity])( + 'should be a no-op for non-positive or non-finite delta %p', + async (delta) => { + vi.mocked(prisma.user.update).mockResolvedValue({} as any); + + await service.incrementStorageUsage(1, delta); + + expect(prisma.user.update).not.toHaveBeenCalled(); + }, + ); + }); + + describe('decrementStorageUsage', () => { + it('should run a clamped UPDATE via $executeRaw', async () => { + vi.mocked(prisma.$executeRaw).mockResolvedValue(1 as any); + + await service.decrementStorageUsage(1, 2048); + + expect(prisma.$executeRaw).toHaveBeenCalledTimes(1); + const callArgs = vi.mocked(prisma.$executeRaw).mock.calls[0]; + // First arg is the tagged template TemplateStringsArray; subsequent args + // are the interpolated values (delta BigInt + userId). + const interpolatedValues = callArgs.slice(1); + expect(interpolatedValues).toContain(BigInt(2048)); + expect(interpolatedValues).toContain(1); + }); + + it.each([0, -1, NaN, Infinity])( + 'should be a no-op for non-positive or non-finite delta %p', + async (delta) => { + vi.mocked(prisma.$executeRaw).mockResolvedValue(0 as any); + + await service.decrementStorageUsage(1, delta); + + expect(prisma.$executeRaw).not.toHaveBeenCalled(); + }, + ); + }); + describe('getStorageInfo', () => { it('should return storage info for user', async () => { vi.mocked(prisma.user.findUnique).mockResolvedValue({