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

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<void> {
     const { totalBytes } = await this.calculateStorageUsage(userId);

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;

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<void> => {
             // 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<void> => {
         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<void> => {
         // 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<void> => {
         }
 
         // 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;

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

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

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

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