From 73dade437555465f171a321dbce0d76df6aa4e26 Mon Sep 17 00:00:00 2001 From: johannesjo Date: Fri, 15 May 2026 01:29:28 +0200 Subject: [PATCH] fix(sync): guard payload bytes backfill rollout --- .../2026-05-14-super-sync-server-perf.md | 56 +++++---- packages/super-sync-server/src/server.ts | 24 +++- .../sync/services/storage-quota.service.ts | 106 ++++++++++++++--- .../src/sync/sync.service.ts | 5 + .../tests/storage-quota-cleanup.spec.ts | 10 +- .../tests/storage-quota.service.spec.ts | 112 ++++++++++++++++-- 6 files changed, 262 insertions(+), 51 deletions(-) diff --git a/docs/plans/2026-05-14-super-sync-server-perf.md b/docs/plans/2026-05-14-super-sync-server-perf.md index b1b1787f81..1c0dcaa8fb 100644 --- a/docs/plans/2026-05-14-super-sync-server-perf.md +++ b/docs/plans/2026-05-14-super-sync-server-perf.md @@ -22,7 +22,7 @@ Scope drawn from an audit of `packages/super-sync-server/` covering: upload proc ``` - **Why:** `snapshot.service.ts:1083` and `:1114` `count(*) WHERE is_payload_encrypted=true` over a seq range. Today this scans the range and filters. With the partial index, the common case (no encrypted ops for the user) becomes an empty-index probe. - **Leave alone:** existing `operations_user_id_full_state_server_seq_idx` already covers the `op_type IN (...)` filter for the `findFirst` at `snapshot.service.ts:1103`. (That `findFirst` also filters `isPayloadEncrypted: false`, which isn't in the partial-index predicate — currently a cheap index scan + flag recheck rather than a single probe. Not worth a second partial index.) -- **Verify:** `EXPLAIN ANALYZE` against a production-like distribution: 1M ops for a user with 0-100 encrypted rows, then `ANALYZE operations` after the migration and record the expected plan. Also re-check the latest full-state `findFirst` path that filters `isPayloadEncrypted: false`; the existing full-state partial index does not include that predicate, so many encrypted full-state rows can still force scan-time rechecks. +- **Verify (post-deploy on staging, NOT a CI merge gate):** `EXPLAIN ANALYZE` against a production-like distribution requires populated DB state. Run on staging after the migration applies, with 1M ops for a user holding 0-100 encrypted rows; run `ANALYZE operations` after the migration and record the expected plan. Also re-check the latest full-state `findFirst` path that filters `isPayloadEncrypted: false`; the existing full-state partial index does not include that predicate, so many encrypted full-state rows can still force scan-time rechecks. ### 0b. Snapshot replay size-check cadence (Finding #2) @@ -121,7 +121,7 @@ The plan-as-originally-drafted underspecified four real cases. The revised desig Drop the per-op re-check at `sync.service.ts:786-794`. The safety it covered is delivered by the **shared `user_sync_state.lastSeq` row-write**, which forces concurrent batches to serialize: the second writer blocks on the row lock, then fails with `40001` (serialization failure) on commit. RR isolation alone does NOT provide this — PostgreSQL RR does not run full serializable snapshot isolation. The row-lock pattern is what makes the new design safe. -**This rationale belongs in `ARCHITECTURE-DECISIONS.md`** so that anyone proposing to remove the `lastSeq` increment from the hot path (e.g. sharded sequence assignment, distributed counters) re-introduces the race they were avoiding. Title the entry "Batch uploads under RepeatableRead — safety derives from `user_sync_state.lastSeq` row-lock, not from RR snapshot isolation." +**Already recorded as `ARCHITECTURE-DECISIONS.md` Decision #4** ("Batch Uploads Under RepeatableRead", line 121). Anyone proposing to remove the `lastSeq` increment from the hot path (e.g. sharded sequence assignment, distributed counters) must re-read that decision before doing so. ### 1c. Tests @@ -134,14 +134,17 @@ Drop the per-op re-check at `sync.service.ts:786-794`. The safety it covered is - **Full-state op in batch:** `_aggregatePriorVectorClock` runs exactly once at the end and sees only rows with `server_seq < fullState.serverSeq`. - **Partial-acceptance batch:** 5 dups + 15 accepted → counters correct, audit log has 20 entries. - **Concurrency:** two parallel batches on same user — outer retry on `P2034` / `40001` handles the loser. (This is unchanged in spirit but the failure mode shifts from "per-op re-check" to "shared row lock" — verify it still works.) + - **Sequence-gap invariant:** mixed-batch `[accept, reject, accept, reject, accept]` → persisted rows have contiguous `serverSeq = N, N+1, N+2`, `lastSeq` advances by exactly 3, no gaps anywhere. + - **No-double-terminal invariant:** every accepted op produces exactly zero rejection audits; every rejected op produces exactly zero persisted rows. Run across the full audit-event taxonomy (`OP_REJECTED`, `DUPLICATE_OPERATION`, `INVALID_OP_ID`, `CONFLICT_*`). + - **TIMESTAMP_CLAMPED additive case:** an op with `timestamp > now + maxClockDriftMs` produces one persisted row with the clamped timestamp **plus** one additional `TIMESTAMP_CLAMPED` audit event. The clamp is not a rejection — both outcomes coexist for the same op. - **E2E:** `e2e/tests/sync/` (not `e2e/sync/`) — add one batch-of-50 upload test and assert latency drop vs. baseline. - **Bench:** docker-compose Postgres, time 25-op and 100-op upload before/after. **Also measure concurrent-batch latency:** the shared row-lock means two simultaneous batches serialize hard — the per-batch latency under contention may be similar to today's per-op design. Total throughput should still win because each batch holds the lock for far less wall time. ### 1d. Risk and rollout - Highest-blast-radius change in the plan. Land behind config flag `SUPERSYNC_BATCH_UPLOAD`, default `false` for one release, `true` the next. -- **Wire the flag:** `src/config.ts` exposes a `ServerConfig` parsed from env in `loadConfigFromEnv`. Add `batchUpload: boolean` parsed from `process.env.SUPERSYNC_BATCH_UPLOAD === 'true'`, gated by `SUPERSYNC_PAYLOAD_BYTES_BACKFILL_COMPLETE=true`, and thread through `SyncService` constructor. -- **Route cap:** enforce an explicit max ops per batch before schema validation/service calls so a decompressed body with hundreds of thousands of tiny ops cannot drive huge `IN (...)` prefetches. +- **Wire the flag** (shipped in `src/config.ts:172-179`): `batchUpload = (SUPERSYNC_BATCH_UPLOAD === 'true') && (SUPERSYNC_PAYLOAD_BYTES_BACKFILL_COMPLETE === 'true')`. The first condition without the second throws at startup with a message pointing operators at `npm run migrate-payload-bytes`. The DB-side complement is the startup self-check (see Cross-cutting): if `batchUpload === true` but `operations` still contains rows with `payload_bytes = 0`, the server refuses to boot. This closes the trust hole if an operator flips the env flag too early. +- **Route cap** (shipped in `sync.routes.ts:85, 601-604`): `MAX_OPS_PER_BATCH = SUPER_SYNC_MAX_OPS_PER_UPLOAD = 100` (from `packages/shared-schema/src/supersync-http-contract.ts:5`). Enforced before Zod parsing, returns HTTP 413 with `errorCode: 'PAYLOAD_TOO_LARGE'`. Same value also enforced inside the Zod schema as `.max(SUPER_SYNC_MAX_OPS_PER_UPLOAD)` so the OpenAPI contract stays in sync. - **Invariant:** every op in the batch produces exactly one terminal-status audit (rejection) OR exactly one persisted row, plus optionally one additive `TIMESTAMP_CLAMPED` audit — never both terminal outcomes, never neither, never gapped sequence numbers. --- @@ -177,17 +180,27 @@ There is no clean SQL equivalent of `Buffer.byteLength(JSON.stringify(payload))` ### 2c. Write path -- In the new bulk-`createMany` path (Phase 1), compute `payload_bytes` per op using `computeOpStorageBytes` so the on-row value matches the increment-counter value the hot path is already adding. This is the consistency `calculateStorageUsage` needs. +- **All insert sites** populate `payload_bytes` per op using `computeOpStorageBytes` so the on-row value matches the increment-counter value the hot path is already adding. Both code paths are wired today: batch path at `sync.service.ts:1115`, legacy per-op path at `:1393`. This is the consistency `calculateStorageUsage` needs and the reason an old per-op insert deployed under `SUPERSYNC_BATCH_UPLOAD=false` does not seed drift while batch is rolled out. ### 2d. Read path -- Replace `storage-quota.service.ts:109-112`: +- Replace `storage-quota.service.ts:109-112` with the `CASE WHEN` form already shipped at `:97-120`: ```sql - SELECT COALESCE(SUM(CASE WHEN payload_bytes > 0 THEN payload_bytes ELSE ... END), 0) AS total + SELECT COALESCE( + SUM( + CASE + WHEN payload_bytes > 0 THEN payload_bytes + ELSE octet_length(payload::text)::bigint + + octet_length(vector_clock::text)::bigint + END + ), + 0 + ) AS total FROM operations WHERE user_id = $1 ``` -- Drop `pg_column_size` entirely. No detoasting, no I/O DoS. +- The `ELSE` branch (option (b) in the design analysis) is the only candidate on the same UTF-8 scale as `computeOpStorageBytes`. Detoasting cost is bounded — only un-backfilled rows hit it, and the set drains monotonically to zero. +- Drop `pg_column_size` entirely. No detoasting on backfilled rows, no I/O DoS. - Snapshot side is handled in 0c. ### 2e. Tests @@ -202,10 +215,11 @@ There is no clean SQL equivalent of `Buffer.byteLength(JSON.stringify(payload))` Pick after profiling Phase 0d in production: -### 3a. Streaming serialize + gzip (preferred if memory still hot) +### 3a. Streaming serialize + gzip (conditional, NOT a default win) - Replace `prepareSnapshotCache` (`snapshot.service.ts:175-184`) with a streaming pipeline: a streaming JSON stringifier feeding `zlib.createGzip()`, collecting chunks into a final `Buffer.concat(...)`. -- Net peak memory: saves the intermediate serialized string buffer (~100MB on large states), but the parsed JS object remains because it already exists. Expect roughly 30-40% peak reduction, not 50%+; also expect slower wall-clock than native `JSON.stringify` because JS streaming stringifiers are typically 3-5x slower on large JSON. +- **Pursue only if Phase 0d profiling shows OOM near `MAX_SNAPSHOT_DECOMPRESSED_BYTES` AND event-loop blocking — not memory pressure alone.** The 3-5× wall-clock regression vs native `JSON.stringify` makes this a memory-vs-latency trade. +- Net peak memory: saves the intermediate serialized string buffer (~100MB on large states), but the parsed JS object remains because it already exists. Expect roughly 30-40% peak reduction, not 50%+. - **Verification gate:** `snapshotData` is only used for byte-count accounting and gunzip-then-parse round-tripping (verified — no hash or content comparison anywhere in `src/`). So byte-for-byte stability is NOT required; round-trip correctness is. Add a property-based test: random state → stream-stringify-gzip → gunzip-parse → deep-equal original. ### 3b. Worker-thread offload (only if replay moves too) @@ -229,16 +243,16 @@ If 0d alone is sufficient in prod (no OOMs, no event-loop-blocking signals), def ### 4b. Invalidation — full surface -All `tokenVersion: { increment: 1 }` write sites plus account deletion must invalidate: +All `tokenVersion: { increment: 1 }` write sites plus account deletion must invalidate. Already wired in code with `// AUTH_CACHE_INVALIDATION:` comments adjacent to each write — kept here for future-PR awareness: -- `auth.ts:77` (`revokeAllTokens`) -- `auth.ts:95` (`replaceToken`) -- `passkey.ts:611` (passkey recovery) -- `api.ts:209` (`prisma.user.delete`) — without this, a deleted user's stale cache entry serves "valid" until TTL expires (ghost token window). +- `auth.ts:77`/`:80` (`revokeAllTokens`) +- `auth.ts:96`/`:102` (`replaceToken`) +- `auth.ts:108` (post-write second invalidate after the new token version is read back) +- `passkey.ts:592`/`:616` (passkey recovery — pre- and post-write) +- `passkey.ts:278` (unverified-user delete in registration flow) +- `api.ts:210`/`:213`/`:215` (`prisma.user.delete` in account deletion — both pre- and post-delete invalidate) -Each call site needs an `authCache.invalidate(userId)` adjacent to the write. Add a code-search-friendly comment near each so future `tokenVersion` writes don't miss it. - -`isVerified` currently has no flip-to-zero path (`passkey.ts:276` deletes unverified users rather than flipping the flag). Add a comment in `auth-cache.ts` noting this assumption — if a future code path adds verification revocation, the cache will serve stale "valid" for up to TTL. +`isVerified` currently has no flip-to-zero path (`passkey.ts:277` deletes unverified users rather than flipping the flag). The assumption is already documented at `auth-cache.ts:80` — if a future code path adds verification revocation, the cache will serve stale "valid" for up to TTL. ### 4c. Multi-instance concerns @@ -255,7 +269,9 @@ Each call site needs an `authCache.invalidate(userId)` adjacent to the write. Ad - **Merge order:** 0a, 0b, 0c, 0d can land in any order, in parallel with Phase 1 design. Phase 2 depends on Phase 1 (same code paths). Phase 3 is conditional. Phase 4 is independent. - **Telemetry first:** before Phase 1 lands, add structured logging of `(opsInBatch, txDurationMs, dbRoundtrips)` to `uploadOps` so we can quantify the win. Existing audit log handles per-op decisions; add a single batch-summary line. -- **ADR:** add a `ARCHITECTURE-DECISIONS.md` entry per §1b ("Batch uploads under RepeatableRead — safety derives from `user_sync_state.lastSeq` row-lock, not from RR snapshot isolation"). +- **Backfill-flag DB self-check:** the env-only `SUPERSYNC_PAYLOAD_BYTES_BACKFILL_COMPLETE=true` flag is operator-trusted. To prevent a too-early flip, the server runs a cheap `EXISTS (SELECT 1 FROM operations WHERE payload_bytes = 0 LIMIT 1)` probe at startup whenever `batchUpload === true` and refuses to boot if any unbackfilled rows remain. +- **Reconcile guard during backfill window:** `calculateStorageUsage` returns a `hasUnbackfilledRows` flag (computed from a `BOOL_OR(payload_bytes = 0)` over the same single scan). `updateStorageUsage` skips the `users.storage_used_bytes` write when the flag is true, so an approximate SUM-with-`octet_length`-fallback never replaces the exact incrementally-maintained counter mid-backfill. The forced-reconcile marker is preserved across the skip so the next call (after backfill completes) reconciles correctly. +- **ADR:** see `ARCHITECTURE-DECISIONS.md` Decision #4 ("Batch Uploads Under RepeatableRead"); already merged. - **Docs:** update `docs/sync-and-op-log/operation-log-architecture-diagrams.md` §upload-path if it diagrams the per-op loop. - **Prisma migrate dev:** document the shadow-DB workaround for migrations containing `CREATE INDEX CONCURRENTLY`; `migrate deploy` can run the production workaround, but `migrate dev` wraps migration SQL in a transaction where `CONCURRENTLY` is forbidden. - **Server seq precision:** any raw `last_seq` read that crosses the JavaScript boundary must hard-fail if it is not a safe integer instead of blindly calling `Number(...)`. @@ -268,7 +284,7 @@ Each call site needs an `authCache.invalidate(userId)` adjacent to the write. Ad | Phase | Hot path affected | Expected win | Risk | | ----- | ----------------------------- | ---------------------------------------------------------------------------------------------------------------------- | -------------------------- | -| 0a | Snapshot fast-path validation | Eliminates seq-range scan on count | very low | +| 0a | Snapshot fast-path validation | Eliminates seq-range scan on **encrypted-op count**; full-state `findFirst` rechecks unchanged | very low | | 0b | Snapshot replay | ~100× fewer full stringifications (modulo full-state ops) | low | | 0c | Quota reconcile | Skips blob load (tens of MB) | very low | | 0d | All routes (memory headroom) | Stops OOMs near snapshot cap | low (ops change) | diff --git a/packages/super-sync-server/src/server.ts b/packages/super-sync-server/src/server.ts index c9f27c387a..6bf6ea406d 100644 --- a/packages/super-sync-server/src/server.ts +++ b/packages/super-sync-server/src/server.ts @@ -11,7 +11,13 @@ import { prisma, disconnectDb } from './db'; import websocket from '@fastify/websocket'; import { apiRoutes } from './api'; import { pageRoutes } from './pages'; -import { syncRoutes, startCleanupJobs, stopCleanupJobs, initSyncService } from './sync'; +import { + syncRoutes, + startCleanupJobs, + stopCleanupJobs, + initSyncService, + getSyncService, +} from './sync'; import { wsRoutes } from './sync/websocket.routes'; import { getWsConnectionService, @@ -221,6 +227,22 @@ export const createServer = ( options: { maxPayload: 1024 }, }); + // Backfill self-check: paired with the env-flag enforcement in + // loadConfigFromEnv. The env flag (SUPERSYNC_PAYLOAD_BYTES_BACKFILL_COMPLETE) + // is operator-set; if it is flipped to true before the migrate-payload-bytes + // script finishes, batch-upload deltas are still correct but the SUM-based + // reconcile in calculateStorageUsage would mix exact bytes with the + // CASE-WHEN fallback for legacy rows. One indexed probe at startup closes + // the trust hole: refuse to boot if any row still has payload_bytes = 0. + if (fullConfig.batchUpload) { + try { + await getSyncService().assertPayloadBytesBackfillComplete(); + } catch (err) { + Logger.error('Startup self-check failed', err); + throw err; + } + } + // Health Check - verifies database connectivity // Exempt from rate limiting (Kubernetes probes hit this every 5-15s) fastifyServer.get('/health', { config: { rateLimit: false } }, async (_, reply) => { 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 7f55e659dd..417d31819e 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 @@ -9,6 +9,7 @@ */ import { AsyncLocalStorage } from 'node:async_hooks'; import { prisma } from '../../db'; +import { Logger } from '../../logger'; /** * Default storage quota per user in bytes (100MB). @@ -86,20 +87,45 @@ export class StorageQuotaService { * * Rows with payload_bytes=0 are pre-backfill rows. They must not be counted * as zero bytes: that would let a reconcile lower the cached counter below - * actual usage. The fallback only touches unbackfilled rows, so once the - * one-time backfill completes this remains a cheap SUM(payload_bytes). + * actual usage. The CASE WHEN fallback only touches unbackfilled rows, so + * once the one-time backfill completes this remains a cheap SUM. + * + * `hasUnbackfilledRows` is computed in the same single scan via BOOL_OR. + * Callers (notably `updateStorageUsage`) treat the SUM as approximate when + * this flag is true, because the fallback's UTF-8 length differs by single + * bytes from the JS-side `computeOpStorageBytes` value used by the hot-path + * counter. Skipping the `users.storage_used_bytes` write while unbackfilled + * rows exist preserves the exact incremental counter. */ async calculateStorageUsage(userId: number): Promise<{ operationsBytes: number; snapshotBytes: number; totalBytes: number; + hasUnbackfilledRows: boolean; }> { const usageResult = await prisma.$queryRaw< - [{ operations_bytes: bigint | null; snapshot_bytes: number | bigint | null }] + [ + { + operations_bytes: bigint | null; + snapshot_bytes: number | bigint | null; + has_unbackfilled?: boolean | null; + }, + ] >` SELECT - ( - SELECT COALESCE( + ops.operations_bytes, + ops.has_unbackfilled, + COALESCE( + ( + SELECT octet_length(snapshot_data) + FROM user_sync_state + WHERE user_id = ${userId} + ), + 0 + )::bigint AS snapshot_bytes + FROM ( + SELECT + COALESCE( SUM( CASE WHEN payload_bytes > 0 THEN payload_bytes @@ -108,25 +134,23 @@ export class StorageQuotaService { END ), 0 - ) - FROM operations - WHERE user_id = ${userId} - ) as operations_bytes, - ( - SELECT COALESCE(octet_length(snapshot_data), 0) - FROM user_sync_state - WHERE user_id = ${userId} - ) as snapshot_bytes + )::bigint AS operations_bytes, + COALESCE(BOOL_OR(payload_bytes = 0), false) AS has_unbackfilled + FROM operations + WHERE user_id = ${userId} + ) AS ops `; const operationsBytes = Number(usageResult[0]?.operations_bytes ?? 0); const snapshotBytes = Number(usageResult[0]?.snapshot_bytes ?? 0); const totalBytes = operationsBytes + snapshotBytes; + const hasUnbackfilledRows = Boolean(usageResult[0]?.has_unbackfilled ?? false); return { operationsBytes, snapshotBytes, totalBytes, + hasUnbackfilledRows, }; } @@ -228,7 +252,19 @@ export class StorageQuotaService { // that is itself queued behind our own lock → deadlock. const inLock = this.storageUsageLockContext.getStore()?.has(userId); if (inLock) { - const { totalBytes } = await this.calculateStorageUsage(userId); + const { totalBytes, hasUnbackfilledRows } = + await this.calculateStorageUsage(userId); + if (hasUnbackfilledRows) { + // Pre-backfill rows make the SUM approximate (CASE-WHEN fallback uses + // postgres-side text length, not JS-side computeOpStorageBytes). Writing + // an approximate value here would replace the exact incrementally + // maintained counter — drift in either direction. Leave the forced + // reconcile marker so a post-backfill call self-heals. + Logger.warn( + `[user:${userId}] Skipping storage usage reconcile: payload_bytes backfill incomplete for this user.`, + ); + return; + } await prisma.user.update({ where: { id: userId }, data: { storageUsedBytes: BigInt(totalBytes) }, @@ -241,7 +277,14 @@ export class StorageQuotaService { if (existing) return existing; const promise = this.runWithStorageUsageLock(userId, async () => { - const { totalBytes } = await this.calculateStorageUsage(userId); + const { totalBytes, hasUnbackfilledRows } = + await this.calculateStorageUsage(userId); + if (hasUnbackfilledRows) { + Logger.warn( + `[user:${userId}] Skipping storage usage reconcile: payload_bytes backfill incomplete for this user.`, + ); + return; + } await prisma.user.update({ where: { id: userId }, data: { storageUsedBytes: BigInt(totalBytes) }, @@ -294,6 +337,37 @@ export class StorageQuotaService { this.forcedReconciles.delete(userId); } + /** + * Backfill self-check. When SUPERSYNC_BATCH_UPLOAD=true the operator is + * trusted to have set SUPERSYNC_PAYLOAD_BYTES_BACKFILL_COMPLETE=true only + * after `npm run migrate-payload-bytes` finished. If the flag was flipped + * too early, batch uploads still write `payload_bytes` correctly but the + * SUM-based reconcile in `calculateStorageUsage` would mix exact bytes with + * the CASE-WHEN fallback for legacy rows — small drift, but unnecessary + * given the env-flag's whole purpose. + * + * One indexed-probe at startup closes the trust hole. The query relies on a + * full table scan-with-LIMIT-1; for a fully backfilled table that is one + * row visit on the first encountered row (cheap), and for a partially + * backfilled table it returns immediately. Worst case (zero rows in + * `operations`, e.g. fresh deployment) is also one round-trip. + */ + async assertPayloadBytesBackfillComplete(): Promise { + const result = await prisma.$queryRaw<[{ exists: boolean }]>` + SELECT EXISTS ( + SELECT 1 FROM operations WHERE payload_bytes = 0 LIMIT 1 + ) AS "exists" + `; + if (result[0]?.exists) { + throw new Error( + 'SUPERSYNC_BATCH_UPLOAD is enabled but the operations table still ' + + 'contains rows with payload_bytes = 0. Run ' + + '`npm run migrate-payload-bytes` to complete the backfill before ' + + 'setting SUPERSYNC_PAYLOAD_BYTES_BACKFILL_COMPLETE=true.', + ); + } + } + /** * Get storage quota and usage for a user. * Used by status endpoint. diff --git a/packages/super-sync-server/src/sync/sync.service.ts b/packages/super-sync-server/src/sync/sync.service.ts index f59e4d2244..3f0d3d828e 100644 --- a/packages/super-sync-server/src/sync/sync.service.ts +++ b/packages/super-sync-server/src/sync/sync.service.ts @@ -1697,10 +1697,15 @@ export class SyncService { operationsBytes: number; snapshotBytes: number; totalBytes: number; + hasUnbackfilledRows: boolean; }> { return this.storageQuotaService.calculateStorageUsage(userId); } + async assertPayloadBytesBackfillComplete(): Promise { + return this.storageQuotaService.assertPayloadBytesBackfillComplete(); + } + async checkStorageQuota( userId: number, additionalBytes: number, 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 d8b7255882..cc6d2e1a52 100644 --- a/packages/super-sync-server/tests/storage-quota-cleanup.spec.ts +++ b/packages/super-sync-server/tests/storage-quota-cleanup.spec.ts @@ -198,7 +198,7 @@ vi.mock('../src/db', () => { .mockImplementation(async (query, userIdArg, deleteUpToSeqArg) => { // The same mock serves two SQL shapes: // 1. calculateStorageUsage's SUM(payload_bytes) reconcile - // — returns `[{ operations_bytes, snapshot_bytes }]`. + // — returns `[{ operations_bytes, snapshot_bytes, has_unbackfilled }]`. // 2. deleteOldestRestorePointAndOps's bounded full-state sum // filtered by `op_type IN (SYNC_IMPORT, BACKUP_IMPORT, REPAIR)` // — returns `[{ exact_bytes, full_state_count }]`. @@ -232,7 +232,13 @@ vi.mock('../src/db', () => { if (isFullStateScan) { return [{ exact_bytes: totalBytes, full_state_count: fullStateCount }]; } - return [{ operations_bytes: totalBytes, snapshot_bytes: BigInt(0) }]; + return [ + { + operations_bytes: totalBytes, + snapshot_bytes: BigInt(0), + has_unbackfilled: false, + }, + ]; }), // decrementStorageUsage uses $executeRaw with a clamped UPDATE. // Simulate by mutating the matching user's storageUsedBytes in-place. 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 aa1aa423d7..20a0da9b39 100644 --- a/packages/super-sync-server/tests/storage-quota.service.spec.ts +++ b/packages/super-sync-server/tests/storage-quota.service.spec.ts @@ -34,7 +34,11 @@ describe('StorageQuotaService', () => { describe('calculateStorageUsage', () => { it('should calculate storage from operations and snapshot', async () => { vi.mocked(prisma.$queryRaw).mockResolvedValue([ - { operations_bytes: BigInt(5000), snapshot_bytes: BigInt(3000) }, + { + operations_bytes: BigInt(5000), + snapshot_bytes: BigInt(3000), + has_unbackfilled: false, + }, ]); const result = await service.calculateStorageUsage(1); @@ -43,12 +47,13 @@ describe('StorageQuotaService', () => { operationsBytes: 5000, snapshotBytes: 3000, totalBytes: 8000, + hasUnbackfilledRows: false, }); }); it('should handle null operation total', async () => { vi.mocked(prisma.$queryRaw).mockResolvedValue([ - { operations_bytes: null, snapshot_bytes: null }, + { operations_bytes: null, snapshot_bytes: null, has_unbackfilled: false }, ]); const result = await service.calculateStorageUsage(1); @@ -57,12 +62,17 @@ describe('StorageQuotaService', () => { operationsBytes: 0, snapshotBytes: 0, totalBytes: 0, + hasUnbackfilledRows: false, }); }); it('should handle missing snapshot', async () => { vi.mocked(prisma.$queryRaw).mockResolvedValue([ - { operations_bytes: BigInt(1000), snapshot_bytes: BigInt(0) }, + { + operations_bytes: BigInt(1000), + snapshot_bytes: BigInt(0), + has_unbackfilled: false, + }, ]); const result = await service.calculateStorageUsage(1); @@ -71,12 +81,17 @@ describe('StorageQuotaService', () => { operationsBytes: 1000, snapshotBytes: 0, totalBytes: 1000, + hasUnbackfilledRows: false, }); }); it('should include fallback bytes for rows that are still unbackfilled', async () => { vi.mocked(prisma.$queryRaw).mockResolvedValue([ - { operations_bytes: BigInt(750), snapshot_bytes: BigInt(100) }, + { + operations_bytes: BigInt(750), + snapshot_bytes: BigInt(100), + has_unbackfilled: true, + }, ]); const result = await service.calculateStorageUsage(1); @@ -85,12 +100,17 @@ describe('StorageQuotaService', () => { operationsBytes: 750, snapshotBytes: 100, totalBytes: 850, + hasUnbackfilledRows: true, }); }); it('should use persisted byte counters with a safe unbackfilled fallback', async () => { vi.mocked(prisma.$queryRaw).mockResolvedValue([ - { operations_bytes: BigInt(1000), snapshot_bytes: BigInt(0) }, + { + operations_bytes: BigInt(1000), + snapshot_bytes: BigInt(0), + has_unbackfilled: false, + }, ]); await service.calculateStorageUsage(1); @@ -104,12 +124,27 @@ describe('StorageQuotaService', () => { expect(query).toContain('SUM('); expect(query).toContain('payload_bytes'); expect(query).toContain('WHEN payload_bytes > 0'); + expect(query).toContain('BOOL_OR(payload_bytes = 0)'); expect(query).toContain('octet_length(snapshot_data)'); expect(query).not.toContain('snapshot_data: true'); expect(query).not.toContain('pg_column_size'); }); }); + describe('assertPayloadBytesBackfillComplete', () => { + it('should resolve when no unbackfilled rows exist', async () => { + vi.mocked(prisma.$queryRaw).mockResolvedValue([{ exists: false }]); + await expect(service.assertPayloadBytesBackfillComplete()).resolves.toBeUndefined(); + }); + + it('should throw when any row has payload_bytes = 0', async () => { + vi.mocked(prisma.$queryRaw).mockResolvedValue([{ exists: true }]); + await expect(service.assertPayloadBytesBackfillComplete()).rejects.toThrow( + /SUPERSYNC_BATCH_UPLOAD is enabled but the operations table still contains rows with payload_bytes = 0/, + ); + }); + }); + describe('checkStorageQuota', () => { it('should allow upload when under quota', async () => { vi.mocked(prisma.user.findUnique).mockResolvedValue({ @@ -169,7 +204,11 @@ describe('StorageQuotaService', () => { describe('updateStorageUsage', () => { it('should update storage usage from calculated total', async () => { vi.mocked(prisma.$queryRaw).mockResolvedValue([ - { operations_bytes: BigInt(75000), snapshot_bytes: BigInt(25000) }, + { + operations_bytes: BigInt(75000), + snapshot_bytes: BigInt(25000), + has_unbackfilled: false, + }, ]); vi.mocked(prisma.user.update).mockResolvedValue({} as any); @@ -181,11 +220,36 @@ describe('StorageQuotaService', () => { }); }); + it('should skip the storage-counter write while unbackfilled rows remain', async () => { + vi.mocked(prisma.$queryRaw).mockResolvedValue([ + { + operations_bytes: BigInt(50000), + snapshot_bytes: BigInt(0), + has_unbackfilled: true, + }, + ]); + vi.mocked(prisma.user.update).mockResolvedValue({} as any); + + // Pre-set the forced-reconcile marker; the skip path must preserve it + // so a post-backfill call self-heals. + service.markNeedsReconcile(1); + await service.updateStorageUsage(1); + + expect(prisma.user.update).not.toHaveBeenCalled(); + expect(service.needsReconcile(1)).toBe(true); + }); + it('should dedupe concurrent reconciles for the same user', async () => { // Simulate a slow exact usage scan so concurrent callers // overlap on the in-flight promise. let releaseScan: ( - value: [{ operations_bytes: bigint; snapshot_bytes: bigint }], + value: [ + { + operations_bytes: bigint; + snapshot_bytes: bigint; + has_unbackfilled: boolean; + }, + ], ) => void = () => undefined; vi.mocked(prisma.$queryRaw).mockReturnValueOnce( new Promise((resolve) => { @@ -198,7 +262,13 @@ describe('StorageQuotaService', () => { const second = service.updateStorageUsage(1); const third = service.updateStorageUsage(1); - releaseScan([{ operations_bytes: BigInt(123), snapshot_bytes: BigInt(0) }]); + releaseScan([ + { + operations_bytes: BigInt(123), + snapshot_bytes: BigInt(0), + has_unbackfilled: false, + }, + ]); await Promise.all([first, second, third]); // Only one scan + one write should have run for three concurrent calls. @@ -208,7 +278,11 @@ describe('StorageQuotaService', () => { it('should re-scan on a subsequent sequential call after the lock clears', async () => { vi.mocked(prisma.$queryRaw).mockResolvedValue([ - { operations_bytes: BigInt(10), snapshot_bytes: BigInt(0) }, + { + operations_bytes: BigInt(10), + snapshot_bytes: BigInt(0), + has_unbackfilled: false, + }, ]); vi.mocked(prisma.user.update).mockResolvedValue({} as any); @@ -228,7 +302,11 @@ describe('StorageQuotaService', () => { // Lock must be cleared so the next call retries the scan rather than // returning the rejected promise forever. vi.mocked(prisma.$queryRaw).mockResolvedValueOnce([ - { operations_bytes: BigInt(0), snapshot_bytes: BigInt(0) }, + { + operations_bytes: BigInt(0), + snapshot_bytes: BigInt(0), + has_unbackfilled: false, + }, ]); await expect(service.updateStorageUsage(1)).resolves.toBeUndefined(); expect(prisma.$queryRaw).toHaveBeenCalledTimes(2); @@ -256,7 +334,11 @@ describe('StorageQuotaService', () => { inflightMap.set(1, neverResolves); vi.mocked(prisma.$queryRaw).mockResolvedValue([ - { operations_bytes: BigInt(42), snapshot_bytes: BigInt(0) }, + { + operations_bytes: BigInt(42), + snapshot_bytes: BigInt(0), + has_unbackfilled: false, + }, ] as any); vi.mocked(prisma.user.update).mockResolvedValue({} as any); @@ -296,7 +378,13 @@ describe('StorageQuotaService', () => { vi.mocked(prisma.$queryRaw).mockImplementation(async () => { events.push('scan'); - return [{ operations_bytes: BigInt(123), snapshot_bytes: BigInt(0) }]; + return [ + { + operations_bytes: BigInt(123), + snapshot_bytes: BigInt(0), + has_unbackfilled: false, + }, + ]; }); vi.mocked(prisma.user.update).mockImplementation(async () => { events.push('write');